Merge pull request #43 from ISISComputingGroup/Ticket8553_pyright_day_genie_blockserver

Ruff and pyright fixes

Author: Tom-Willemsen

src/genie_python/genie_blockserver.py

@@ -2,16 +2,24 @@
 
 import time
 from builtins import object
+from typing import TYPE_CHECKING, Any, Callable, ParamSpec, TypeVar
 
 from genie_python.utilities import compress_and_hex, dehex_decompress_and_dejson
 
+if TYPE_CHECKING:
+    from genie_python.genie import PVValue
+    from genie_python.genie_epics_api import API
+
+Param = ParamSpec("Param")
+RetType = TypeVar("RetType")
+
 # Prefix for block server pvs
 PV_BLOCK_NAMES = "BLOCKNAMES"
 BLOCK_SERVER_PREFIX = "CS:BLOCKSERVER:"
 
 
-def _blockserver_retry(func):
-    def wrapper(*args, **kwargs):
+def _blockserver_retry(func: Callable[Param, RetType]) -> Callable[Param, RetType]:
+    def wrapper(*args: Param.args, **kwargs: Param.kwargs) -> RetType:
         while True:
             try:
                 return func(*args, **kwargs)
@@ -27,38 +35,38 @@ def wrapper(*args, **kwargs):
 
 
 class BlockServer(object):
-    def __init__(self, api):
-        self.api = api
+    def __init__(self, api: "API") -> None:
+        self.api: "API" = api
 
-    def _get_pv_value(self, pv, as_string=False):
+    def _get_pv_value(self, pv: str, as_string: bool = False) -> "PVValue":
         """Just a convenient wrapper for calling the api's get_pv_value method"""
         return self.api.get_pv_value(self.api.prefix_pv_name(pv), as_string)
 
-    def _set_pv_value(self, pv, value, wait=False):
+    def _set_pv_value(self, pv: str, value: "PVValue | bytes", wait: bool = False) -> None:
         """Just a convenient wrapper for calling the api's set_pv_value method"""
         return self.api.set_pv_value(self.api.prefix_pv_name(pv), value, wait)
 
     @_blockserver_retry
-    def get_sample_par_names(self):
+    def get_sample_par_names(self) -> Any:  # noqa: ANN401
         """Get the current sample parameter names as a list."""
         # Get the names from the blockserver
         raw = self._get_pv_value(BLOCK_SERVER_PREFIX + "SAMPLE_PARS", True)
         return dehex_decompress_and_dejson(raw)
 
     @_blockserver_retry
-    def get_beamline_par_names(self):
+    def get_beamline_par_names(self) -> Any:  # noqa: ANN401
         """Get the current beamline parameter names as a list."""
         # Get the names from the blockserver
         raw = self._get_pv_value(BLOCK_SERVER_PREFIX + "BEAMLINE_PARS", True)
         return dehex_decompress_and_dejson(raw)
 
     @_blockserver_retry
-    def get_runcontrol_settings(self):
+    def get_runcontrol_settings(self) -> Any:  # noqa: ANN401
         """Get the current run-control settings."""
         raw = self._get_pv_value(BLOCK_SERVER_PREFIX + "GET_RC_PARS", True)
         return dehex_decompress_and_dejson(raw)
 
-    def reload_current_config(self):
+    def reload_current_config(self) -> None:
         """Reload the current configuration."""
         raw = compress_and_hex("1")
         self._set_pv_value(BLOCK_SERVER_PREFIX + "RELOAD_CURRENT_CONFIG", raw, True)

src/genie_python/genie_cachannel_wrapper.py

@@ -157,7 +157,7 @@ def putCB(epics_args: Tuple[str, int, int, int], user_args: Tuple[Event, ...]) -
     @staticmethod
     def set_pv_value(
         name: str,
-        value: "PVValue",
+        value: "PVValue|bytes",
         wait: bool = False,
         timeout: float = TIMEOUT,
         safe_not_quick: bool = True,
@@ -403,7 +403,7 @@ def connect_to_pv(ca_channel: CaChannel) -> None:
             raise UnableToConnectToPVException(ca_channel.name(), "Connection timeout (state)")
 
     @staticmethod
-    def check_for_enum_value(value: "PVValue", chan: CaChannel, name: str) -> "PVValue":
+    def check_for_enum_value(value: "PVValue|bytes", chan: CaChannel, name: str) -> "PVValue|bytes":
         """
         Check for string input for MBBI/BI records and replace with the equivalent index value.
 

src/genie_python/genie_epics_api.py

@@ -312,7 +312,7 @@ def init_instrument(
     def set_pv_value(
         self,
         name: str,
-        value: "PVValue",
+        value: "PVValue|bytes",
         wait: bool = False,
         attempts: int = 3,
         is_local: bool = False,

src/genie_python/genie_p4p_wrapper.py

@@ -47,7 +47,7 @@ def _log_error(message: str) -> None:
     @staticmethod
     def set_pv_value(
         name: str,
-        value: "PVValue",
+        value: "PVValue|bytes",
         wait: bool = False,
         timeout: float = TIMEOUT,
         safe_not_quick: bool = True,

src/genie_python/genie_pv_connection_protocol.py

@@ -9,7 +9,7 @@
 class GeniePvConnectionProtocol(Protocol):
     @staticmethod
     def set_pv_value(
-        name: str, value: "PVValue", wait: bool, timeout: float, safe_not_quick: bool
+        name: str, value: "PVValue|bytes", wait: bool, timeout: float, safe_not_quick: bool
     ) -> None: ...
 
     @staticmethod