pyright fixes for genie_advanced

src/genie_python/genie_advanced.py

@@ -5,10 +5,11 @@
 """
 
 import contextlib
-from datetime import datetime, timedelta
+from datetime import UTC, datetime, timedelta
 from time import sleep
 from typing import Any, Callable, Iterator, TypedDict
 
+import numpy as np
 import numpy.typing as npt
 
 from genie_python.genie_api_setup import (
@@ -52,7 +53,7 @@ def assert_in_manager_mode() -> None:
 
 
 @contextlib.contextmanager
-def motor_in_set_mode(pv_name: str) -> Iterator:
+def motor_in_set_mode(pv_name: str) -> Iterator[None]:
     """
     Uses a context to place motor into set mode and ensure that it leaves
     set mode after context has ended. If it can not set the mode correctly
@@ -134,7 +135,7 @@ def get_pv_from_block(block: str) -> str:
 @usercommand
 @helparglist("pv str")
 @log_command_and_handle_exception
-def pv_exists(pv: str, is_local: bool = False) -> None:
+def pv_exists(pv: str, is_local: bool = False) -> bool:
     """
     Check if PV exists.
 
@@ -159,13 +160,13 @@ def wait_for_pv(
         value: The value to wait for
         maxwait (int, optional): The maximum time to wait for in seconds
     """
-    start_time = datetime.utcnow()
+    start_time = datetime.now(UTC)
     while True:
         curr_value = __api.get_pv_value(pv)
         if curr_value == value:
             break
         if maxwait is not None:
-            if timedelta(seconds=maxwait) < datetime.utcnow() - start_time:
+            if timedelta(seconds=maxwait) < datetime.now(UTC) - start_time:
                 break
         sleep(DELAY_IN_WAIT_FOR_SLEEP_LOOP)
         check_break(2)
@@ -400,7 +401,9 @@ def get_exp_data(
 @usercommand
 @helparglist("")
 @log_command_and_handle_exception
-def get_spectrum_data(with_spec_zero: bool = True, with_time_bin_zero: bool = False) -> npt.NDArray:
+def get_spectrum_data(
+    with_spec_zero: bool = True, with_time_bin_zero: bool = False
+) -> npt.NDArray[np.float32]:
     """
     Get the event mode spectrum data as ND array.
 

src/genie_python/genie_pre_post_cmd_manager.py

@@ -1,21 +1,44 @@
 from builtins import object
+from typing import Any, Callable
 
 
 class PrePostCmdManager(object):
     """
-    A class to manager the precmd and postcmd commands such as used in begin, end, abort, resume, pause.
+    A class to manager the precmd and postcmd commands such as used in begin, end, abort, resume,
+     pause.
     """
+    def begin_precmd(self, **kwargs: Any) -> str | None:
+        return
 
-    def __init__(self):
-        self.begin_precmd = lambda **pars: None
-        self.begin_postcmd = lambda **pars: None
-        self.abort_precmd = lambda **pars: None
-        self.abort_postcmd = lambda **pars: None
-        self.end_precmd = lambda **pars: None
-        self.end_postcmd = lambda **pars: None
-        self.pause_precmd = lambda **pars: None
-        self.pause_postcmd = lambda **pars: None
-        self.resume_precmd = lambda **pars: None
-        self.resume_postcmd = lambda **pars: None
-        self.cset_precmd = lambda **pars: True
-        self.cset_postcmd = lambda **pars: None
+    def begin_postcmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def abort_precmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def abort_postcmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def end_precmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def end_postcmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def pause_precmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def pause_postcmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def resume_precmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def resume_postcmd(self, **kwargs: Any) -> str | None:
+        return
+
+    def cset_precmd(self, **kwargs: Any) -> bool:
+        return True
+
+    def cset_postcmd(self, **kwargs: Any) -> str | None:
+        return

src/genie_python/genie_simulate_impl.py

@@ -1085,12 +1085,16 @@ def __init__(
         self.sample_pars = {}
         self.strict_block = strict_block
         self.logger = GenieLogger(sim_mode=True)
+        self.exp_data = None
 
     def set_instrument(
         self, pv_prefix: str, globs: dict | None, import_instrument_init: bool
     ) -> None:
         self.inst_prefix = pv_prefix
 
+    def get_instrument(self) -> str:
+        return self.inst_prefix
+
     def prefix_pv_name(self, name: str) -> str:
         """Adds the instrument prefix to the specified PV"""
         if self.inst_prefix is not None and not name.startswith(self.inst_prefix):
@@ -1100,7 +1104,7 @@ def prefix_pv_name(self, name: str) -> str:
         return name
 
     def set_pv_value(
-        self, name: str, value: str, wait: bool = False, is_local: bool = False
+        self, name: str, value: "PVValue", wait: bool = False, is_local: bool = False
     ) -> None:
         if is_local:
             name = self.prefix_pv_name(name)
@@ -1119,7 +1123,7 @@ def get_pv_value(
             % (name, to_string, attempts, is_local)
         )
 
-    def pv_exists(self, name: str) -> bool:
+    def pv_exists(self, name: str, is_local: bool = False) -> bool:
         return True
 
     def reload_current_config(self) -> None:
@@ -1189,6 +1193,9 @@ def get_block_data(self, block: str, fail_fast: bool = False) -> dict():
         ans["alarm"] = self.get_alarm_from_block(block)
         return ans
 
+    def get_pv_from_block(self, block_name: str) -> str:
+        return block_name
+
     def set_multiple_blocks(self, names: list, values: list) -> None:
         temp = list(zip(names, values))
         for name, value in temp: