Implement connect() and disconnect()

CONTRIBUTING.md

@@ -22,7 +22,7 @@ See [GitHub's official documentation](https://help.github.com/articles/using-pul
 # Getting Started
 
 This is the command to generate the files in /src/ni/pythonpanel/v1/:
-`poetry run python -m grpc_tools.protoc --proto_path=protos --python_out=src/ --grpc_python_out=src/ --plugin=protoc-gen-mypy=.venv\Scripts\protoc-gen-mypy.exe --mypy_out=src/ --mypy_grpc_out=src/ ni/pythonpanel/v1/python_panel_service.proto`
+`poetry run python -m grpc_tools.protoc --proto_path=protos --python_out=src/ --grpc_python_out=src/ --mypy_out=src/ --mypy_grpc_out=src/ ni/pythonpanel/v1/python_panel_service.proto`
 
 # Testing
 

pyproject.toml

@@ -15,6 +15,7 @@ ni-measurement-plugin-sdk = {version=">=2.3"}
 [tool.poetry.group.dev.dependencies]
 grpc-stubs = "^1.53"
 types-protobuf = ">=4.21"
+pyupgrade = "^3.19.1"
 
 [tool.poetry.group.lint.dependencies]
 bandit = { version = ">=1.7", extras = ["toml"] }
@@ -41,6 +42,9 @@ build-backend = "poetry.core.masonry.api"
 [tool.ni-python-styleguide]
 extend_exclude = ".tox,docs,src/ni/pythonpanel/v1"
 
+[tool.black]
+extend-exclude = '\.tox/|docs/|src/ni/pythonpanel/v1/'
+
 [tool.mypy]
 files = "examples/,src/nipanel/,tests/"
 namespace_packages = true
@@ -60,9 +64,4 @@ skips = [
 
 [tool.pytest.ini_options]
 addopts = "--doctest-modules --strict-markers"
-testpaths = ["src/nipanel", "tests"]
-
-[tool.black]
-extend-exclude = '''
-/src/ni/pythonpanel/v1/
-'''
+testpaths = ["src/nipanel", "tests"]

src/nipanel/_panel.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
 
 import sys
-from abc import ABC, abstractmethod
+from abc import ABC
 from types import TracebackType
-from typing import TYPE_CHECKING, Optional, Type
+from typing import TYPE_CHECKING
 
-from ni_measurement_plugin_sdk_service.discovery import DiscoveryClient
+import grpc
 
 from nipanel._panel_client import PanelClient
 
@@ -25,9 +25,21 @@ class Panel(ABC):
 
     __slots__ = ["_panel_client", "_panel_id", "_panel_uri", "__weakref__"]
 
-    def __init__(self, panel_id: str, panel_uri: str) -> None:
+    def __init__(
+        self,
+        *,
+        panel_id: str,
+        panel_uri: str,
+        provided_interface: str,
+        service_class: str,
+        grpc_channel: grpc.Channel | None = None,
+    ) -> None:
         """Initialize the panel."""
-        self._panel_client = PanelClient(self._resolve_service_address)
+        self._panel_client = PanelClient(
+            provided_interface=provided_interface,
+            service_class=service_class,
+            grpc_channel=grpc_channel,
+        )
         self._panel_id = panel_id
         self._panel_uri = panel_uri
 
@@ -48,10 +60,10 @@ def __enter__(self) -> Self:
 
     def __exit__(
         self,
-        exctype: Optional[Type[BaseException]],
-        excinst: Optional[BaseException],
-        exctb: Optional[TracebackType],
-    ) -> Optional[bool]:
+        exctype: type[BaseException] | None,
+        excinst: BaseException | None,
+        exctb: TracebackType | None,
+    ) -> bool | None:
         """Exit the runtime context related to this object."""
         self.disconnect()
         return None
@@ -85,8 +97,3 @@ def set_value(self, value_id: str, value: object) -> None:
         """
         # TODO: AB#3095681 - Convert the value to an Any and pass it to _client.set_value
         pass
-
-    @abstractmethod
-    def _resolve_service_address(self, discovery_client: DiscoveryClient) -> str:
-        """Resolve the service address for the panel."""
-        raise NotImplementedError

src/nipanel/_panel_client.py

@@ -4,13 +4,14 @@
 
 import logging
 import threading
-from typing import Any, Callable
+from typing import Callable
 
 import grpc
 from ni.pythonpanel.v1.python_panel_service_pb2 import ConnectRequest, DisconnectRequest
 from ni.pythonpanel.v1.python_panel_service_pb2_grpc import PythonPanelServiceStub
 from ni_measurement_plugin_sdk_service.discovery import DiscoveryClient
 from ni_measurement_plugin_sdk_service.grpc.channelpool import GrpcChannelPool
+from typing_extensions import ParamSpec, TypeVar
 
 _logger = logging.getLogger(__name__)
 
@@ -20,47 +21,69 @@ class PanelClient:
 
     def __init__(
         self,
-        resolve_service_address_fn: Callable[[DiscoveryClient], str],
         *,
+        provided_interface: str,
+        service_class: str,
         discovery_client: DiscoveryClient | None = None,
         grpc_channel: grpc.Channel | None = None,
         grpc_channel_pool: GrpcChannelPool | None = None,
     ) -> None:
         """Initialize the panel client.
 
         Args:
-            resolve_service_address_fn: A function to resolve the service location.
+            provided_interface: The interface provided by the service.
+            service_class: The class of the service.
             discovery_client: An optional discovery client.
             grpc_channel: An optional panel gRPC channel.
             grpc_channel_pool: An optional gRPC channel pool.
         """
         self._initialization_lock = threading.Lock()
-        self._resolve_service_address_fn = resolve_service_address_fn
+        self._provided_interface = provided_interface
+        self._service_class = service_class
         self._discovery_client = discovery_client
         self._grpc_channel_pool = grpc_channel_pool
+        self._grpc_channel = grpc_channel
         self._stub: PythonPanelServiceStub | None = None
 
-        if grpc_channel is not None:
-            self._stub = PythonPanelServiceStub(grpc_channel)
+    def connect(self, panel_id: str, panel_uri: str) -> None:
+        """Connect to the panel and open it."""
+        connect_request = ConnectRequest(panel_id=panel_id, panel_uri=panel_uri)
+        self._invoke_with_retry(self._get_stub().Connect, connect_request)
+
+    def disconnect(self, panel_id: str) -> None:
+        """Disconnect from the panel (does not close the panel)."""
+        disconnect_request = DisconnectRequest(panel_id=panel_id)
+        self._invoke_with_retry(self._get_stub().Disconnect, disconnect_request)
 
     def _get_stub(self) -> PythonPanelServiceStub:
         if self._stub is None:
-            with self._initialization_lock:
-                if self._grpc_channel_pool is None:
-                    _logger.debug("Creating unshared GrpcChannelPool.")
-                    self._grpc_channel_pool = GrpcChannelPool()
-                if self._discovery_client is None:
-                    _logger.debug("Creating unshared DiscoveryClient.")
-                    self._discovery_client = DiscoveryClient(
-                        grpc_channel_pool=self._grpc_channel_pool
+            if self._grpc_channel is not None:
+                self._stub = PythonPanelServiceStub(self._grpc_channel)
+            else:
+                with self._initialization_lock:
+                    if self._grpc_channel_pool is None:
+                        _logger.debug("Creating unshared GrpcChannelPool.")
+                        self._grpc_channel_pool = GrpcChannelPool()
+                    if self._discovery_client is None:
+                        _logger.debug("Creating unshared DiscoveryClient.")
+                        self._discovery_client = DiscoveryClient(
+                            grpc_channel_pool=self._grpc_channel_pool
+                        )
+
+                    service_location = self._discovery_client.resolve_service(
+                        provided_interface=self._provided_interface,
+                        service_class=self._service_class,
                     )
-                if self._stub is None:
-                    service_address = self._resolve_service_address_fn(self._discovery_client)
-                    channel = self._grpc_channel_pool.get_channel(service_address)
+                    channel = self._grpc_channel_pool.get_channel(service_location.insecure_address)
                     self._stub = PythonPanelServiceStub(channel)
         return self._stub
 
-    def _invoke_with_retry(self, method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+    _T = TypeVar("_T")
+    _P = ParamSpec("_P")
+
+    def _invoke_with_retry(
+        self, method: Callable[_P, _T], *args: _P.args, **kwargs: _P.kwargs
+    ) -> _T:
         """Invoke a gRPC method with retry logic."""
         try:
             return method(*args, **kwargs)
@@ -69,13 +92,4 @@ def _invoke_with_retry(self, method: Callable[..., Any], *args: Any, **kwargs: A
                 # if the service is unavailable, we can retry the connection
                 self._stub = None
                 return method(*args, **kwargs)
-
-    def connect(self, panel_id: str, panel_uri: str) -> None:
-        """Connect to the panel and open it."""
-        connect_request = ConnectRequest(panel_id=panel_id, panel_uri=panel_uri)
-        self._invoke_with_retry(self._get_stub().Connect, connect_request)
-
-    def disconnect(self, panel_id: str) -> None:
-        """Disconnect from the panel (does not close the panel)."""
-        disconnect_request = DisconnectRequest(panel_id=panel_id)
-        self._invoke_with_retry(self._get_stub().Disconnect, disconnect_request)
+            raise

src/nipanel/_streamlit_panel.py

@@ -1,4 +1,6 @@
-from ni_measurement_plugin_sdk_service.discovery import DiscoveryClient
+from __future__ import annotations
+
+import grpc
 
 from nipanel._panel import Panel
 
@@ -10,20 +12,23 @@ class StreamlitPanel(Panel):
 
     __slots__ = ()
 
-    def __init__(self, panel_id: str, streamlit_script_uri: str) -> None:
+    def __init__(
+        self, panel_id: str, streamlit_script_uri: str, *, grpc_channel: grpc.Channel | None = None
+    ) -> None:
         """Create a panel using a Streamlit script for the user interface.
 
         Args:
             panel_id: A unique identifier for the panel.
             streamlit_script_uri: The file path of the Streamlit script.
+            grpc_channel: An optional gRPC channel to use for communication with the panel service.
 
         Returns:
             A new StreamlitPanel instance.
         """
-        super().__init__(panel_id, streamlit_script_uri)
-
-    def _resolve_service_address(self, discovery_client: DiscoveryClient) -> str:
-        service_location = discovery_client.resolve_service(
-            provided_interface=self.PYTHON_PANEL_SERVICE, service_class=self.PYTHON_PANEL_SERVICE
+        super().__init__(
+            panel_id=panel_id,
+            panel_uri=streamlit_script_uri,
+            provided_interface=self.PYTHON_PANEL_SERVICE,
+            service_class=self.PYTHON_PANEL_SERVICE,
+            grpc_channel=grpc_channel,
         )
-        return service_location.insecure_address

tests/conftest.py

@@ -1,37 +1,42 @@
 """Fixtures for testing."""
 
+from collections.abc import Generator
 from concurrent import futures
-from typing import Any, Generator
 
 import grpc
 import pytest
 from ni.pythonpanel.v1.python_panel_service_pb2_grpc import (
     PythonPanelServiceStub,
-    add_PythonPanelServiceServicer_to_server,
 )
 
-from tests.utils._fake_python_panel_servicer import FakePythonPanelServicer
+from tests.utils._fake_python_panel_service import FakePythonPanelService
 
 
 @pytest.fixture
-def fake_python_panel_service() -> Generator[tuple[FakePythonPanelServicer, int], Any, None]:
+def fake_python_panel_service() -> Generator[FakePythonPanelService]:
     """Fixture to create a FakePythonPanelServicer for testing."""
-    thread_pool = futures.ThreadPoolExecutor(max_workers=10)
-    server = grpc.server(thread_pool)
-    servicer = FakePythonPanelServicer()
-    add_PythonPanelServiceServicer_to_server(servicer, server)
-    port = server.add_insecure_port("[::]:0")
-    server.start()
-    yield servicer, port
-    server.stop(None)
+    with futures.ThreadPoolExecutor(max_workers=10) as thread_pool:
+        service = FakePythonPanelService()
+        service.start(thread_pool)
+        yield service
+        service.stop()
 
 
 @pytest.fixture
-def fake_python_panel_service_stub(
-    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
-) -> Generator[PythonPanelServiceStub, Any, None]:
-    """Fixture to attach a PythonPanelSericeStub to a FakePythonPanelService."""
-    _, port = fake_python_panel_service
-    channel = grpc.insecure_channel(f"localhost:{port}")
-    yield PythonPanelServiceStub(channel)
+def grpc_channel_and_fake_panel_service(
+    fake_python_panel_service: FakePythonPanelService,
+) -> Generator[tuple[grpc.Channel, FakePythonPanelService]]:
+    """Fixture to get a channel to the FakePythonPanelService."""
+    service = fake_python_panel_service
+    channel = grpc.insecure_channel(f"localhost:{service.port}")
+    yield channel, service
     channel.close()
+
+
+@pytest.fixture
+def python_panel_service_stub(
+    grpc_channel_and_fake_panel_service: tuple[grpc.Channel, FakePythonPanelService],
+) -> Generator[PythonPanelServiceStub]:
+    """Fixture to get a PythonPanelServiceStub, attached to a FakePythonPanelService."""
+    channel, _ = grpc_channel_and_fake_panel_service
+    yield PythonPanelServiceStub(channel)