mypy - FSM

pyproject.toml

@@ -1,57 +1,56 @@
 [build-system]
-requires = ["setuptools >= 61.0"]
 build-backend = "setuptools.build_meta"
+requires = ["setuptools >= 61.0"]
 
 [project]
-name = "drunc"
 description = "A flexible run control infrastructure for a distributed DAQ system"
-version = "1.1.4"
+name = "drunc"
 readme = "docs/README.md"
 requires-python = ">=3.10"
+version = "1.1.4"
 
 dependencies = [
-    "click",
-    "click-shell",
-    "protobuf",
-    "types-protobuf",
-    "grpcio",
-    "grpcio-status",
-    "grpcio-tools",
-    "types-grpcio",
-    "gunicorn",
-    "kafka-python",
-    "nest-asyncio",
-    "rich",
-    "requests",
-    "Flask",
-    "Flask-RESTful",
-    "sh",
-    "kubernetes",
-    "pytz",
-    "psutil",
-    "paramiko"
+  "click",
+  "click-shell",
+  "protobuf",
+  "types-protobuf",
+  "grpcio",
+  "grpcio-status",
+  "grpcio-tools",
+  "types-grpcio",
+  "gunicorn",
+  "kafka-python",
+  "nest-asyncio",
+  "rich",
+  "requests",
+  "Flask",
+  "Flask-RESTful",
+  "sh",
+  "kubernetes",
+  "pytz",
+  "psutil",
+  "paramiko",
 ]
 
 [project.optional-dependencies]
+dev = ["ruff", "pre-commit", "pytest", "pytest-cov", "grpcio-testing", "grpcio==1.75", "grpcio-tools==1.75", "grpcio-status==1.75", "types-requests"]
 prod = ["paramiko[gssapi]"]
-dev = ["ruff", "pre-commit", "pytest", "pytest-cov", "grpcio-testing", "grpcio==1.75", "grpcio-tools==1.75", "grpcio-status==1.75"]
 test = ["pytest", "pytest-cov", "grpcio-testing", "grpcio==1.75", "grpcio-tools==1.75", "grpcio-status==1.75"]
 
 [project.scripts]
-fake_daq_application = "drunc.apps.fake_daqapp_rest:main"
+application-registry-service = "drunc.apps.app_connectivity_server:main"
+drunc-check-np0x-cluster = "drunc.apps.check_np0x_cluster:main"
+drunc-check-np0x-hw = "drunc.apps.check_np0x_hw_status:main"
 drunc-controller = "drunc.apps.controller:main"
 drunc-controller-shell = "drunc.apps.controller_shell:main"
+drunc-fsm-tests = "drunc.tests.fsm:main"
 drunc-process-manager = "drunc.apps.pm:main"
 drunc-process-manager-shell = "drunc.apps.pm_shell:main"
 drunc-session-manager = "drunc.apps.session_manager:main"
-drunc-unified-shell = "drunc.apps.unified_shell:main"
-drunc-fsm-tests = "drunc.tests.fsm:main"
-application-registry-service = "drunc.apps.app_connectivity_server:main"
-drunc-ssh-doctor = "drunc.apps.ssh_doctor:main"
 drunc-setup-ssh-config = "drunc.apps.ssh_configurator:main"
-drunc-check-np0x-hw = "drunc.apps.check_np0x_hw_status:main"
-drunc-check-np0x-cluster = "drunc.apps.check_np0x_cluster:main"
-
+drunc-ssh-doctor = "drunc.apps.ssh_doctor:main"
+drunc-unified-shell = "drunc.apps.unified_shell:main"
+fake_daq_application = "drunc.apps.fake_daqapp_rest:main"
 
 [tool.setuptools.packages.find]
 where = ["src"]
@@ -61,57 +60,56 @@ where = ["src"]
 "drunc.data.process_manager.schema" = ["*.json"]
 
 [tool.pytest.ini_options]
+addopts = "-v --tb=short --cov=drunc --cov=src/drunc"
 markers = [
-    "grpc: marks tests for gRPC isolation (run with --test-grpc)",
-    "paramiko: marks tests for paramiko isolation (run with --test-paramiko)",
+  "grpc: marks tests for gRPC isolation (run with --test-grpc)",
+  "paramiko: marks tests for paramiko isolation (run with --test-paramiko)",
 ]
-addopts = "-v --tb=short --cov=drunc --cov=src/drunc"
 testpaths = ["tests"]
 
 [tool.coverage.run]
-source = ["drunc"]
 omit = ["tests/*"]
+source = ["drunc"]
 
 # * See https://docs.astral.sh/ruff/rules/ for details on Ruff's linting options
 [tool.ruff.lint]
-select = [
-    "E",    # pycodestyle errors
-    "F",    # check for errors using PyFlakes
-    "I",    # best practices for import calls
-    "UP",   # suggestions for code modernization
-    "RUF",  # build in Ruff warnings
-    "R",    # refactoring suggestions
-]
 ignore = [
-    "E501", # Don't enforce line lengths within a linting context
+  "E501", # Don't enforce line lengths within a linting context
+]
+select = [
+  "E", # pycodestyle errors
+  "F", # check for errors using PyFlakes
+  "I", # best practices for import calls
+  "UP", # suggestions for code modernization
+  "RUF", # build in Ruff warnings
+  "R", # refactoring suggestions
 ]
-
 
 [tool.mypy]
-disallow_untyped_defs = true
-disallow_incomplete_defs = true
 check_untyped_defs = true
+disallow_any_generics = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_defs = true
 no_implicit_optional = true
+strict_equality = true
+warn_redundant_casts = true
 warn_return_any = true
 warn_unused_ignores = true
-warn_redundant_casts = true
-strict_equality = true
-disallow_any_generics = true
-disallow_subclassing_any = true
 
+disallow_any_decorated = true
 disallow_any_explicit = true
 disallow_any_expr = false
-disallow_any_decorated = true
 disallow_any_unimported = true
 
-warn_unused_configs = true
 show_error_codes = true
+warn_unused_configs = true
 
 # These overrides are because the library stubs dont exist and not on typeshed
 [[tool.mypy.overrides]]
-module = ["google.rpc.*"]
 ignore_missing_imports = true
+module = ["google.rpc.*"]
 
 [[tool.mypy.overrides]]
+ignore_missing_imports = true
 module = ["conffwk"]
-ignore_missing_imports = true

src/drunc/fsm/_protocols.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Protocol
+
+if TYPE_CHECKING:
+    from druncschema.controller_pb2 import FSMSequence
+
+    from drunc.fsm.core import PreOrPostTransitionSequence
+    from drunc.fsm.transition import Transition
+
+
+class ParameterProtocol(Protocol):
+    name: str
+    value: str
+
+
+class InitConfigurationProtocol(Protocol):
+    parameters: Iterable[ParameterProtocol]
+
+
+class ActorProtocol(Protocol):
+    def get_user_name(self) -> str: ...
+
+
+class DetConfigProtocol(Protocol):
+    id: str
+
+
+class DALProtocol(Protocol):
+    detector_configuration: DetConfigProtocol
+
+
+class DBProtocol(Protocol):
+    def get_dal(self, class_name: str, uid: str) -> DALProtocol: ...
+
+
+class OksKeyProtocol(Protocol):
+    session: str
+
+
+class RuntimeConfigurationProtocol(Protocol):
+    initial_data: str
+    oks_key: OksKeyProtocol
+
+
+class ConfigurationProtocol(Protocol):
+    id: str
+    db: DBProtocol
+    oks_key: OksKeyProtocol
+    initial_data: str
+    parameters: Iterable[ParameterProtocol]
+    name: str
+
+
+class ContextProtocol(Protocol):
+    actor: ActorProtocol
+    configuration: ConfigurationProtocol
+    runinfo: Dict[str, object]
+
+
+class SessionDalProtocol(Protocol):
+    segment: object
+    rte_script: Optional[str]
+
+
+class SSHCommandProtocol(Protocol):
+    def __call__(self, *args: str, _err_to_out: bool = ...) -> object: ...
+
+
+class ShErrorProtocol(Protocol):
+    stdout: bytes
+    stderr: bytes
+
+
+class ActionMethodProtocol(Protocol):
+    __name__: str
+    __module__: str
+    __self__: object
+
+    def __call__(self, *args: object, **kwargs: object) -> object: ...
+
+
+class FSMActionProtocol(Protocol):
+    name: str
+
+
+class FSMxTransitionProtocol(Protocol):
+    transition: str
+    order: list[str]
+    mandatory: list[str]
+
+
+class ConfigProtocol(Protocol):
+    def get_initial_state(self) -> str: ...
+    def get_states(self) -> List[str]: ...
+    def get_transitions(self) -> List[Transition]: ...
+    def get_sequences(self) -> List[FSMSequence]: ...
+    def get_pre_transitions_sequences(
+        self,
+    ) -> Dict[Transition, PreOrPostTransitionSequence]: ...
+    def get_post_transitions_sequences(
+        self,
+    ) -> Dict[Transition, PreOrPostTransitionSequence]: ...
+
+
+class ActionMethod(Protocol):
+    def __call__(self, *args: object, **kwargs: object) -> object: ...

src/drunc/fsm/action_factory.py

@@ -1,9 +1,15 @@
-import inspect
+from __future__ import annotations
 
-import conffwk
+import inspect
+from typing import Dict, Type, cast
 
 import drunc.fsm.exceptions as fsme
 from drunc.exceptions import DruncSetupException
+from drunc.fsm._protocols import (
+    ActionMethodProtocol,
+    ConfigurationProtocol,
+    FSMActionProtocol,
+)
 from drunc.fsm.actions.db_run_registry import DBRunRegistry
 from drunc.fsm.actions.file_logbook import FileLogbook
 from drunc.fsm.actions.file_run_registry import FileRunRegistry
@@ -17,26 +23,34 @@
 
 
 class FSMActionFactory:
-    def __init__(self):
+    _instance: FSMActionFactory | None = None
+
+    def __init__(self) -> None:
         raise DruncSetupException("Call get() instead")
 
-    def _get_pre_transitions(self, action):
-        retr = {}
+    def _get_pre_transitions(
+        self, action: FSMActionProtocol
+    ) -> Dict[str, ActionMethodProtocol]:
+        retr: Dict[str, ActionMethodProtocol] = {}
         for name, method in inspect.getmembers(action):
             if inspect.ismethod(method):
                 if name.startswith("pre_"):
-                    retr[name] = method
+                    retr[name] = cast(ActionMethodProtocol, method)
         return retr
 
-    def _get_post_transitions(self, action):
-        retr = {}
+    def _get_post_transitions(
+        self, action: FSMActionProtocol
+    ) -> Dict[str, ActionMethodProtocol]:
+        retr: Dict[str, ActionMethodProtocol] = {}
         for name, method in inspect.getmembers(action):
             if inspect.ismethod(method):
                 if name.startswith("post_"):
-                    retr[name] = method
+                    retr[name] = cast(ActionMethodProtocol, method)
         return retr
 
-    def _validate_signature(self, name, method, action):
+    def _validate_signature(
+        self, name: str, method: ActionMethodProtocol, action: str
+    ) -> None:
         sig = inspect.signature(method)
 
         if (
@@ -53,7 +67,7 @@ def _validate_signature(self, name, method, action):
             if p.annotation is inspect._empty:
                 raise fsme.MethodSignatureMissingAnnotation(action, name, pname)
 
-    def _validate_action(self, action):
+    def _validate_action(self, action: FSMActionProtocol) -> None:
         pre_transition = self._get_pre_transitions(action)
         post_transition = self._get_post_transitions(action)
 
@@ -67,8 +81,8 @@ def _validate_action(self, action):
             self._validate_signature(k, v, action.name)
 
     def get_action(
-        self, action_name: str, action_configuration: "conffwk.dal.FSMaction"
-    ):
+        self, action_name: str, action_configuration: ConfigurationProtocol
+    ) -> FSMActionProtocol:
         """
         Construct the action interface for the given action name and configuration.
 
@@ -84,7 +98,7 @@ def get_action(
             fsme.InvalidAction: If the constructed action does not have valid pre/post
                 transition methods.
         """
-        iface = None
+        iface: FSMActionProtocol | None = None
         match action_name:
             case "user-provided-run-number":
                 iface = UserProvidedRunNumber(action_configuration)
@@ -116,10 +130,8 @@ def get_action(
 
         return iface
 
-    _instance = None
-
     @classmethod
-    def get(cls):
+    def get(cls: Type[FSMActionFactory]) -> FSMActionFactory:
         if cls._instance is None:
             cls._instance = cls.__new__(cls)