Support iseg power supplies (#17)

* conclicts fixed * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * created devices/caen module * fix import * core classes: CAEN, ISEG * some methods of class Channel setters are not labeled as such yet * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * add function to list serial ports * add test for module * add test for channel * working alarm parsing * fix parse response when not returning parameter * fix bad bit array parsing * correctly parse alarm/stat * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * do not run tests if no serial device exists * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * decorator to skip test * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * attempt to fix skip * use correct annotations * use correct annotations * rest of channel commands and a few module commands * iseg tests * fix conflicts * implement more iseg * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * rest of module methods * add mypy to pre-commit * removed replicated code, added setter's decorators and missing type hints Some methods in iseg.devices.channel were duplicated * use ruff instead of mypy * all iseg methods * test_utils updated * comments indicating the devices where some methods only work + some fixes * annotations imported from future * annotations imported from future, again * improved imports * add "--show-fixes" to ruff * reduce code duplication * add .ruff_cache/ to .gitignore * add .ruff_cache/ to .gitignore * some fixes and almost all tests * rest of iseg tests * fixed errors due to automatic merge * Fix imports * fix test_caen_commands * fixed test_caen_commands --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Luis Antonio Obis Aparicio <[email protected]> Co-authored-by: Luis Antonio Obis Aparicio <[email protected]>
lobis · Jun 29, 2023 · f6e7e0f · f6e7e0f
1 parent eab3d16
commit f6e7e0f
Show file tree

Hide file tree

Showing 24 changed files with 2,655 additions and 339 deletions.
diff --git a/.gitignore b/.gitignore
@@ -132,3 +132,6 @@ dmypy.json
 .vscode/
 
 node_modules/
+
+# ruff
+.ruff_cache/
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -18,3 +18,9 @@ repos:
     rev: 23.3.0
     hooks:
       - id: black
+
+  - repo: https://github.com/charliermarsh/ruff-pre-commit
+    rev: v0.0.275
+    hooks:
+      - id: ruff
+        args: [ "--fix-only" , "--show-fixes" ]
diff --git a/src/__init__.py b/src/__init__.py
diff --git a/src/hvps/__init__.py b/src/hvps/__init__.py
@@ -1,2 +1,5 @@
-from .devices import Caen, Iseg
-from .version import __version__
+from hvps.devices.caen.caen import Caen
+from hvps.devices.iseg.iseg import Iseg
+from .version import __version__
+
+__all__ = ["Caen", "Iseg", "__version__"]
diff --git a/src/hvps/__main__.py b/src/hvps/__main__.py
@@ -1,4 +1,4 @@
-from hvps import __version__ as hvps_version, Caen, Iseg
+from hvps import __version__ as hvps_version, Caen
 from hvps.commands.caen.module import _set_module_commands, _mon_module_commands
 from hvps.commands.caen.channel import _set_channel_commands, _mon_channel_commands
 
@@ -55,7 +55,7 @@ def main():
     )
 
     # iseg
-    iseg_parser = subparsers.add_parser("iseg", help="iseg HVPS")
+    subparsers.add_parser("iseg", help="iseg HVPS")
 
     args = parser.parse_args()
     logging.basicConfig(level=args.log.upper())

diff --git a/src/hvps/commands/caen/__init__.py b/src/hvps/commands/caen/__init__.py
@@ -1,15 +1,6 @@
 from __future__ import annotations
 import re
 
-from .module import (
-    _get_mon_module_command,
-    _get_set_module_command,
-)
-from .channel import (
-    _get_mon_channel_command,
-    _get_set_channel_command,
-)
-
 import logging
 import serial
 
@@ -55,7 +46,8 @@ def _parse_response(response: bytes) -> (int, str):
     logger.debug(f"Response: {response}")
     if response == b"":
         raise ValueError(
-            "Empty response. There was no response from the device. Check that the device is connected and correctly configured (baudrate)."
+            "Empty response. There was no response from the device. Check that the device is connected and correctly "
+            "configured (baudrate)."
         )
 
     try:

diff --git a/src/hvps/commands/iseg/__init__.py b/src/hvps/commands/iseg/__init__.py
@@ -1,27 +1,18 @@
 from __future__ import annotations
-import re
 from typing import List
 
-"""
-from .module import (
-    _get_mon_module_command,
-    _get_set_module_command,
-)
-"""
-
-from .channel import (
-    _get_mon_channel_command,
-    _get_set_channel_command,
-)
-
 import logging
 import serial
+import re
 
 
 def _write_command(
-    ser: serial.Serial, command: bytes, response: bool = True
+    ser: serial.Serial,
+    command: bytes,
+    expected_response_type: type | None,
+    response: bool = True,
 ) -> List[str] | None:
-    """Write a command to a device.
+    """Write a command to a device and parses the response.
 
     Args:
         ser (Serial): The serial connection to the device.
@@ -41,12 +32,12 @@ def _write_command(
             f"Invalid handshake echo response: {response_value}. expected {command}"
         )
     # response reading
-    response_value = _parse_response(ser.readline())
+    response_value = _parse_response(ser.readline(), expected_response_type)
 
     return response_value
 
 
-def _parse_response(response: bytes) -> List[str]:
+def _parse_response(response: bytes, expected_response_type: type | None) -> List[str]:
     """Parse the response from a device.
 
     Args:
@@ -66,4 +57,29 @@ def _parse_response(response: bytes) -> List[str]:
     except UnicodeDecodeError:
         raise ValueError(f"Invalid response: {response}")
 
-    return response.split(",")
+    if expected_response_type == float or expected_response_type == List[float]:
+        # pattern for a float in scientific notation followed or not by units
+        pattern = (
+            r"^[-+]?[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?(\s*[a-zA-Z]+(/+[a-zA-Z]+)?)?$"
+        )
+    elif expected_response_type == int or expected_response_type == List[int]:
+        pattern = r"^[-+]?\d+$"
+    elif expected_response_type == str or expected_response_type == List[str]:
+        pattern = r"^[\x00-\x7F]+$"
+    elif expected_response_type is None:
+        return response.split(",")
+    else:
+        raise ValueError(
+            f"expected value type of {response}, {expected_response_type}, is not float, int or str"
+        )
+
+    split_response = response.split(",")
+
+    for r in split_response:
+        match = re.match(pattern, r)
+        if not match:
+            raise ValueError(
+                f"Invalid response: {response}, can't be identified as a {expected_response_type}, missmatch in RE"
+            )
+
+    return split_response
diff --git a/src/hvps/commands/iseg/channel.py b/src/hvps/commands/iseg/channel.py
@@ -1,24 +1,101 @@
 from __future__ import annotations
 
+_mon_channel_commands = {
+    ":CONF:TRIP:ACTION": "Query the current action to be taken when a current trip occurs for the channel.",
+    ":CONF:TRIP:TIME": "Query the current action to be taken when a current trip occurs for the channel.",
+    ":CONF:INHP:ACTION": "Query the current action to be taken when a current trip occurs for the channel.",
+    ":CONF:OUTPUT:MODE": "Query the configured channel output mode.",
+    ":CONF:OUTPUT:MODE:LIST": "Query the available channel output modes as a list.",
+    ":CONF:OUTPUT:POL": "Query the current output polarity of the channel.",
+    ":CONF:OUTPUT:POL:LIST": "Query the available channel output polarities as a list.",
+    ":READ:VOLT": "Query the voltage set Vset in Volt.",
+    ":READ:VOLT:LIM": "Query the voltage limit Vlim in Volt.",
+    ":READ:VOLT:NOM": "Query the channel voltage nominal Vnom in Volt.",
+    ":READ:VOLT:ON": "Query the channel control bit 'Set On'.",
+    ":READ:VOLT:EMCY": "Query the channel control bit 'Set Emergency Off'.",
+    ":READ:CURR": "Query the current set (Iset) in Ampere.",
+    ":READ:CURR:LIM": "Query the current limit (Ilim) in Ampere.",
+    ":READ:CURR:NOM": "Query the current nominal (Inom) in Ampere.",
+    ":READ:CURR:MODE": "Query the configured channel current mode in Ampere.",
+    ":READ:CURR:MODE:LIST": "Query the available channel current modes as a list.",
+    ":READ:CURR:BOUNDS": "Query the channel current bounds in Ampere.",
+    ":READ:RAMP:CURR": "Query the channel current ramp speed in Ampere/second.",
+    ":READ:RAMP:VOLT": "Query the channel voltage ramp speed in Volt/second.",
+    ":READ:RAMP:VOLT:MIN": "Query the channel voltage ramp speed minimum in Volt/second.",
+    ":READ:RAMP:VOLT:MAX": "Query the channel voltage ramp speed maximum in Volt/second.",
+    ":READ:RAMP:CURR:MIN": "Query the channel current ramp speed minimum in Ampere/second.",
+    ":READ:RAMP:CURR:MAX": "Query the channel current ramp speed maximum in Ampere/second.",
+    ":READ:CHAN:CONTROL": "Query the Channel Control register.",
+    ":READ:CHAN:STATUS": "Query the Channel Status register.",
+    ":READ:CHAN:EVENT:MASK": "Query the Channel Event Mask register.",
+    ":MEAS:VOLT": "Query the measured channel voltage in Volt.",
+    ":MEAS:CURR": "Query the measured channel current in Ampere.",
+    ":CONF:RAMP:VOLT:UP": "Query the channel voltage ramp up speed in Volt/second.",
+    ":CONF:RAMP:VOLT:DOWN": "Query the channel voltage ramp down speed in Volt/second.",
+    ":CONF:RAMP:CURR:UP": "Query the channel current ramp up speed in Ampere/second.",
+    ":CONF:RAMP:CURR:DOWN": "Query the channel current ramp down speed in Ampere/second.",
+}
+
+_set_channel_commands = {
+    ":CONF:TRIP:ACTION": "Set the action to be taken when a current trip occurs for the channel.",
+    ":CONF:TRIP:TIME": "Set the trip timeout with one millisecond resolution.",
+    ":CONF:INHP:ACTION": "Set the action to be taken when an External Inhibit event occurs for the channel.",
+    ":CONF:OUTPUT:MODE": "Set the channel output mode.",
+    ":CONF:OUTPUT:POL": "Set the output polarity of the channel.",
+    ":VOLT": "Set the channel voltage set.",
+    ":VOLT:BOUNDS": "Set the channel voltage bounds.",
+    ":VOLT EMCY": "Clear the channel from state emergency off.",
+    ":CURR": "Set the channel current set.",
+    ":CURR:BOUNDS": "Set the channel current bounds in Ampere.",
+    ":CONF:RAMP:VOLT": "Set the channel voltage ramp speed for up and down direction in Volt/second.",
+    ":CONF:RAMP:VOLT:UP": "Set the channel voltage ramp up speed in Volt/second.",
+    ":CONF:RAMP:VOLT:DOWN": "Set the channel voltage ramp down speed in Volt/second.",
+    ":CONF:RAMP:CURR": "Set the channel current ramp speed for up and down direction in Ampere/second.",
+    ":CONF:RAMP:CURR:UP": "Set the channel current ramp up speed in Ampere/second.",
+    ":CONF:RAMP:CURR:DOWN": "Set the channel current ramp down speed in Ampere/second.",
+    ":VOLT ON": "Switch on the high voltage with the configured ramp speed.",
+    ":VOLT OFF": "Switch off the high voltage with the configured ramp speed.",
+    ":VOLT EMCY OFF": "Shut down the channel high voltage (without ramp) or clear the channel from state emergency off.",
+    ":VOLT EMCY CLR": "Clear the channel from state emergency off. The channel goes to state off.",
+    ":EVENT CLEAR": "Clear the Channel Event Status register.",
+    ":EVENT": "Clears single bits or bit combinations in the Channel Event Status register by writing a one to the corresponding bit position.",
+    ":EVENT:MASK": "Clears single bits or bit combinations in the Channel Event Status register by writing a one to the corresponding bit position.",
+}
+
 
 def _get_mon_channel_command(channel: int, command: str) -> bytes:
     """
     Generates a query command string for monitoring a specific channel.
 
     Args:
-        command (str): The base command without the query symbol.
         channel (int): The channel number.
+        command (str): The base command without the query symbol.
 
     Returns:
-        str: The query command string.
+        bytes: The query command string as bytes.
+
+    Raises:
+        ValueError: If the provided channel is not a valid positive integer.
+        ValueError: If the provided command is not a valid command.
 
     Example:
         command = ":MEAS:CURR"
         channel = 3
-        query_command = _get_mon_channel_command(command, channel)
+        query_command = _get_mon_channel_command(channel, command)
         print(query_command)
-        :MEAS:CURR? (@3)
+        b':MEAS:CURR? (@3)\r\n'
     """
+    if channel < 0:
+        raise ValueError(
+            f"Invalid channel '{channel}'. Valid channels are positive integers."
+        )
+    command = command.upper()
+    if command not in _mon_channel_commands:
+        valid_commands = ", ".join(_mon_channel_commands.keys())
+        raise ValueError(
+            f"Invalid command '{command}'. Valid commands are: {valid_commands}."
+        )
+
     return f"{command.strip()}? (@{channel})\r\n".encode("ascii")
 
 
@@ -36,12 +113,30 @@ def _get_set_channel_command(
     Returns:
         bytes: The order command as a bytes object.
 
+    Raises:
+        ValueError: If the provided channel is not a valid positive integer.
+        ValueError: If the provided command is not a valid command.
+
     Example:
         channel = 4
         command = ":VOLT"
         value = 200
         order_command = _get_set_channel_command(channel, command, value)
         print(order_command)
-        :VOLT 200,(@4)\r\n'
+        b':VOLT 200,(@4);*OPC?\r\n'
     """
+    if channel < 0:
+        raise ValueError(
+            f"Invalid channel '{channel}'. Valid channels are positive integers."
+        )
+    command = command.upper()
+    if command not in _set_channel_commands:
+        valid_commands = ", ".join(_mon_channel_commands.keys())
+        raise ValueError(
+            f"Invalid command '{command}'. Valid commands are: {valid_commands}."
+        )
+
+    if isinstance(value, float):
+        value = f"{value:E}"
+
     return f"{command.strip()} {value},(@{channel});*OPC?\r\n".encode("ascii")
diff --git a/src/hvps/commands/iseg/module.py b/src/hvps/commands/iseg/module.py
@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+_mon_module_commands = {
+    ":READ:MODULE:CHANNELNUMBER": "The number of channels in the module.",
+    ":READ:FIRMWARE:RELEASE": "Read out Firmware Release (XX.X)",
+    ":READ:MODULE:STATUS": "Read out module status register",
+    ":CONF:AVER": "Query the digital filter averaging steps.",
+    ":CONF:KILL": "Get the current value for the kill enable function.",
+    ":CONF:ADJUST": "Get the fine adjustment state.",
+    ":CONF:CAN:ADDR": "Query the module's CAN bus address.",
+    ":CONF:CAN:BITRATE": "Query the module's CAN bus bit rate.",
+    ":CONF:SERIAL:BAUD": "Query the device's serial baud rate.",
+    ":CONF:SERIAL:ECHO": "Check if serial echo is enabled or disabled.",
+    ":READ:VOLT:LIM": "Query the module's voltage limit in percent.",
+    ":READ:CURR:LIM": "Query the module's current limit in percent.",
+    ":READ:RAMP:VOLT": "Query the module's voltage ramp speed in percent/second.",
+    ":READ:RAMP:CURR": "Query the module's current ramp speed in percent/second.",
+    ":READ:MODULE:CONTROL": "Query the Module Control register",
+    ":READ:MODULE:EVENT:STATUS": "Query the Module Event Status register",
+    ":READ:MODULE:EVENT:MASK": "Query the Module Event Mask register",
+    ":READ:MODULE:EVENT:CHANSTAT": "Query the Module Event Channel Status register",
+    ":READ:MODULE:EVENT:CHANMASK": "Query the Module Event Channel Mask register",
+    ":READ:MODULE:SUPPLY? (@0-6)": "Query the module supply voltages",
+    ":READ:MODULE:SUPPLY:P24V": "Query the module supply voltage +24 Volt",
+    ":READ:MODULE:SUPPLY:N24V": "Query the module supply voltage -24 Volt",
+    ":READ:MODULE:SUPPLY:P5V": "Query the module supply voltage +5 Volt",
+    ":READ:MODULE:SUPPLY:P3V": "Query the module internal supply voltage +3.3 Volt",
+    ":READ:MODULE:SUPPLY:P12V": "Query the module internal supply voltage +12 Volt",
+    ":READ:MODULE:SUPPLY:N12V": "Query the module internal supply voltage -12 Volt",
+    ":READ:MODULE:TEMPERATURE": "Query the module temperature in degree Celsius",
+    ":READ:MODULE:SETVALUE": "Query the setvalue changes counter",
+    ":READ:FIRMWARE:NAME": "Query the module's firmware name",
+    ":CONF:EVENT:MASK": "Query the Module Event Mask register",
+    ":CONF:EVENT:CHANMASK": "Query the Module Event Channel Mask register",
+}
+
+_set_module_commands = {
+    ":CONF:AVER": "Query the digital filter averaging steps.",
+    ":CONF:KILL": "Get the current value for the kill enable function.",
+    ":CONF:ADJUST": "Get the fine adjustment state.",
+    ":CONF:CAN:ADDR": "Query the module's CAN bus address.",
+    ":CONF:CAN:BITRATE": "Query the module's CAN bus bit rate.",
+    ":CONF:SERIAL:BAUD": "Query the device's serial baud rate.",
+    ":CONF:SERIAL:ECHO": "Check if serial echo is enabled or disabled.",
+    ":CONF:EVENT:MASK": "Set the Module Event Mask register",
+    ":CONF:EVENT:CHANMASK": "Set the Module Event Channel Mask register",
+    ":SYSTEM:USER:CONFIG": "Set the device to configuration mode to change the CAN bitrate or address",
+    ":CONF:EVENT CLEAR": "Reset the Module Event Status register",
+    ":CONF:EVENT": "Clear single bits or bit combinations in the Module Event Status register",
+}
+
+
+def _get_mon_module_command(command: str) -> bytes:
+    """
+    Generates a query command string for monitoring a specific module.
+
+    Args:
+        command (str): The base command without the query symbol.
+
+    Returns:
+        bytes: The query command string as bytes.
+
+    Raises:
+        ValueError: If the provided command is not a valid command.
+
+    Example:
+        command = ":MEAS:CURR"
+        query_command = _get_mon_module_command(command)
+        print(query_command)
+        b':MEAS:CURR?\r\n'
+    """
+    command = command.upper()
+    if command not in _mon_module_commands:
+        valid_commands = ", ".join(_mon_module_commands.keys())
+        raise ValueError(
+            f"Invalid command '{command}'. Valid commands are: {valid_commands}."
+        )
+
+    return f"{command.strip()}?\r\n".encode("ascii")
+
+
+def _get_set_module_command(command: str, value: str | int | float | None) -> bytes:
+    """
+    Generates an order command as a bytes object to set a value for a specific module.
+
+    Args:
+        command (str): The base command without the value and channel suffix.
+        value (str | int | float | None): The value to be set. Can be a string, integer, float, or None.
+
+    Returns:
+        bytes: The order command as a bytes object.
+
+    Raises:
+        ValueError: If the provided command is not a valid command.
+
+    Example:
+        command = ":VOLT"
+        value = 200
+        order_command = _get_set_module_command(command, value)
+        print(order_command)
+        b':VOLT 200;*OPC?\r\n'
+    """
+    command = command.upper()
+    if command not in _set_module_commands:
+        valid_commands = ", ".join(_set_module_commands.keys())
+        raise ValueError(
+            f"Invalid command '{command}'. Valid commands are: {valid_commands}."
+        )
+    if isinstance(value, float):
+        value = f"{value:E}"
+
+    return f"{command.strip()} {value};*OPC?\r\n".encode("ascii")
diff --git a/src/hvps/devices/__init__.py b/src/hvps/devices/__init__.py
@@ -1,2 +0,0 @@
-from .caen import Caen
-from .iseg import Iseg