service: Add driver-specific create_session(s) APIs for MI drivers

ni_measurementlink_service/_drivers/_nidcpower.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+import nidcpower
+
+from ni_measurementlink_service import session_management  # circular import
+from ni_measurementlink_service._channelpool import GrpcChannelPool
+from ni_measurementlink_service._configuration import NIDCPOWER_OPTIONS
+from ni_measurementlink_service._drivers._grpcdevice import (
+    get_insecure_grpc_device_channel,
+)
+from ni_measurementlink_service._internal.discovery_client import DiscoveryClient
+from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior
+
+_INITIALIZATION_BEHAVIOR = {
+    SessionInitializationBehavior.AUTO: nidcpower.SessionInitializationBehavior.AUTO,
+    SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: nidcpower.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION,
+    SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: nidcpower.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION,
+}
+
+
+class SessionConstructor:
+    """Constructs sessions based on SessionInformation."""
+
+    def __init__(
+        self,
+        discovery_client: DiscoveryClient,
+        grpc_channel_pool: GrpcChannelPool,
+        reset: bool,
+        options: Optional[Dict[str, Any]],
+        initialization_behavior: SessionInitializationBehavior,
+    ) -> None:
+        """Initialize the session constructor."""
+        self._grpc_channel = get_insecure_grpc_device_channel(
+            discovery_client, grpc_channel_pool, nidcpower.GRPC_SERVICE_INTERFACE_NAME
+        )
+        self._reset = reset
+        self._options = NIDCPOWER_OPTIONS.to_dict() if options is None else options
+        self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior]
+
+    def __call__(self, session_info: session_management.SessionInformation) -> nidcpower.Session:
+        """Construct a session object."""
+        kwargs: Dict[str, Any] = {}
+        if self._grpc_channel:
+            kwargs["grpc_options"] = nidcpower.GrpcSessionOptions(
+                grpc_channel=self._grpc_channel,
+                session_name=session_info.session_name,
+                initialization_behavior=self._initialization_behavior,
+            )
+
+        return nidcpower.Session(
+            resource_name=session_info.resource_name,
+            reset=self._reset,
+            options=self._options,
+            **kwargs,
+        )

ni_measurementlink_service/_drivers/_nifgen.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+import nifgen
+
+from ni_measurementlink_service import session_management  # circular import
+from ni_measurementlink_service._channelpool import GrpcChannelPool
+from ni_measurementlink_service._configuration import NIFGEN_OPTIONS
+from ni_measurementlink_service._drivers._grpcdevice import (
+    get_insecure_grpc_device_channel,
+)
+from ni_measurementlink_service._internal.discovery_client import DiscoveryClient
+from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior
+
+_INITIALIZATION_BEHAVIOR = {
+    SessionInitializationBehavior.AUTO: nifgen.SessionInitializationBehavior.AUTO,
+    SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: nifgen.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION,
+    SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: nifgen.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION,
+}
+
+
+class SessionConstructor:
+    """Constructs sessions based on SessionInformation."""
+
+    def __init__(
+        self,
+        discovery_client: DiscoveryClient,
+        grpc_channel_pool: GrpcChannelPool,
+        reset_device: bool,
+        options: Optional[Dict[str, Any]],
+        initialization_behavior: SessionInitializationBehavior,
+    ) -> None:
+        """Initialize the session constructor."""
+        self._grpc_channel = get_insecure_grpc_device_channel(
+            discovery_client, grpc_channel_pool, nifgen.GRPC_SERVICE_INTERFACE_NAME
+        )
+        self._reset_device = reset_device
+        self._options = NIFGEN_OPTIONS.to_dict() if options is None else options
+        self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior]
+
+    def __call__(self, session_info: session_management.SessionInformation) -> nifgen.Session:
+        """Construct a session object."""
+        kwargs: Dict[str, Any] = {}
+        if self._grpc_channel:
+            kwargs["grpc_options"] = nifgen.GrpcSessionOptions(
+                grpc_channel=self._grpc_channel,
+                session_name=session_info.session_name,
+                initialization_behavior=self._initialization_behavior,
+            )
+
+        return nifgen.Session(
+            resource_name=session_info.resource_name,
+            reset_device=self._reset_device,
+            options=self._options,
+            **kwargs,
+        )

ni_measurementlink_service/_drivers/_niswitch.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+import niswitch
+
+from ni_measurementlink_service import session_management  # circular import
+from ni_measurementlink_service._channelpool import GrpcChannelPool
+from ni_measurementlink_service._configuration import NISWITCH_OPTIONS
+from ni_measurementlink_service._drivers._grpcdevice import (
+    get_insecure_grpc_device_channel,
+)
+from ni_measurementlink_service._internal.discovery_client import DiscoveryClient
+from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior
+
+_INITIALIZATION_BEHAVIOR = {
+    SessionInitializationBehavior.AUTO: niswitch.SessionInitializationBehavior.AUTO,
+    SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: niswitch.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION,
+    SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: niswitch.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION,
+}
+
+
+class SessionConstructor:
+    """Constructs sessions based on SessionInformation."""
+
+    def __init__(
+        self,
+        discovery_client: DiscoveryClient,
+        grpc_channel_pool: GrpcChannelPool,
+        topology: Optional[str],
+        simulate: Optional[bool],
+        reset_device: bool,
+        initialization_behavior: SessionInitializationBehavior,
+    ) -> None:
+        """Initialize the session constructor."""
+        self._grpc_channel = get_insecure_grpc_device_channel(
+            discovery_client, grpc_channel_pool, niswitch.GRPC_SERVICE_INTERFACE_NAME
+        )
+        self._topology = NISWITCH_OPTIONS.topology if topology is None else topology
+        self._simulate = NISWITCH_OPTIONS.simulate if simulate is None else simulate
+        self._reset_device = reset_device
+        self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior]
+
+    def __call__(self, session_info: session_management.SessionInformation) -> niswitch.Session:
+        """Construct a session object."""
+        kwargs: Dict[str, Any] = {}
+        if self._grpc_channel:
+            kwargs["grpc_options"] = niswitch.GrpcSessionOptions(
+                grpc_channel=self._grpc_channel,
+                session_name=session_info.session_name,
+                initialization_behavior=self._initialization_behavior,
+            )
+
+        return niswitch.Session(
+            resource_name=session_info.resource_name,
+            topology=self._topology,
+            simulate=self._simulate,
+            reset_device=self._reset_device,
+            **kwargs,
+        )

ni_measurementlink_service/_sessiontypes.py

@@ -0,0 +1,43 @@
+"""Session management data types.
+
+This submodule is intended to prevent circular imports between
+session_management.py and driver-specific code such as
+_drivers/_nidcpower.py.
+"""
+from enum import IntEnum
+
+
+class SessionInitializationBehavior(IntEnum):
+    """Specifies whether to initialize a new session or attach to an existing session."""
+
+    AUTO = 0
+    """
+    The NI gRPC Device Server will attach to an existing session with the
+    specified name if it exists, otherwise the server will initialize a new
+    session.
+
+    Note: When using the Session as a context manager and the context exits, the
+    behavior depends on what happened when the constructor was called. If it
+    resulted in a new session being initialized on the NI gRPC Device Server,
+    then it will automatically close the server session. If it instead attached
+    to an existing session, then it will detach from the server session and
+    leave it open.
+    """
+
+    INITIALIZE_SERVER_SESSION = 1
+    """
+    Require the NI gRPC Device Server to initialize a new session with the
+    specified name.
+
+    Note: When using the Session as a context manager and the context exits, it
+    will automatically close the server session.
+    """
+
+    ATTACH_TO_SERVER_SESSION = 2
+    """
+    Require the NI gRPC Device Server to attach to an existing session with the
+    specified name.
+
+    Note: When using the Session as a context manager and the context exits, it
+    will detach from the server session and leave it open.
+    """