Ticket8378

.git-blame-ignore-revs

@@ -0,0 +1,2 @@
+dad5500447f876c5aebfdb49d8a1cc4593561ce5
+ca3635948d7c51c269d2b4ec00507a39e9822a10

.github/workflows/linter.yml

@@ -0,0 +1,7 @@
+name: Linter
+on: [pull_request]
+jobs:
+  call-workflow:
+    uses: ISISComputingGroup/reusable-workflows/.github/workflows/linters.yml@main
+    with:
+      compare-branch: origin/main

system_tests/lewis_emulators/PearlPC/__init__.py

@@ -1,5 +1,5 @@
-from .device import SimulatedPearlPC
 from ..lewis_versions import LEWIS_LATEST
+from .device import SimulatedPearlPC
 
 framework_version = LEWIS_LATEST
-__all__ = ['SimulatedPearlPC']
+__all__ = ["SimulatedPearlPC"]

system_tests/lewis_emulators/PearlPC/device.py

@@ -1,14 +1,12 @@
-import time
-import threading
+from collections import OrderedDict
+
 from lewis.devices import StateMachineDevice
-from lewis.core import approaches
+
 from .states import DefaultState
-from collections import OrderedDict
 
 
 class SimulatedPearlPC(StateMachineDevice):
-
     def _initialize_data(self, status_dictionary=None):
         if status_dictionary is None:
             status_dictionary = {}
         self.status_dictionary = status_dictionary
@@ -19,7 +17,7 @@
         self.out_error = "}{<7f>w"
         self.out_terminator_in_error = ""
 
     def add_to_dict(self, value_id: str, unvalidated_value: object):
         """
         Add device state parameters to a dictionary.
         @param value_id: (str) dictionary key for each device parameter
@@ -27,39 +25,39 @@
         """
         self.status_dictionary[value_id] = unvalidated_value
 
     def _get_state_handlers(self):
         return {
-            'default': DefaultState(),
+            "default": DefaultState(),
         }
 
     def _get_initial_state(self):
-        return 'default'
+        return "default"
 
     def _get_transition_handlers(self):
         return OrderedDict([])
 
     def re_initialise(self):
         self.connected = True
 
-        self.initial_id_prefix = 1111 # 4 digits
-        self.secondary_id_prefix = 1111 # 4 digits
+        self.initial_id_prefix = 1111  # 4 digits
+        self.secondary_id_prefix = 1111  # 4 digits
         self.em_stop_status = 0  # Bool [0-1]
         self.run_bit = 0  # Bool [0-1]
-        self.reset_value = 0 # [0-4]
+        self.reset_value = 0  # [0-4]
         self.piston_reset_phase = 0
         self.stop_bit = 0  # Bool [0-1]
         self.busy_bit = 0  # Bool [0-1]
         self.go_status = 0
-        self.am_mode = 1 # auto mode (run from host) rather than manual
+        self.am_mode = 1  # auto mode (run from host) rather than manual
         self.loop_mode = 0  # Bool [0-1]
         self.seal_fail_value = 0
-        self.seal_fail_status = 0 # set to 1 if sudden pressure drop more than seal_fail_value
+        self.seal_fail_status = 0  # set to 1 if sudden pressure drop more than seal_fail_value
         self.last_error_code = 0
         self.pressure_rate = 0
         self.min_value_pre_servoing = 0
         self.setpoint_value = 0
         self.max_value_pre_servoing = 0
-        self.inputs = 0 # a 9 digit number like 111111001 showing input status
+        self.inputs = 0  # a 9 digit number like 111111001 showing input status
         self.cell_pressure = 0
         self.pump_pressure = 0
         self.transducer_difference_threshold = 2
@@ -75,9 +73,9 @@
         self.run_requested = 0
         self.stop_requested = 0
         self.reset_requested = 0
-        self.ramping = 0 # ramping to setpoint as opposed to closed loop stabilisation?
+        self.ramping = 0  # ramping to setpoint as opposed to closed loop stabilisation?
 
     def get_pressure(self):
         value = 0.0
         if self.algorithm == "a":
             value = (self.cell_pressure + self.pump_pressure) / 2.0
@@ -94,15 +92,15 @@
             value = weight * self.cell_pressure + (1.0 - weight) * self.pump_pressure
         return int(value)
 
     def stop(self):
         self.stop_requested = 1
-    
+
     def reset(self):
         self.reset_requested = 1
-    
+
     def run(self):
         self.run_requested = 1
-    
+
     def poller(self):
         self.inputs = int("011110000") + int("000000001") * self.am_mode
         if self.reset_requested:
@@ -135,7 +133,7 @@
             self.last_error_code = 12
             self.stop_requested = 1
 
-# need to do closed loop better
+    # need to do closed loop better
     def running(self):
         pressure = self.get_pressure()
         if self.ramping == 1:
@@ -149,12 +147,12 @@
                 incr = self.pressure_rate
             if pressure < self.setpoint_value:
                 self.pump_pressure = self.pump_pressure + incr
-                self.cell_pressure = self.pump_pressure # for simplicity
+                self.cell_pressure = self.pump_pressure  # for simplicity
             else:
                 self.pump_pressure = self.pump_pressure - incr
-                self.cell_pressure = self.pump_pressure # for simplicity
-        
-        # if self.loop_mode == 1:    maintain between min_value_pre_servoing and max_value_pre_servoing             
+                self.cell_pressure = self.pump_pressure  # for simplicity
+
+        # if self.loop_mode == 1:    maintain between min_value_pre_servoing and max_value_pre_servoing
 
         if abs(self.cell_pressure - self.pump_pressure) > self.transducer_difference_threshold:
             self.last_error_code = 10
@@ -218,7 +216,7 @@
         print(f"Received seal fail bit {sf_status}")
         self.seal_fail_status = sf_status
         self.add_to_dict(value_id="sf_status", unvalidated_value=self.seal_fail_status)
-    
+
     def set_go(self, go_status: int):
         """
         Set GO status to to 1 if system was initiated by host.

system_tests/lewis_emulators/PearlPC/interfaces/__init__.py

@@ -1,3 +1,3 @@
 from .stream_interface import PearlPCStreamInterface
 
-__all__ = ['PearlPCStreamInterface']
+__all__ = ["PearlPCStreamInterface"]

system_tests/lewis_emulators/PearlPC/interfaces/stream_interface.py

@@ -1,6 +1,6 @@
-from lewis.adapters.stream import StreamInterface, Cmd
-from lewis.utils.command_builder import CmdBuilder
+from lewis.adapters.stream import StreamInterface
 from lewis.core.logging import has_log
+from lewis.utils.command_builder import CmdBuilder
 from lewis.utils.replies import conditional_reply
 
 
@@ -21,21 +21,33 @@ class PearlPCStreamInterface(StreamInterface):
         CmdBuilder("set_mn").escape("mn").arg("[0-9]{4}", argument_mapping=int).eos().build(),
         CmdBuilder("set_sp").escape("sp").arg("[0-9]{4}", argument_mapping=int).eos().build(),
         CmdBuilder("set_mx").escape("mx").arg("[0-9]{4}", argument_mapping=int).eos().build(),
-        CmdBuilder("reset").escape("reset").eos().build(), 
-        CmdBuilder("run").escape("run").eos().build(), 
-        CmdBuilder("stop").escape("stop").eos().build(), 
-        CmdBuilder("set_t").escape("t").arg("[1-2]0[1-3][0-9]{2}0[1-3]").eos().build(), 
+        CmdBuilder("reset").escape("reset").eos().build(),
+        CmdBuilder("run").escape("run").eos().build(),
+        CmdBuilder("stop").escape("stop").eos().build(),
+        CmdBuilder("set_t").escape("t").arg("[1-2]0[1-3][0-9]{2}0[1-3]").eos().build(),
         CmdBuilder("set_th").escape("th").arg("[0-9]{4}", argument_mapping=int).eos().build(),
-        CmdBuilder("transducer_reset").escape("tr").eos().build(), 
+        CmdBuilder("transducer_reset").escape("tr").eos().build(),
         CmdBuilder("set_algorithm").escape("a").arg("[a12hlw][0-9]{0,2}").eos().build(),
         CmdBuilder("get_dt").escape("dt").eos().build(),
-        CmdBuilder("set_user_stop_limit").escape("ul").arg("[0-9]{4}", argument_mapping=int).eos().build(),
+        CmdBuilder("set_user_stop_limit")
+        .escape("ul")
+        .arg("[0-9]{4}", argument_mapping=int)
+        .eos()
+        .build(),
         CmdBuilder("show_limits").escape("ls").eos().build(),
         CmdBuilder("get_memory").escape("vr").arg("[0-9]{4}", argument_mapping=int).eos().build(),
         CmdBuilder("set_pos_lim").escape("d+").arg("[0-9]{4}", argument_mapping=int).eos().build(),
         CmdBuilder("set_neg_lim").escape("d-").arg("[0-9]{4}", argument_mapping=int).eos().build(),
-        CmdBuilder("set_pos_offset").escape("o+").arg("[0-9]{1}", argument_mapping=int).eos().build(),
-        CmdBuilder("set_neg_offset").escape("o-").arg("[0-9]{1}", argument_mapping=int).eos().build(),
+        CmdBuilder("set_pos_offset")
+        .escape("o+")
+        .arg("[0-9]{1}", argument_mapping=int)
+        .eos()
+        .build(),
+        CmdBuilder("set_neg_offset")
+        .escape("o-")
+        .arg("[0-9]{1}", argument_mapping=int)
+        .eos()
+        .build(),
     }
 
     in_terminator = "\r"
@@ -51,33 +63,37 @@ def get_st(self):
         @return: (str) A formatted string containing all set device parameters describing current device status.
         """
         self._device.poller()
-        return f"Status Report{self.out_terminator}" \
-               f"Em Ru Re St By Go AM sl sf Er   ra    mn    sp    mx  Press    Inputs{self.out_terminator}" \
-               f"{self._device.em_stop_status} " \
-               f"{self._device.run_bit} " \
-               f"{self._device.reset_value} " \
-               f"{self._device.stop_bit} " \
-               f"{self._device.busy_bit} " \
-               f"{self._device.go_status} " \
-               f"{self._device.am_mode} " \
-               f"{self._device.loop_mode} " \
-               f"{self._device.seal_fail_status} " \
-               f"{self._device.last_error_code} " \
-               f"{self._device.pressure_rate} " \
-               f"{self._device.min_value_pre_servoing} " \
-               f"{self._device.setpoint_value} " \
-               f"{self._device.max_value_pre_servoing} " \
-               f"{self._device.get_pressure()} " \
-               f"{self._device.inputs:09d}{self.out_terminator}" \
-               f"OK"
+        return (
+            f"Status Report{self.out_terminator}"
+            f"Em Ru Re St By Go AM sl sf Er   ra    mn    sp    mx  Press    Inputs{self.out_terminator}"
+            f"{self._device.em_stop_status} "
+            f"{self._device.run_bit} "
+            f"{self._device.reset_value} "
+            f"{self._device.stop_bit} "
+            f"{self._device.busy_bit} "
+            f"{self._device.go_status} "
+            f"{self._device.am_mode} "
+            f"{self._device.loop_mode} "
+            f"{self._device.seal_fail_status} "
+            f"{self._device.last_error_code} "
+            f"{self._device.pressure_rate} "
+            f"{self._device.min_value_pre_servoing} "
+            f"{self._device.setpoint_value} "
+            f"{self._device.max_value_pre_servoing} "
+            f"{self._device.get_pressure()} "
+            f"{self._device.inputs:09d}{self.out_terminator}"
+            f"OK"
+        )
 
     @conditional_reply("connected")
     def get_id_prefix(self):
         """
         Returns ID prefixes (first 2 numbers selectable by the user formatted as: "<IDP_1> <IDP_2>")
         @return: (str) formatted string returning ID prefixes set by default or by user.
         """
-        print(f"ID prefix set to: {self._device.initial_id_prefix} {self._device.secondary_id_prefix}")
+        print(
+            f"ID prefix set to: {self._device.initial_id_prefix} {self._device.secondary_id_prefix}"
+        )
         return f"{self._device.initial_id_prefix:04d} {self._device.secondary_id_prefix:04d} ISIS PEARL pressure intensifier V2.3"
 
     @conditional_reply("connected")
@@ -88,7 +104,7 @@ def set_si(self, id_prefix: int):
         @param id_prefix: (int) Prefix to ID for a unit - range [0000-9999]
         """
         print(f"SI prefix value received: {id_prefix}")
-        if (id_prefix < 0 or id_prefix > 9999):
+        if id_prefix < 0 or id_prefix > 9999:
             print("ERROR: invalid si value")
 
         self._device.initial_id_prefix = id_prefix
@@ -104,7 +120,7 @@ def set_sd(self, secondary_id_prefix: int):
         @param secondary_id_prefix: (int) Prefix to ID for a unit - range [0000-9999]
         """
         print(f"SD prefix value received: {secondary_id_prefix}")
-        if (secondary_id_prefix < 0 or secondary_id_prefix > 9999):
+        if secondary_id_prefix < 0 or secondary_id_prefix > 9999:
             print("ERROR: invalid sd value")
         self._device.secondary_id_prefix = secondary_id_prefix
         self._device.add_to_dict(value_id="sd", unvalidated_value=self._device.secondary_id_prefix)
@@ -116,8 +132,8 @@ def reset(self):
         Reset to fully open pistons
         """
         if self._device.get_pressure() < 100:
-            print(f"starting reset")
-            self._device.reset() # set phase to resetting, this starts reset
+            print("starting reset")
+            self._device.reset()  # set phase to resetting, this starts reset
         else:
             print("ERROR: cannot reset as pressure too high")
             self._device.last_error_code = 1
@@ -130,12 +146,12 @@ def set_sloop(self, sloop: int):
         0 = open loop mode
         1 = close loop mode
         Open loop mode ramps to pressure in control system and then stop until anther command is sent.
-        Close loop mode will ramp to setpoint pressure value and remain active monitoring delivered pressure, 
+        Close loop mode will ramp to setpoint pressure value and remain active monitoring delivered pressure,
         acting on mn and mx pressure values.
         @param sloop: (int) integer value setting system to open or closed loop - range [0-1]
         """
         print(f"sloop value recieved: {sloop}")
-        if (sloop < 0 or sloop > 1):
+        if sloop < 0 or sloop > 1:
             print("ERROR: invalid sloop")
         self._device.loop_mode = sloop
         self._device.add_to_dict(value_id="sloop", unvalidated_value=self._device.loop_mode)
@@ -149,8 +165,8 @@ def set_sf(self, seal_fail_value: int):
         @param seal_fail_value: (int) Seal Fail Mode Trigger Value - range [0001-0999]
         """
         print(f"Seal Fail mode trigger value received: {seal_fail_value}")
-        if (seal_fail_value < 1 or seal_fail_value > 999):
-           print("ERROR: invalid seal fail value")
+        if seal_fail_value < 1 or seal_fail_value > 999:
+            print("ERROR: invalid seal fail value")
         self._device.seal_fail_value = seal_fail_value
         self._device.add_to_dict(value_id="sf", unvalidated_value=self._device.seal_fail_value)
         return ""
@@ -160,7 +176,7 @@ def error_reset(self):
         """
         reset the last error code execpt for code 8 (seal fail)
         """
-        if (self._device.last_error_code != 8):
+        if self._device.last_error_code != 8:
             self._device.last_error_code = 0
             print(f"Resetting last error code: {self._device.last_error_code}")
         else:
@@ -176,11 +192,11 @@ def set_ra(self, pressure_rate: int):
         @param pressure_rate: (int) Pressure rate within range [0001-0040]
         """
         print(f"Pressure Rate Received: {pressure_rate}")
-        if (pressure_rate < 0 or pressure_rate > 40):
+        if pressure_rate < 0 or pressure_rate > 40:
             print("ERROR: invalid pressure rate")
-        if (pressure_rate == 0):
-            self._device.pressure_rate = 10 # maximum slew rate of the motor?
-        else:            
+        if pressure_rate == 0:
+            self._device.pressure_rate = 10  # maximum slew rate of the motor?
+        else:
             self._device.pressure_rate = pressure_rate
         self._device.add_to_dict(value_id="ra", unvalidated_value=self._device.pressure_rate)
         return ""
@@ -192,11 +208,13 @@ def set_mn(self, min_measured: int):
         This will only be acted upon in closed loop mode.
         @param min_measured: (int) minimum pressure value before re-servoing - range [0001-9999]
         """
-        if (min_measured < 1 or min_measured > 9999):
+        if min_measured < 1 or min_measured > 9999:
             print("ERROR: invalid min measured")
         print(f"Minimum value before re-servoing received: {min_measured}")
         self._device.min_value_pre_servoing = min_measured
-        self._device.add_to_dict(value_id="mn", unvalidated_value=self._device.min_value_pre_servoing)
+        self._device.add_to_dict(
+            value_id="mn", unvalidated_value=self._device.min_value_pre_servoing
+        )
         return ""
 
     @conditional_reply("connected")
@@ -208,7 +226,7 @@ def set_sp(self, setpoint: int):
         @param setpoint: (int) Set Point trigger value - range [0001-1000]
         """
         print(f"Setpoint value received: {setpoint}")
-        if (setpoint < 1 or setpoint > 1000):
+        if setpoint < 1 or setpoint > 1000:
             print("ERROR: invalid setpoint")
         self._device.setpoint_value = setpoint
         self._device.add_to_dict(value_id="sp", unvalidated_value=self._device.setpoint_value)
@@ -220,14 +238,16 @@ def set_mx(self, max_measured: int):
         set the maximum measured value before re-servoing
         @param max_measured: (integer) maximum measured value before re-servoing - range [0001-9999]
         """
-        if (max_measured < 1 or max_measured > 9999):
+        if max_measured < 1 or max_measured > 9999:
             print("ERROR: invalid max measured")
         print(f"Maximum measured value before re-servoing received: {max_measured}")
         self._device.max_value_pre_servoing = max_measured
-        self._device.add_to_dict(value_id="mx", unvalidated_value=self._device.max_value_pre_servoing)
+        self._device.add_to_dict(
+            value_id="mx", unvalidated_value=self._device.max_value_pre_servoing
+        )
         return ""
 
-    def handle_error(self, request:object, error:object):
+    def handle_error(self, request: object, error: object):
         """
         Return any errors which have occurred when sending requests to device.
         @return: (str) Formatted error message
@@ -287,9 +307,11 @@ def set_user_stop_limit(self, value: int):
     @conditional_reply("connected")
     def show_limits(self):
         print("show_limits")
-        return f"User +Change +Offset -Change -Offset{self.out_terminator}" \
-               f"{self._device.user_stop_limit} {self._device.dir_plus} {self._device.offset_plus} {self._device.dir_minus} {self._device.offset_minus}{self.out_terminator}" \
-               f"OK"
+        return (
+            f"User +Change +Offset -Change -Offset{self.out_terminator}"
+            f"{self._device.user_stop_limit} {self._device.dir_plus} {self._device.offset_plus} {self._device.dir_minus} {self._device.offset_minus}{self.out_terminator}"
+            f"OK"
+        )
 
     @conditional_reply("connected")
     def get_memory(self, address: int):