diff --git a/pylabrobot/hamilton/protocol/__init__.py b/pylabrobot/hamilton/protocol/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/pylabrobot/hamilton/protocol/text/__init__.py b/pylabrobot/hamilton/protocol/text/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/pylabrobot/hamilton/protocol/text/framing.py b/pylabrobot/hamilton/protocol/text/framing.py new file mode 100644 index 00000000000..0ed915c3d53 --- /dev/null +++ b/pylabrobot/hamilton/protocol/text/framing.py @@ -0,0 +1,170 @@ +"""Hamilton firmware reply parsing. + +A reply repeats the module and command it answers, then carries fields whose names are two +characters followed by their value: `id####` for the identifier a command was sent with, and one +per parameter the machine reports. + +This grammar is shared across Hamilton machines and across transports - the same shape comes back +from a STAR over USB and from a tilt module over a serial line - so nothing here is specific to +either. What a given value *means* belongs with the machine that reports it. +""" + +import datetime +import re + + +def parse_fw_string(resp: str, fmt: str = "") -> dict: + """Parse a machine command or response string according to a format string. + + The format contains names of parameters (always length 2), + followed by an arbitrary number of the following, but always + the same: + - '&': char + - '#': decimal + - '*': hex + + The order of parameters in the format and response string do not + have to (and often do not) match. + + The identifier parameter (id####) is added automatically. + + TODO: string parsing + The firmware docs mention strings in the following format: '...' + However, the length of these is always known (except when reading + barcodes), so it is easier to convert strings to the right number + of '&'. With barcode reading the length of the barcode is included + with the response string. We'll probably do a custom implementation + for that. + + TODO: spaces + We should also parse responses where integers are separated by spaces, + like this: `ua#### #### ###### ###### ###### ######` + + Args: + resp: The response string to parse. + fmt: The format string. + + Raises: + ValueError: if the format string is incompatible with the response. + + Returns: + A dictionary containing the parsed values. + + Examples: + Parsing a string containing decimals (`1111`), hex (`0xB0B`) and chars (`'rw'`): + + ``` + >>> parse_fw_string("aa1111bbrwccB0B", "aa####bb&&cc***") + {'aa': 1111, 'bb': 'rw', 'cc': 2827} + ``` + """ + + # Remove device and cmd identifier from response. + resp = resp[4:] + + # Parse the parameters in the fmt string. + info = {} + + def find_param(param): + name, data = param[0:2], param[2:] + type_ = {"#": "int", "*": "hex", "&": "str"}[data[0]] + + # Build a regex to match this parameter. + exp = { + "int": r"[-+]?[\d ]", + "hex": r"[\da-fA-F ]", + "str": ".", + }[type_] + len_ = len(data.split(" ")[0]) # Get length of first block. + regex = f"{name}((?:{exp}{ {len_} }" + + if param.endswith(" (n)"): + regex += " ?)+)" + is_list = True + else: + regex += "))" + is_list = False + + # Match response against regex, save results in right datatype. + r = re.search(regex, resp) + if r is None: + raise ValueError(f"could not find matches for parameter {name}") + + g = r.groups() + if len(g) == 0: + raise ValueError(f"could not find value for parameter {name}") + m = g[0] + + if is_list: + m = m.split(" ") + + if type_ == "str": + info[name] = m + elif type_ == "int": + info[name] = [int(m_) for m_ in m if m_ != ""] + elif type_ == "hex": + info[name] = [int(m_, base=16) for m_ in m if m_ != ""] + else: + if type_ == "str": + info[name] = m + elif type_ == "int": + info[name] = int(m) + elif type_ == "hex": + info[name] = int(m, base=16) + + # Find params in string. All params are identified by 2 lowercase chars. + param = "" + prevchar = None + for char in fmt: + if char.islower() and prevchar != "(": + if len(param) > 2: + find_param(param) + param = "" + param += char + prevchar = char + if param != "": + find_param(param) # last parameter is not closed by loop. + + # If id not in fmt, add it. + if "id" not in info: + find_param("id####") + + return info + + +def parse_firmware_version_date(fw_version: str) -> datetime.date: + """Extract a date from a firmware version string. + + Supports several common Hamilton firmware version formats: + - Full dates: ``"v2021.03.15"`` or ``"2023_01_05"`` or ``"2020-06-12"`` + - Quarter formats: ``"2023_Q2"`` -> first day of the quarter (2023-04-01) + - Year only: ``"2021"`` -> January 1st of that year + + Args: + fw_version: Firmware version string. + + Returns: + A ``datetime.date`` representing the extracted date. + + Raises: + ValueError: If no year can be parsed from the string. + """ + # Prefer full date patterns like YYYY.MM.DD / YYYY_MM_DD / YYYY-MM-DD + date_match = re.search(r"\b(20\d{2})[._-](\d{2})[._-](\d{2})\b", fw_version) + if date_match: + y, m, d = map(int, date_match.groups()) + return datetime.date(y, m, d) + + # Handle quarter formats like 2023_Q2 -> first day of the quarter + q_match = re.search(r"\b(20\d{2})_Q([1-4])\b", fw_version, flags=re.IGNORECASE) + if q_match: + y = int(q_match.group(1)) + q = int(q_match.group(2)) + month = (q - 1) * 3 + 1 + return datetime.date(y, month, 1) + + # Fall back to year only -> Jan 1st of that year + year_match = re.search(r"\b(20\d{2})\b", fw_version) + if year_match is None: + raise ValueError(f"Could not parse year from firmware version string: '{fw_version}'") + return datetime.date(int(year_match.group(1)), 1, 1) diff --git a/pylabrobot/hamilton/protocol/text/framing_tests.py b/pylabrobot/hamilton/protocol/text/framing_tests.py new file mode 100644 index 00000000000..107d41ccbea --- /dev/null +++ b/pylabrobot/hamilton/protocol/text/framing_tests.py @@ -0,0 +1,58 @@ +import datetime +import unittest + +from pylabrobot.hamilton.protocol.text.framing import ( + parse_firmware_version_date, + parse_fw_string, +) + + +class TestParseFirmwareString(unittest.TestCase): + """The format string names each field and how wide it is; one character per type.""" + + def test_identifier_is_parsed_without_being_named(self): + self.assertEqual(parse_fw_string("C0QMid1111", ""), {"id": 1111}) + self.assertEqual(parse_fw_string("C0QMid1111", "id####"), {"id": 1111}) + + def test_field_types(self): + self.assertEqual(parse_fw_string("C0QMid1112aaabc", "aa&&&"), {"id": 1112, "aa": "abc"}) + self.assertEqual(parse_fw_string("C0QMid1112aa-21", "aa##"), {"id": 1112, "aa": -21}) + self.assertEqual(parse_fw_string("C0QMid1113pqABC", "pq***"), {"id": 1113, "pq": 0xABC}) + + def test_repeated_field_reads_as_a_list(self): + self.assertEqual(parse_fw_string("C0RTid0001rt1 0 1", "rt# (n)"), {"id": 1, "rt": [1, 0, 1]}) + + def test_missing_field_raises(self): + with self.assertRaises(ValueError): + parse_fw_string("C0QMid1111", "aa####") + + +class TestParseFirmwareVersionDate(unittest.TestCase): + """Two layouts are in circulation, and the module reports which one it uses by how it writes + the date rather than by saying so.""" + + def test_both_layouts(self): + self.assertEqual( + parse_firmware_version_date("C0RFid0001rf7.6S 2021-11-05"), datetime.date(2021, 11, 5) + ) + self.assertEqual( + parse_firmware_version_date("H0RFid0001rf5.0S i 2021-10-22 (H0 XE167)"), + datetime.date(2021, 10, 22), + ) + + +class TestOldNamesStillImport(unittest.TestCase): + """The functions moved out of the USB transport; the names they were reached by still work.""" + + def test_transport_and_backend_re_export_the_same_object(self): + from pylabrobot.hamilton.transport.usb.protocol import ( + parse_star_firmware_version_date, + parse_star_fw_string, + ) + from pylabrobot.legacy.liquid_handling.backends.hamilton.STAR_backend import ( + parse_star_fw_string as from_backend, + ) + + self.assertIs(parse_star_fw_string, parse_fw_string) + self.assertIs(from_backend, parse_fw_string) + self.assertIs(parse_star_firmware_version_date, parse_firmware_version_date) diff --git a/pylabrobot/hamilton/transport/usb/protocol.py b/pylabrobot/hamilton/transport/usb/protocol.py index 889156421f2..9f6127d71de 100644 --- a/pylabrobot/hamilton/transport/usb/protocol.py +++ b/pylabrobot/hamilton/transport/usb/protocol.py @@ -1,161 +1,15 @@ -"""STAR firmware response parsing utilities.""" +"""STAR firmware response parsing utilities. -import datetime -import re +Moved to `pylabrobot.hamilton.protocol.text.framing`, which holds the wire format on its own rather +than inside the USB transport that happens to carry it. Re-exported here under the old names so +existing imports keep working. +""" +from pylabrobot.hamilton.protocol.text.framing import ( + parse_firmware_version_date as parse_star_firmware_version_date, +) +from pylabrobot.hamilton.protocol.text.framing import ( + parse_fw_string as parse_star_fw_string, +) -def parse_star_fw_string(resp: str, fmt: str = "") -> dict: - """Parse a machine command or response string according to a format string. - - The format contains names of parameters (always length 2), - followed by an arbitrary number of the following, but always - the same: - - '&': char - - '#': decimal - - '*': hex - - The order of parameters in the format and response string do not - have to (and often do not) match. - - The identifier parameter (id####) is added automatically. - - TODO: string parsing - The firmware docs mention strings in the following format: '...' - However, the length of these is always known (except when reading - barcodes), so it is easier to convert strings to the right number - of '&'. With barcode reading the length of the barcode is included - with the response string. We'll probably do a custom implementation - for that. - - TODO: spaces - We should also parse responses where integers are separated by spaces, - like this: `ua#### #### ###### ###### ###### ######` - - Args: - resp: The response string to parse. - fmt: The format string. - - Raises: - ValueError: if the format string is incompatible with the response. - - Returns: - A dictionary containing the parsed values. - - Examples: - Parsing a string containing decimals (`1111`), hex (`0xB0B`) and chars (`'rw'`): - - ``` - >>> parse_fw_string("aa1111bbrwccB0B", "aa####bb&&cc***") - {'aa': 1111, 'bb': 'rw', 'cc': 2827} - ``` - """ - - # Remove device and cmd identifier from response. - resp = resp[4:] - - # Parse the parameters in the fmt string. - info = {} - - def find_param(param): - name, data = param[0:2], param[2:] - type_ = {"#": "int", "*": "hex", "&": "str"}[data[0]] - - # Build a regex to match this parameter. - exp = { - "int": r"[-+]?[\d ]", - "hex": r"[\da-fA-F ]", - "str": ".", - }[type_] - len_ = len(data.split(" ")[0]) # Get length of first block. - regex = f"{name}((?:{exp}{ {len_} }" - - if param.endswith(" (n)"): - regex += " ?)+)" - is_list = True - else: - regex += "))" - is_list = False - - # Match response against regex, save results in right datatype. - r = re.search(regex, resp) - if r is None: - raise ValueError(f"could not find matches for parameter {name}") - - g = r.groups() - if len(g) == 0: - raise ValueError(f"could not find value for parameter {name}") - m = g[0] - - if is_list: - m = m.split(" ") - - if type_ == "str": - info[name] = m - elif type_ == "int": - info[name] = [int(m_) for m_ in m if m_ != ""] - elif type_ == "hex": - info[name] = [int(m_, base=16) for m_ in m if m_ != ""] - else: - if type_ == "str": - info[name] = m - elif type_ == "int": - info[name] = int(m) - elif type_ == "hex": - info[name] = int(m, base=16) - - # Find params in string. All params are identified by 2 lowercase chars. - param = "" - prevchar = None - for char in fmt: - if char.islower() and prevchar != "(": - if len(param) > 2: - find_param(param) - param = "" - param += char - prevchar = char - if param != "": - find_param(param) # last parameter is not closed by loop. - - # If id not in fmt, add it. - if "id" not in info: - find_param("id####") - - return info - - -def parse_star_firmware_version_date(fw_version: str) -> datetime.date: - """Extract a date from a firmware version string. - - Supports several common Hamilton firmware version formats: - - Full dates: ``"v2021.03.15"`` or ``"2023_01_05"`` or ``"2020-06-12"`` - - Quarter formats: ``"2023_Q2"`` -> first day of the quarter (2023-04-01) - - Year only: ``"2021"`` -> January 1st of that year - - Args: - fw_version: Firmware version string. - - Returns: - A ``datetime.date`` representing the extracted date. - - Raises: - ValueError: If no year can be parsed from the string. - """ - # Prefer full date patterns like YYYY.MM.DD / YYYY_MM_DD / YYYY-MM-DD - date_match = re.search(r"\b(20\d{2})[._-](\d{2})[._-](\d{2})\b", fw_version) - if date_match: - y, m, d = map(int, date_match.groups()) - return datetime.date(y, m, d) - - # Handle quarter formats like 2023_Q2 -> first day of the quarter - q_match = re.search(r"\b(20\d{2})_Q([1-4])\b", fw_version, flags=re.IGNORECASE) - if q_match: - y = int(q_match.group(1)) - q = int(q_match.group(2)) - month = (q - 1) * 3 + 1 - return datetime.date(y, month, 1) - - # Fall back to year only -> Jan 1st of that year - year_match = re.search(r"\b(20\d{2})\b", fw_version) - if year_match is None: - raise ValueError(f"Could not parse year from firmware version string: '{fw_version}'") - return datetime.date(int(year_match.group(1)), 1, 1) +__all__ = ["parse_star_fw_string", "parse_star_firmware_version_date"] diff --git a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_backend.py index f57ec4b3582..bf8e7c6d432 100644 --- a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_backend.py +++ b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_backend.py @@ -34,6 +34,9 @@ from typing import Concatenate, ParamSpec from pylabrobot import audio +from pylabrobot.hamilton.protocol.text.framing import ( + parse_fw_string as parse_star_fw_string, +) from pylabrobot.legacy.arms.standard import CartesianCoords from pylabrobot.legacy.heating_shaking.hamilton_backend import HamiltonHeaterShakerInterface from pylabrobot.legacy.liquid_handling.backends.hamilton.base import ( @@ -152,125 +155,6 @@ async def wrapper(self: "STARBackend", *args, **kwargs): return wrapper -def parse_star_fw_string(resp: str, fmt: str = "") -> dict: - """Parse a machine command or response string according to a format string. - - The format contains names of parameters (always length 2), - followed by an arbitrary number of the following, but always - the same: - - '&': char - - '#': decimal - - '*': hex - - The order of parameters in the format and response string do not - have to (and often do not) match. - - The identifier parameter (id####) is added automatically. - - TODO: string parsing - The firmware docs mention strings in the following format: '...' - However, the length of these is always known (except when reading - barcodes), so it is easier to convert strings to the right number - of '&'. With barcode reading the length of the barcode is included - with the response string. We'll probably do a custom implementation - for that. - - TODO: spaces - We should also parse responses where integers are separated by spaces, - like this: `ua#### #### ###### ###### ###### ######` - - Args: - resp: The response string to parse. - fmt: The format string. - - Raises: - ValueError: if the format string is incompatible with the response. - - Returns: - A dictionary containing the parsed values. - - Examples: - Parsing a string containing decimals (`1111`), hex (`0xB0B`) and chars (`'rw'`): - - ``` - >>> parse_fw_string("aa1111bbrwccB0B", "aa####bb&&cc***") - {'aa': 1111, 'bb': 'rw', 'cc': 2827} - ``` - """ - - # Remove device and cmd identifier from response. - resp = resp[4:] - - # Parse the parameters in the fmt string. - info = {} - - def find_param(param): - name, data = param[0:2], param[2:] - type_ = {"#": "int", "*": "hex", "&": "str"}[data[0]] - - # Build a regex to match this parameter. - exp = { - "int": r"[-+]?[\d ]", - "hex": r"[\da-fA-F ]", - "str": ".", - }[type_] - len_ = len(data.split(" ")[0]) # Get length of first block. - regex = f"{name}((?:{exp}{ {len_} }" - - if param.endswith(" (n)"): - regex += " ?)+)" - is_list = True - else: - regex += "))" - is_list = False - - # Match response against regex, save results in right datatype. - r = re.search(regex, resp) - if r is None: - raise ValueError(f"could not find matches for parameter {name}") - - g = r.groups() - if len(g) == 0: - raise ValueError(f"could not find value for parameter {name}") - m = g[0] - - if is_list: - m = m.split(" ") - - if type_ == "str": - info[name] = m - elif type_ == "int": - info[name] = [int(m_) for m_ in m if m_ != ""] - elif type_ == "hex": - info[name] = [int(m_, base=16) for m_ in m if m_ != ""] - else: - if type_ == "str": - info[name] = m - elif type_ == "int": - info[name] = int(m) - elif type_ == "hex": - info[name] = int(m, base=16) - - # Find params in string. All params are identified by 2 lowercase chars. - param = "" - prevchar = None - for char in fmt: - if char.islower() and prevchar != "(": - if len(param) > 2: - find_param(param) - param = "" - param += char - prevchar = char - if param != "": - find_param(param) # last parameter is not closed by loop. - - # If id not in fmt, add it. - if "id" not in info: - find_param("id####") - - return info - - class STARModuleError(Exception, metaclass=ABCMeta): """Base class for all Hamilton backend errors, raised by a single module."""