more ruff

Author: rerpha

installation_and_upgrade/calibration_files_updater.py

@@ -7,6 +7,7 @@
 import logging
 import os
 import subprocess
+from typing import Any, List
 
 import git
 from epics import caget
@@ -25,7 +26,7 @@ class CalibrationsFolder:
     DRIVE_LETTER = "q"
 
     @staticmethod
-    def disconnect_from_drive():
+    def disconnect_from_drive() -> bool:
         """
         Returns: True if disconnect is successful, else False.
         """
@@ -38,7 +39,7 @@ def disconnect_from_drive():
             == 0
         )
 
-    def connect_to_drive(self):
+    def connect_to_drive(self) -> bool:
         """
         Returns: True if the connection is successful, else False
         """
@@ -58,23 +59,24 @@ def connect_to_drive(self):
             == 0
         )
 
-    def __init__(self, instrument_host, username, password):
+    def __init__(self, instrument_host: str, username: str, password: str) -> None:
         self.username_with_domain = f"{instrument_host}\\{username}"
         self.network_location = r"\\{}\c$\Instrument\Settings\config\common".format(instrument_host)
         self.password = password
 
-    def __enter__(self):
+    def __enter__(self) -> git.Repo | None:
         """
-        Returns: A git repository for the remote calibration folder if connection is successful, else None.
+        Returns: A git repository for the
+        remote calibration folder if connection is successful, else None.
         """
         self.disconnect_from_drive()
         return git.Repo(self.network_location) if self.connect_to_drive() else None
 
-    def __exit__(self, *args):
+    def __exit__(self, *args: Any) -> None:  # noqa: ANN401
         self.disconnect_from_drive()
 
 
-def get_instrument_hosts():
+def get_instrument_hosts() -> List[str]:
     """
     Returns: A collection of instrument host names
     """
@@ -86,7 +88,9 @@ def get_instrument_hosts():
     )
 
 
-def update_instrument(host, username, password, logger, dry_run=False):
+def update_instrument(
+    host: str, username: str, password: str, logger: logging.Logger, dry_run: bool = False
+) -> bool:
     """
     Updates the calibration files on the named host.
 

installation_and_upgrade/ibex_install_utils/ca_utils.py

@@ -2,9 +2,10 @@
 import zlib
 from ast import literal_eval
 from socket import gethostname
-from typing import Any
+from typing import Any, List
 
 import epicscorelibs.path.pyepics  # noqa: F401
+import numpy as np
 from epics import caget, caput
 
 from ibex_install_utils.file_utils import FileUtils
@@ -59,7 +60,7 @@ def dehex_and_decompress(value: bytes | str) -> str:
     return zlib.decompress(bytes.fromhex(value)).decode("utf-8")
 
 
-def dehex_decompress_and_dejson(value: str | bytes) -> Any:  # No known type
+def dehex_decompress_and_dejson(value: str | bytes) -> Any:  # noqa: ANN401
     """
     Convert string from zipped hexed json to a python representation
     :param value: value to convert
@@ -125,14 +126,15 @@ def get_machine_details_from_identifier(
     # then find the first match where pvPrefix equals the machine identifier
     # that's been passed to this function if it is not found instrument_details will be None
     instrument_details = None
-    try:
+    instlist_raw = caget("CS:INSTLIST")
+    if instlist_raw is not None and isinstance(instlist_raw, np.ndarray):
         instrument_list = dehex_decompress_and_dejson(
-            caget("CS:INSTLIST").tobytes().decode().rstrip("\x00")
+            instlist_raw.tobytes().decode().rstrip("\x00")
         )
         instrument_details = next(
             (inst for inst in instrument_list if inst["pvPrefix"] == machine_identifier), None
         )
-    except AttributeError:
+    else:
         print("Error getting instlist, \nContinuing execution...")
 
     if instrument_details is not None:
@@ -171,14 +173,13 @@ class CaWrapper:
     Wrapper around genie python's channel access class providing some useful abstractions.
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         """
-        Setting instrument is necessary because genie_python is being run from a network drive so it may not know
-        where it is.
+        Get the current PV prefix.
         """
         _, _, self.prefix = get_machine_details_from_identifier()
 
-    def get_local_pv(self, name):
+    def get_local_pv(self, name: str) -> str | int | float | None:
         """
         Get PV with the local PV prefix appended
 
@@ -188,9 +189,9 @@ def get_local_pv(self, name):
         Returns:
             None if the PV was not connected
         """
-        return caget(f"{self.prefix}{name}")
+        return caget(f"{self.prefix}{name}")  # type: ignore
 
-    def get_object_from_compressed_hexed_json(self, name):
+    def get_object_from_compressed_hexed_json(self, name: str) -> bytes | None:
         """
         Gets an object from a compressed hexed json PV
 
@@ -206,26 +207,24 @@ def get_object_from_compressed_hexed_json(self, name):
         else:
             return FileUtils.dehex_and_decompress(data)
 
-    def get_blocks(self):
+    def get_blocks(self) -> List[str]:
         """
         Returns:
             A collection of blocks, or None if the PV was not connected
         """
-        try:
-            blocks_hexed = caget(f"{self.prefix}CS:BLOCKSERVER:BLOCKNAMES").tobytes()
-        except AttributeError as e:
-            print("Error getting blocks.")
-            raise e
-        return literal_eval(FileUtils.dehex_and_decompress(blocks_hexed).decode())
-
-    def cget(self, block: str) -> Any:
+        blocks_hexed = caget(f"{self.prefix}CS:BLOCKSERVER:BLOCKNAMES")
+        if blocks_hexed is None:
+            raise Exception("Error getting blocks from blockserver PV.")
+        return literal_eval(FileUtils.dehex_and_decompress(blocks_hexed.tobytes()).decode())
+
+    def cget(self, block: str) -> str | int | float | None:
         """
         Returns:
             A collection of blocks, or None if the PV was not connected.
         """
-        return caget(f"{self.prefix}CS:SB:{block}")
+        return caget(f"{self.prefix}CS:SB:{block}")  # type: ignore
 
-    def set_pv(self, pv, value, is_local: bool = False):
+    def set_pv(self, pv: str, value: str | float | int, is_local: bool = False) -> None:
         """
         Sets the value of a PV.
         """

installation_and_upgrade/ibex_install_utils/motor_params.py

@@ -1,9 +1,11 @@
 """
-Script to extract information from motors to be consumed by the motion controls team. Exports data as CSV. To run,
-load the script into a genie_python console and run as a standard user script.
+Script to extract information from motors to be consumed by the motion controls team.
+Exports data as CSV.
+To run, load the script into a genie_python console and run as a standard user script.
 """
 
 import csv
+from typing import BinaryIO
 
 from aioca import CANothing, caget
 
@@ -24,7 +26,7 @@
     "Encoder Step Size (step per EGU)",
 )
 DECEL_DIST = "Max Deceleration Distance (EGU)"
-P, I, D = "P", "I", "D"
+P, I, D = "P", "I", "D"  # noqa: E741
 
 K1, K2, K3 = "K1", "K2", "K3"
 MTR_TYPE, ENC_TYPE, AUX_ENC_TYPE = "Motor Type", "Encoder Type", "Aux Encoder Type"
@@ -110,13 +112,15 @@
 }
 
 
-async def get_params_and_save_to_file(file_reference, num_of_controllers=8):
+async def get_params_and_save_to_file(
+    file_reference: BinaryIO, num_of_controllers: int = 8
+) -> None:
     """
     Gets all the motor parameters and saves them to an open file reference as a csv.
 
     Args:
         file_reference (BinaryIO): The csv file to save the data to.
-        num_of_controllers (int, optional): The number of motor controllers on the instrument (default is 8)
+        num_of_controllers (int, optional): The number of motor controllers on the instrument
     """
     list_of_axis_pvs = []
     _, _, pv_prefix = get_machine_details_from_identifier()