added hydra driver

comet/driver/generic/motion_controller.py

@@ -99,4 +99,4 @@ def joystick_enabled(self) -> bool:
     @joystick_enabled.setter  # type: ignore
     @abstractmethod
     def joystick_enabled(self, value: bool) -> None:
-        ...
+        ...

comet/driver/itk/hydra.py

@@ -0,0 +1,137 @@
+from typing import Dict, Optional
+
+from comet.driver.generic import InstrumentError
+from comet.driver.generic.motion_controller import Position, MotionControllerAxis, MotionController
+
+__all__ = ["Hydra"]
+
+ERROR_MESSAGES: Dict[int, str] = {
+    0: "no error",
+    4: "internal error",
+    100: "devicenumber out of range",
+    101: "stack underflow or cmd not found at 0",
+    102: "undefined symbol",
+    1001: "wrong parameter type",
+    1002: "stack underflow: too few parameters on stack",
+    1003: "parameter out of range",
+    1004: "move out of limits requested",
+    2000: "undefined command",
+    3000: "no configuration file available",
+    3001: "error in configuration file",
+    3100: "most recent valid parameter set restored due to file corruption",
+}
+
+
+def parse_error(response: str) -> Optional[InstrumentError]:
+    code = int(response)
+    if code:
+        message = ERROR_MESSAGES.get(code, "unknown error")
+        return InstrumentError(code, message)
+    return None
+
+
+class HydraAxis(MotionControllerAxis):
+
+    def calibrate(self) -> None:
+        self.resource.write(f"{self.index:d} ncal")
+
+    def range_measure(self) -> None:
+        self.resource.write(f"{self.index:d} nrm")
+
+    @property
+    def is_calibrated(self) -> bool:
+        """Return True if axis is calibrated and range measured."""
+        result = self.resource.query(f"{self.index:d} nst")
+        return bool(int(result) & 0x18)
+
+    def move_absolute(self, value: float) -> None:
+        self.resource.write(f"{value:.3f} {self.index:d} nm")
+
+    def move_relative(self, value: float) -> None:
+        self.resource.write(f"{value:.3f} {self.index:d} nr")
+
+    @property
+    def position(self) -> float:
+        result = self.resource.query(f"{self.index:d} np")
+        return float(result)
+
+    @property
+    def is_moving(self) -> bool:
+        result = self.resource.query(f"{self.index:d} nst")
+        return bool(int(result) & 0x1)
+
+
+class Hydra(MotionController):
+
+    AXES = (1, 2)
+
+    def identify(self) -> str:
+        return self.resource.query("identify").strip()
+
+    def reset(self) -> None:
+        ...
+
+    def clear(self) -> None:
+        ...
+
+    def next_error(self) -> Optional[InstrumentError]:
+        response = self.resource.query("ge")
+        return parse_error(response)
+
+    def __getitem__(self, index: int) -> HydraAxis:
+        if index not in type(self).AXES:
+            raise IndexError(index)
+        return HydraAxis(self.resource, index)
+
+    def calibrate(self) -> None:
+        for index in type(self).AXES:
+            self.resource.write(f"{index:d} ncal")
+
+    def range_measure(self) -> None:
+        for index in type(self).AXES:
+            self.resource.write(f"{index:d} nrm")
+
+    @property
+    def is_calibrated(self) -> bool:
+        """Return True if all active axes are calibrated and range measured."""
+        status = int(self.resource.query(f"st"))
+        return bool(int(status & 0x18))
+
+    def move_absolute(self, position: Position) -> None:
+        values = [format(value, '.3f') for value in position]
+        self.resource.write(f"{values[0]} {values[1]} m")
+
+    def move_relative(self, position: Position) -> None:
+        values = [format(value, '.3f') for value in position]
+        self.resource.write(f"{values[0]} {values[1]} r")
+
+    def abort(self):
+        for index in type(self).AXES:
+            self.resource.write(f"{index:d} nabort")
+
+    def force_abort(self):
+        self.resource.write(chr(0x03))  # Ctrl+C
+
+    @property
+    def position(self) -> Position:
+        x, y = self.resource.query("p").split()
+        return [float(x), float(y)]
+
+    @property
+    def is_moving(self) -> bool:
+        result = self.resource.query("st")
+        return bool(int(result) & 0x1)
+
+    @property
+    def joystick_enabled(self) -> bool:
+        results = []
+        for index in type(self).AXES:
+            result = self.resource.query(f"{index:d} getmanctrl")
+            results.append(int(result))
+        return any(results)
+
+    @joystick_enabled.setter
+    def joystick_enabled(self, value: bool) -> None:
+        states = 0xf if value else 0x0
+        for index in type(self).AXES:
+            self.resource.write(f"{states:d} {index:d} setmanctrl")

comet/emulator/itk/hydra.py

@@ -21,8 +21,6 @@ def __init__(self):
         self.pos = {"1": 0.0, "2": 0.0}
         self.calibrate = {"1": 3, "2": 3}
 
-        self.cpu_temp = 24.0
-
         self.axes_moving = 0
         self.manual_move = 0
 
@@ -48,7 +46,7 @@ def get_productid(self):
 
     @message(r'^getcputemp$')
     def get_cputemp(self):
-        return self.cpu_temp
+        return float(self.options.get("cputemp", 40.0))
 
     @message(r'^reset$')
     def set_reset(self):

tests/test_driver_itk_hydra.py

@@ -0,0 +1,90 @@
+import pytest
+
+from comet.driver.itk import Hydra
+
+from .test_driver import resource
+
+
+@pytest.fixture
+def driver(resource):
+    return Hydra(resource)
+
+
+def test_hydra(driver, resource):
+    resource.buffer = ["Hydra 1 312 1 10F"]
+    assert driver.identify() == "Hydra 1 312 1 10F"
+    assert resource.buffer == ["identify"]
+
+    resource.buffer = []
+    assert driver.calibrate() is None
+    assert resource.buffer == ["1 ncal", "2 ncal"]
+
+    resource.buffer = []
+    assert driver.range_measure() is None
+    assert resource.buffer == ["1 nrm", "2 nrm"]
+
+    resource.buffer = []
+    assert driver.move_absolute([0, 4.2]) is None
+    assert resource.buffer == ["0.000 4.200 m"]
+
+    resource.buffer = []
+    assert driver.move_relative([0, 2.1]) is None
+    assert resource.buffer == ["0.000 2.100 r"]
+
+    resource.buffer = []
+    assert driver.abort() is None
+    assert resource.buffer == ["1 nabort", "2 nabort"]
+
+    resource.buffer = []
+    assert driver.force_abort() is None
+    assert resource.buffer == ["\x03"]
+
+    resource.buffer = ["2.100 4.200"]
+    assert driver.position == [2.1, 4.2]
+    assert resource.buffer == ["p"]
+
+    resource.buffer = ["3"]
+    assert driver.is_moving
+    assert resource.buffer == ["st"]
+
+    resource.buffer = ["2"]
+    assert not driver.is_moving
+    assert resource.buffer == ["st"]
+
+    resource.buffer = ["1", "0"]
+    assert driver.joystick_enabled
+    assert resource.buffer == ["1 getmanctrl", "2 getmanctrl"]
+
+    resource.buffer = ["0", "0"]
+    assert not driver.joystick_enabled
+    assert resource.buffer == ["1 getmanctrl", "2 getmanctrl"]
+
+    resource.buffer = []
+    driver.joystick_enabled = True
+    assert resource.buffer == ["15 1 setmanctrl", "15 2 setmanctrl"]
+
+
+def test_hydra_axes(driver, resource):
+    resource.buffer = []
+    assert driver[1].calibrate() is None
+    assert resource.buffer == ["1 ncal"]
+
+    resource.buffer = []
+    assert driver[2].range_measure() is None
+    assert resource.buffer == ["2 nrm"]
+
+    resource.buffer = []
+    assert driver[2].move_relative(1.2) is None
+    assert resource.buffer == ["1.200 2 nr"]
+
+    resource.buffer = []
+    assert driver[1].move_absolute(2.2) is None
+    assert resource.buffer == ["2.200 1 nm"]
+
+    resource.buffer = ["4.200"]
+    assert driver[1].position == 4.2
+    assert resource.buffer == ["1 np"]
+
+    resource.buffer = ["2"]
+    assert not driver[1].is_moving
+    assert resource.buffer == ["1 nst"]

tests/test_emulator_itk_hydra.py

@@ -0,0 +1,58 @@
+import pytest
+
+from comet.emulator.itk.hydra import HydraEmulator
+
+
+@pytest.fixture
+def emulator():
+    return HydraEmulator()
+
+
+def test_basic(emulator):
+    assert emulator("identify") == "Hydra 0 0 0 0"
+    assert emulator("getversion") == "1.0"
+    assert emulator("version") == "1.0"
+    assert emulator("getmacadr") == "00:00:00:00:00:00"
+    assert emulator("getserialno") == "01010042"
+    assert emulator("getproductid") == "hydra"
+    assert emulator("getcputemp") == "40.0"
+    assert emulator("reset") is None
+
+
+def test_status(emulator):
+    assert emulator("status") == "24"
+    assert emulator("st") == "24"
+
+    assert emulator("1 nstatus") == "24"
+    assert emulator("1 nst") == "24"
+    assert emulator("1 est") == "24"
+    assert emulator("1 ast") == "24"
+
+    assert emulator("2 nstatus") == "24"
+    assert emulator("2 nst") == "24"
+    assert emulator("2 est") == "24"
+    assert emulator("2 ast") == "24"
+
+
+def test_position(emulator):
+    assert float(emulator("1 np")) == 0.0
+    assert float(emulator("2 np")) == 0.0
+    assert emulator("10 20 m") is None
+    assert float(emulator("1 np")) == 10.0
+    assert float(emulator("2 np")) == 20.0
+    assert emulator("10 -10 r") is None
+    assert float(emulator("1 np")) == 20.0
+    assert float(emulator("2 np")) == 10.0
+    assert emulator("1 nrandmove") is None
+    assert float(emulator("1 np")) != 20.0
+    assert float(emulator("2 np")) == 10.0
+    assert emulator("2 nrandmove") is None
+    assert float(emulator("1 np")) != 20.0
+    assert float(emulator("2 np")) != 10.0
+
+
+def test_calibration(emulator):
+    assert emulator("1 ncalibrate") is None
+    assert emulator("2 ncal") is None
+    assert emulator("1 nrangemeasure") is None
+    assert emulator("2 nrm") is None