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

Read common register schema using importlib

Author: glopesdev

harp/__init__.py

@@ -84,7 +84,7 @@ def read(
         time = micros * _SECONDS_PER_TICK + seconds
         payloadtype = payloadtype & ~0x10
         if epoch is not None:
-            time = epoch + pd.to_timedelta(time, "s") # type: ignore
+            time = epoch + pd.to_timedelta(time, "s")  # type: ignore
         index = pd.Series(time)
         index.name = "time"
 
@@ -111,6 +111,6 @@ def read(
         msgtype = np.ndarray(
             nrows, dtype=np.uint8, buffer=data, offset=0, strides=stride
         )
-        msgtype = pd.Categorical.from_codes(msgtype, categories=_messagetypes) # type: ignore
+        msgtype = pd.Categorical.from_codes(msgtype, categories=_messagetypes)  # type: ignore
         result["type"] = msgtype
     return result

harp/io.py

@@ -1,33 +1,29 @@
 from os import PathLike
-from pathlib import Path
 from typing import TextIO, Union
 from harp.model import Model, Registers
 from pydantic_yaml import parse_yaml_raw_as
+from importlib import resources
 
-_common_yaml_path = Path(__file__).absolute().parent.joinpath("common.yml")
 
-
-def _read_common_registers(file: Union[str, PathLike, TextIO]) -> Registers:
-    if isinstance(file, TextIO):
-        return parse_yaml_raw_as(Registers, file.read())
-    else:
-        with open(file) as fileIO:
-            return _read_common_registers(fileIO)
+def _read_common_registers() -> Registers:
+    file = resources.files(__package__) / "common.yml"
+    with file.open("rt") as fileIO:
+        return parse_yaml_raw_as(Registers, fileIO.read())
 
 
 def read_schema(
     file: Union[str, PathLike, TextIO], include_common_registers: bool = True
 ) -> Model:
-    if isinstance(file, TextIO):
+    if isinstance(file, (str, PathLike)):
+        with open(file) as fileIO:
+            return read_schema(fileIO)
+    else:
         schema = parse_yaml_raw_as(Model, file.read())
         if not "WhoAmI" in schema.registers and include_common_registers:
-            common = _read_common_registers(_common_yaml_path)
+            common = _read_common_registers()
             schema.registers = dict(common.registers, **schema.registers)
             if schema.bitMasks is not None and common.bitMasks is not None:
                 schema.bitMasks = dict(common.bitMasks, **schema.bitMasks)
             if schema.groupMasks is not None and common.groupMasks is not None:
                 schema.groupMasks = dict(common.groupMasks, **schema.groupMasks)
         return schema
-    else:
-        with open(file) as fileIO:
-            return read_schema(fileIO, include_common_registers)

harp/schema.py

@@ -0,0 +1,35 @@
+%YAML 1.1
+---
+# yaml-language-server: $schema=https://raw.githubusercontent.com/harp-tech/reflex-generator/main/schema/device.json
+device: TestDevice
+whoAmI: 0000
+firmwareVersion: "0.1"
+hardwareTargets: "0.1"
+registers:
+  DigitalInputState:
+    address: 32
+    access: Event
+    type: U8
+    maskType: DigitalInputs
+    description: Reports the state of the digital input lines.
+  DigitalInputMode:
+    address: 33
+    access: Write
+    type: U8
+    maskType: DigitalInputs
+    description: Reports the state of the digital input lines.
+bitMasks:
+  DigitalInputs:
+    description: Specifies the state of the digital input lines.
+    bits:
+      DI0: 0x1
+      DI1: 0x2
+      DI2: 0x4
+      DI3: 0x8
+groupMasks:
+  InputMode:
+    description: Specifies when the device reports the state of digital input lines.
+    values:
+      Rising: 0
+      Falling: 1
+      Both: 2

tests/data/device.yml

@@ -0,0 +1,43 @@
+import pytest
+from os import PathLike
+from typing import Iterable, Optional, Type, Union
+from pytest import mark
+from pathlib import Path
+from dataclasses import dataclass
+from harp.schema import read_schema
+
+datapath = Path(__file__).parent
+
+
+@dataclass
+class DeviceSchemaParam:
+    path: Union[str, PathLike]
+    expected_whoAmI: int
+    expected_device: Optional[int] = None
+    expected_registers: Optional[Iterable[str]] = None
+    expected_error: Optional[Type[BaseException]] = None
+
+    def __post_init__(self):
+        self.path = datapath / self.path
+
+
+testdata = [
+    DeviceSchemaParam(
+        path="data/device.yml",
+        expected_whoAmI=0,
+        expected_registers=["DigitalInputMode"],
+    )
+]
+
+
+@mark.parametrize("schemaFile", testdata)
+def test_read_schema(schemaFile: DeviceSchemaParam):
+    schema = read_schema(schemaFile.path)
+    assert schema.whoAmI == schemaFile.expected_whoAmI
+    if schemaFile.expected_registers:
+        for register in schemaFile.expected_registers:
+            assert register in schema.registers
+
+
+if __name__ == "__main__":
+    pytest.main()