add group 3 hall probe emulator

Author: rerpha

lewis/devices/group3hallprobe/__init__.py

@@ -0,0 +1,3 @@
+from .device import SimulatedGroup3HallProbe
+
+__all__ = ["SimulatedGroup3HallProbe"]

lewis/devices/group3hallprobe/device.py

@@ -0,0 +1,91 @@
+from collections import OrderedDict
+from enum import Enum
+from typing import Callable, assert_never
+
+from lewis.devices import StateMachineDevice
+
+from .states import DefaultState, State
+
+GAUSS_PER_TESLA = 10_000.0
+
+
+class Ranges(Enum):
+    """
+    Expresses a measurement range that a Probe can be in.
+
+    Values and indices from device manual.
+    """
+
+    R0 = 0  # 0.3 Tesla range
+    R1 = 1  # 0.6 Tesla range
+    R2 = 2  # 1.2 Tesla range
+    R3 = 3  # 3.0 Tesla range
+
+
+def range_to_max_gauss(r: Ranges) -> float:
+    match r:
+        case Ranges.R0:
+            return 0.3 * GAUSS_PER_TESLA
+        case Ranges.R1:
+            return 0.6 * GAUSS_PER_TESLA
+        case Ranges.R2:
+            return 1.2 * GAUSS_PER_TESLA
+        case Ranges.R3:
+            return 3.0 * GAUSS_PER_TESLA
+
+    assert_never(r)
+
+
+class Probe:
+    """
+    A single probe.
+    """
+
+    def __init__(self) -> None:
+        self.field = 0.0
+        self.temperature = 0.0
+        self.sensor_range = Ranges.R3
+        self.initialized = True
+
+    def is_over_range(self) -> bool:
+        return abs(self.field) > range_to_max_gauss(self.sensor_range)
+
+    def initialize(self) -> None:
+        self.sensor_range = Ranges.R3
+        self.initialized = True
+
+
+class SimulatedGroup3HallProbe(StateMachineDevice):
+    def _initialize_data(self) -> None:
+        """
+        Initialize all of the device's attributes.
+        """
+        self.connected = True
+        self.probes = {
+            0: Probe(),
+            1: Probe(),
+            2: Probe(),
+        }
+
+    def reset(self) -> None:
+        self._initialize_data()
+
+    def backdoor_set_field(self, probe_id: int, field: float) -> None:
+        self.probes[probe_id].field = field
+
+    def backdoor_set_temperature(self, probe_id: int, temperature: float) -> None:
+        self.probes[probe_id].temperature = temperature
+
+    def backdoor_set_initialized(self, probe_id: int, initialized: bool) -> None:
+        self.probes[probe_id].initialized = initialized
+
+    def _get_state_handlers(self) -> dict[str, State]:
+        return {
+            "default": DefaultState(),
+        }
+
+    def _get_initial_state(self) -> str:
+        return "default"
+
+    def _get_transition_handlers(self) -> dict[tuple[str, str], Callable[[], bool]]:
+        return OrderedDict([])

lewis/devices/group3hallprobe/interfaces/__init__.py

@@ -0,0 +1,3 @@
+from .stream_interface import Group3HallProbeStreamInterface
+
+__all__ = ["Group3HallProbeStreamInterface"]

lewis/devices/group3hallprobe/interfaces/stream_interface.py

@@ -0,0 +1,66 @@
+import logging
+
+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
+from lewis_emulators.group3hallprobe.device import Ranges, SimulatedGroup3HallProbe
+
+if_connected = conditional_reply("connected")
+
+
+@has_log
+class Group3HallProbeStreamInterface(StreamInterface):
+    in_terminator = "\n\r"  # Yes, really LF-CR not CR-LF
+    out_terminator = "\n\r"
+
+    def __init__(self) -> None:
+        super(Group3HallProbeStreamInterface, self).__init__()
+
+        self.log: logging.Logger
+        self.device: SimulatedGroup3HallProbe
+
+        # Commands that we expect via serial during normal operation
+        self.commands = {
+            CmdBuilder(self.initialize).escape("A").int().escape(" SE0GDR3GCNNUFG").eos().build(),
+            CmdBuilder(self.get_field).escape("A").int().escape(" F").eos().build(),
+            CmdBuilder(self.get_temperature).escape("A").int().escape(" T").eos().build(),
+            CmdBuilder(self.set_range).escape("A").int().escape(" R").int().eos().build(),
+        }
+
+    def handle_error(self, request: str, error: str | Exception) -> None:
+        """
+        If command is not recognised print and error
+
+        Args:
+            request: requested string
+            error: problem
+
+        """
+        self.log.error("An error occurred at request " + repr(request) + ": " + repr(error))
+
+    @if_connected
+    def initialize(self, probe_id: int) -> str:
+        self.device.probes[probe_id].initialize()
+        return f"A{probe_id} SE0GDR3GCNNUFG"
+
+    @if_connected
+    def get_field(self, probe_id: int) -> str:
+        probe = self.device.probes[probe_id]
+        if not probe.initialized:
+            return f"A{probe_id} F\n\runinitialized_bad_data"
+        if probe.is_over_range():
+            return f"A{probe_id} F\n\rOVER RANGE"
+        return f"A{probe_id} F\n\r{probe.field}"
+
+    @if_connected
+    def get_temperature(self, probe_id: int) -> str:
+        probe = self.device.probes[probe_id]
+        if not probe.initialized:
+            return f"A{probe_id} T\n\runinitialized_bad_data"
+        return f"A{probe_id} T\n\r{probe.temperature}C"
+
+    @if_connected
+    def set_range(self, probe_id: int, range_id: int) -> str:
+        self.device.probes[probe_id].sensor_range = Ranges(range_id)
+        return f"A{probe_id} R{range_id}"

lewis/devices/group3hallprobe/states.py

@@ -0,0 +1,5 @@
+from lewis.core.statemachine import State
+
+
+class DefaultState(State):
+    pass