Merge pull request #4 from harp-tech/gl-dev

Add dynamic schema-based register reader classes

Author: glopesdev

harp/io.py

@@ -1,10 +1,10 @@
 from os import PathLike
-from typing import Any, Optional, Union
+from typing import Any, BinaryIO, Optional, Union
 from pandas._typing import Axes
 import numpy as np
 import pandas as pd
 
-_SECONDS_PER_TICK = 32e6
+_SECONDS_PER_TICK = 32e-6
 payloadtypes = {
     1: np.dtype(np.uint8),
     2: np.dtype(np.uint16),
@@ -19,7 +19,7 @@
 
 
 def read(
-    file: Union[str, bytes, PathLike[Any], np._IOProtocol],
+    file: Union[str, bytes, PathLike[Any], BinaryIO],
     columns: Optional[Axes] = None,
 ):
     """

harp/reader.py

@@ -0,0 +1,109 @@
+import re
+from functools import partial
+from pandas import DataFrame, Series
+from typing import Iterable, Callable, Union
+from harp.model import BitMask, GroupMask, MaskValueItem, Model, Register
+from harp.io import read
+
+_camel_to_snake_regex = re.compile(r"(?<!^)(?=[A-Z])")
+
+
+class RegisterReader:
+    register: Register
+    read: Callable[[str], DataFrame]
+
+    def __init__(self, register: Register, read: Callable[[str], DataFrame]) -> None:
+        self.register = register
+        self.read = read
+
+
+class DeviceReader:
+    device: Model
+    registers: dict[str, RegisterReader]
+
+    def __init__(self, device: Model, registers: dict[str, RegisterReader]) -> None:
+        self.device = device
+        self.registers = registers
+
+    def __dir__(self) -> Iterable[str]:
+        return self.registers.keys()
+
+    def __getattr__(self, __name: str) -> RegisterReader:
+        return self.registers[__name]
+
+
+def _compose(f, g):
+    return lambda *a, **kw: f(g(*a, **kw))
+
+
+def _id_camel_to_snake(id: str):
+    return _camel_to_snake_regex.sub("_", id).lower()
+
+
+def _keys_camel_to_snake(keys: Iterable[str]):
+    return [_id_camel_to_snake(k) for k in keys]
+
+
+def _create_bit_parser(mask: Union[int, MaskValueItem]):
+    def parser(xs: Series) -> Series:
+        return (xs & mask) != 0
+
+    return parser
+
+
+def _create_bitmask_parser(bitMask: BitMask):
+    lookup = [
+        (_id_camel_to_snake(k), _create_bit_parser(v.root))
+        for k, v in bitMask.bits.items()
+    ]
+
+    def parser(df: DataFrame):
+        return DataFrame({n: f(df[0]) for n, f in lookup}, index=df.index)
+
+    return parser
+
+
+def _create_groupmask_parser(name: str, groupMask: GroupMask):
+    name = _id_camel_to_snake(name)
+    lookup = {v.root: n for n, v in groupMask.values.items()}
+
+    def parser(df: DataFrame):
+        return DataFrame({name: df.map(lambda x: lookup[x])})
+
+    return parser
+
+
+def _create_register_reader(device: Model, name: str):
+    register = device.registers[name]
+    reader = read
+
+    if register.maskType is not None:
+        key = register.maskType.root
+        bitMask = device.bitMasks.get(key)
+        if bitMask is not None:
+            parser = _create_bitmask_parser(bitMask)
+            reader = _compose(parser, reader)
+            return RegisterReader(register, reader)
+
+        groupMask = device.groupMasks.get(key)
+        if groupMask is not None:
+            parser = _create_groupmask_parser(name, groupMask)
+            reader = _compose(parser, reader)
+            return RegisterReader(register, reader)
+
+    if register.payloadSpec is not None:
+        columns = register.payloadSpec.keys()
+        columns = _keys_camel_to_snake(columns)
+        reader = partial(reader, columns=columns)
+        return RegisterReader(register, reader)
+
+    columns = [_id_camel_to_snake(name)]
+    reader = partial(reader, columns=columns)
+    return RegisterReader(register, reader)
+
+
+def create_reader(device: Model):
+    reg_readers = {
+        name: _create_register_reader(device, name) for name in device.registers.keys()
+    }
+    return DeviceReader(device, reg_readers)

pyproject.toml

@@ -31,15 +31,17 @@ classifiers = [
 [project.optional-dependencies]
 dev = [
     "datamodel-code-generator",
-    "setuptools_scm",
     "black"
 ]
+jupyter = [
+    "ipykernel"
+]
 
 [build-system]
 requires = [
-    "setuptools>=45",
     "wheel",
-    "setuptools_scm[toml]>=6.2",
+    "setuptools",
+    "setuptools_scm[toml]",
 ]
 build-backend = "setuptools.build_meta"
 
@@ -62,5 +64,6 @@ extend-exclude = '''
   ^/LICENSE
   ^/README.md
   | harp/model.py
+  | reflex-generator
 )
 '''