Zultron/2022 09 19 rebase for PRs: 4 config io

hw_device_mgr/cia_301/command.py

@@ -25,7 +25,9 @@ def scan_bus(self, bus=0):
         """Scan bus, returning list of addresses and IDs for each device."""
 
     @abc.abstractmethod
-    def upload(self, address=None, index=None, subindex=0, datatype=None):
+    def upload(
+        self, address=None, index=None, subindex=0, datatype=None, **kwargs
+    ):
         """Upload a value from a device SDO."""
 
     @abc.abstractmethod
@@ -36,6 +38,7 @@ def download(
         subindex=0,
         value=None,
         datatype=None,
+        **kwargs,
     ):
         """Download a value to a device SDO."""
 

hw_device_mgr/cia_301/config.py

@@ -1,6 +1,7 @@
 from .data_types import CiA301DataType
-from .command import CiA301Command, CiA301SimCommand
+from .command import CiA301Command, CiA301SimCommand, CiA301CommandException
 from .sdo import CiA301SDO
+from functools import cached_property
 
 
 class CiA301Config:
@@ -98,28 +99,45 @@ def sdo_ix(cls, ix):
         ix = (dtc.uint16(ix[0]), dtc.uint8(ix[1]))
         return ix
 
+    @cached_property
+    def sdos(self):
+        assert self.model_id in self._model_sdos
+        return self._model_sdos[self.model_id].values()
+
     def sdo(self, ix):
         if isinstance(ix, self.sdo_class):
             return ix
         ix = self.sdo_ix(ix)
         return self._model_sdos[self.model_id][ix]
 
+    def dump_param_values(self):
+        res = dict()
+        for sdo in self.sdos:
+            try:
+                res[sdo] = self.upload(sdo, stderr_to_devnull=True)
+            except CiA301CommandException as e:
+                # Objects may not exist, like variable length PDO mappings
+                self.logger.debug(f"Upload {sdo} failed:  {e}")
+                pass
+        return res
+
     #
     # Param read/write
     #
 
-    def upload(self, sdo):
+    def upload(self, sdo, **kwargs):
         # Get SDO object
         sdo = self.sdo(sdo)
         res_raw = self.command().upload(
             address=self.address,
             index=sdo.index,
             subindex=sdo.subindex,
             datatype=sdo.data_type,
+            **kwargs,
         )
         return sdo.data_type(res_raw)
 
-    def download(self, sdo, val, dry_run=False):
+    def download(self, sdo, val, dry_run=False, **kwargs):
         # Get SDO object
         sdo = self.sdo(sdo)
         # Check before setting value to avoid unnecessary NVRAM writes
@@ -128,6 +146,7 @@ def download(self, sdo, val, dry_run=False):
             index=sdo.index,
             subindex=sdo.subindex,
             datatype=sdo.data_type,
+            **kwargs,
         )
         if sdo.data_type(res_raw) == val:
             return  # SDO value already correct
@@ -140,6 +159,7 @@ def download(self, sdo, val, dry_run=False):
             subindex=sdo.subindex,
             value=val,
             datatype=sdo.data_type,
+            **kwargs,
         )
 
     #
@@ -187,24 +207,21 @@ def set_device_config(cls, config):
         cls._device_config.clear()
         cls._device_config.extend(config)
 
-    def munge_config(self, config_raw):
+    @classmethod
+    def munge_config(cls, config_raw, position):
+        config_cooked = config_raw.copy()
+        # Convert model ID ints
+        model_id = (config_raw["vendor_id"], config_raw["product_code"])
+        model_id = cls.format_model_id(model_id)
+        config_cooked["vendor_id"], config_cooked["product_code"] = model_id
         # Flatten out param_values key
-        pv = dict()
+        config_cooked["param_values"] = dict()
         for ix, val in config_raw.get("param_values", dict()).items():
-            ix = self.sdo_class.parse_idx_str(ix)
+            ix = cls.sdo_class.parse_idx_str(ix)
             if isinstance(val, list):
-                pos_ix = config_raw["positions"].index(self.position)
+                pos_ix = config_raw["positions"].index(position)
                 val = val[pos_ix]
-            pv[ix] = val
-        dtc = self.data_type_class
-        config_raw["vendor_id"] = dtc.uint32(config_raw["vendor_id"])
-        config_raw["product_code"] = dtc.uint32(config_raw["product_code"])
-        config_cooked = dict(
-            vendor_id=config_raw["vendor_id"],
-            product_code=config_raw["product_code"],
-            param_values=pv,
-            sync_manager=config_raw.get("sync_manager", dict()),
-        )
+            config_cooked["param_values"][ix] = val
         # Return pruned config dict
         return config_cooked
 
@@ -225,7 +242,7 @@ def config(self):
             else:
                 raise KeyError(f"No config for device at {self.address}")
             # Prune & cache config
-            self._config = self.munge_config(conf)
+            self._config = self.munge_config(conf, self.position)
 
         # Return cached config
         return self._config

hw_device_mgr/cia_301/tests/base_test_class.py

@@ -20,10 +20,12 @@ class BaseCiA301TestClass(BaseTestClass):
     #
 
     # The device configuration, as in a real system
-    device_config_yaml = "cia_301/tests/device_config.yaml"
+    device_config_package = "hw_device_mgr.cia_301.tests"
+    device_config_yaml = "device_config.yaml"
 
     # Device model SDOs; for test fixture
-    device_sdos_yaml = "cia_301/tests/bogus_devices/sim_sdo_data.yaml"
+    device_sdos_package = "hw_device_mgr.cia_301.tests"
+    device_sdos_yaml = "sim_sdo_data.yaml"
 
     # Classes under test in this module
     data_type_class = CiA301DataType
@@ -42,18 +44,27 @@ class BaseCiA301TestClass(BaseTestClass):
 
     @classmethod
     def init_sim(cls, **kwargs):
+        """Create sim device objects with configured SDOs."""
         if cls.pass_init_sim_device_sdos:
             # Init sim SDO data
-            path, sdo_data = cls.load_yaml(cls.device_sdos_yaml, True)
-            print(f"  Raw sdo_data from {path}")
+            sdo_data = cls.load_sdo_data()
+            print(f"  Raw sdo_data from {cls.sdo_data_resource()}")
             kwargs["sdo_data"] = cls.munge_sdo_data(sdo_data)
         # Init sim device data
         super().init_sim(**kwargs)
 
     @classmethod
     def munge_device_config(cls, device_config):
-        # Make device_config.yaml reusable by monkey-patching device vendor_id
-        # and product_code keys based on test_category key
+        """
+        Munge raw device config.
+
+        Return a copy of `device_config` with minor processing.
+
+        Optionally, to make the YAML file reusable, each configuration's
+        `vendor_id` and `product_code` keys may be replaced with a `category`
+        key matching a parent of classes listed; this fixture will re-add those
+        keys.
+        """
         new_device_config = list()
         for conf in device_config:
             device_cls = cls.test_category_class(conf["test_category"])
@@ -84,30 +95,34 @@ def config_cls(self, device_cls, device_config):
     def command_cls(self, device_cls):
         yield self.command_class
 
+    @classmethod
+    def load_device_config(cls):
+        """
+        Load device configuration from package resource.
+
+        The `importlib.resources` resource is named by
+        `device_config_package` and `device_config_yaml` attrs.
+        """
+        rsrc = (cls.device_config_package, cls.device_config_yaml)
+        dev_conf = cls.load_yaml_resource(*rsrc)
+        assert dev_conf, f"Empty device config in package resource {rsrc}"
+        print(f"  Raw device_config from {rsrc}")
+        return dev_conf
+
     @pytest.fixture
     def device_config(self):
         """
         Device configuration data fixture.
 
-        Load device configuration from file named in
-        `device_config_yaml` attr.
+        Load device configuration with `load_device_config()` and munge with
+        `mung_device_config()`.
 
         Device configuration in the same format as non-test
         configuration, described in `Config` classes.
-
-        The absolute path is stored in the test object
-        `device_config_path` attribute.
-
-        Optionally, to make the YAML file reusable, each
-        configuration's `vendor_id` and `product_code` keys may be
-        replaced with a `category` key matching a parent of classes
-        listed; this fixture will re-add those keys.
         """
-        path, dev_conf = self.load_yaml(self.device_config_yaml, True)
-        print(f"  Raw device_config from {path}")
-        dev_conf = self.munge_device_config(dev_conf)
+        conf_raw = self.load_device_config()
+        dev_conf = self.munge_device_config(conf_raw)
         self.device_config = dev_conf
-        self.device_config_path = path
         yield dev_conf
 
     @pytest.fixture
@@ -217,17 +232,26 @@ def munge_sim_device_data(cls, sim_device_data):
             dev["test_address"] = (dev["bus"], dev["position"])
         return sim_device_data
 
+    @classmethod
+    def sdo_data_resource(cls):
+        return (cls.device_sdos_package, cls.device_sdos_yaml)
+
+    @classmethod
+    def load_sdo_data(cls):
+        rsrc = cls.sdo_data_resource()
+        sdo_data = cls.load_yaml_resource(*rsrc)
+        assert sdo_data, f"Empty SDO data in package resource {rsrc}"
+        return sdo_data
+
     def pytest_generate_tests(self, metafunc):
         # Dynamic parametrization from sim_device_data_yaml:
         # - _sim_device_data:  iterate through `sim_device_data` list
         #   - with `_sdo_data`:  add matching entry in `sdo_data` list
         # - _sdo_data:  iterate through `sdo_data` values
         # - bus:  iterate through `sim_device_data` unique `bus` values
         # *Note all three cases are mutually exclusive
-        path, dev_data = self.load_yaml(self.sim_device_data_yaml, True)
-        dev_data = self.munge_sim_device_data(dev_data)
-        path, sdo_data = self.load_yaml(self.device_sdos_yaml, True)
-        sdo_data = self.munge_sdo_data(sdo_data, conv_sdos=True)
+        dev_data = self.munge_sim_device_data(self.load_sim_device_data())
+        sdo_data = self.munge_sdo_data(self.load_sdo_data(), conv_sdos=True)
         names = list()
         vals, ids = (list(), list())
         if "_sim_device_data" in metafunc.fixturenames:

hw_device_mgr/cia_301/tests/test_device.py

@@ -10,8 +10,8 @@ class TestCiA301Device(BaseCiA301TestClass, _TestDevice):
         *_TestDevice.expected_mro,
     ]
 
-    # Test CiA NMT init:  online & operational status
-    read_update_write_yaml = "cia_301/tests/read_update_write.cases.yaml"
+    # CiA NMT init online & operational status test cases
+    read_update_write_package = "hw_device_mgr.cia_301.tests"
 
     @pytest.fixture
     def obj(self, device_cls, sim_device_data):

hw_device_mgr/cia_402/device.py

@@ -173,10 +173,10 @@ def get_feedback(self):
                 goal_reasons.append(f"state flag {flag_name} != {not flag_val}")
 
         if not goal_reached:
-            fb_out.update(
-                goal_reached=False, goal_reason="; ".join(goal_reasons)
-            )
-            self.logger.debug(f"Device {self.address}:  Goal not reached:")
+            goal_reason = "; ".join(goal_reasons)
+            fb_out.update(goal_reached=False, goal_reason=goal_reason)
+            if fb_out.changed("goal_reason"):
+                self.logger.debug(f"{self}:  Goal not reached: {goal_reason}")
         return fb_out
 
     state_bits = {
@@ -492,13 +492,13 @@ def set_sim_feedback(self):
             # Log changes
             if self.sim_feedback.changed("control_mode_fb"):
                 cm = self.sim_feedback.get("control_mode_fb")
-                self.logger.info(f"{self} next control_mode_fb:  0x{cm:04X}")
+                self.logger.info(f"{self} sim control_mode_fb:  0x{cm:04X}")
             if self.sim_feedback.changed("status_word"):
                 sw = self.sim_feedback.get("status_word")
                 flags = ",".join(k for k, v in sw_flags.items() if v)
                 flags = f" flags:  {flags}" if flags else ""
                 self.logger.info(
-                    f"{self} sim next status_word:  0x{sw:04X} {state} {flags}"
+                    f"{self} sim status_word:  0x{sw:04X} {state} {flags}"
                 )
 
         return sfb

hw_device_mgr/cia_402/tests/base_test_class.py

@@ -10,7 +10,8 @@
 class BaseCiA402TestClass(BaseCiA301TestClass):
 
     # test_read_update_write_402() configuration
-    read_update_write_402_yaml = "cia_402/tests/read_update_write.cases.yaml"
+    read_update_write_402_package = "hw_device_mgr.cia_402.tests"
+    read_update_write_402_yaml = "read_update_write.cases.yaml"
 
     # Classes under test in this module
     device_class = BogusCiA402Device

hw_device_mgr/cia_402/tests/test_device.py

@@ -29,11 +29,12 @@ def read_update_write_conv_test_data(self):
                     if key in intf_data:
                         intf_data[key] = uint16(intf_data[key])
 
-    def test_read_update_write(self, obj, fpath):
+    def test_read_update_write(self, obj):
         if hasattr(obj, "MODE_CSP"):
             # CiA 402 device
+            self.read_update_write_package = self.read_update_write_402_package
             self.read_update_write_yaml = self.read_update_write_402_yaml
             self.is_402_device = True
         else:
             self.is_402_device = False
-        super().test_read_update_write(obj, fpath)
+        super().test_read_update_write(obj)

hw_device_mgr/config_io.py

@@ -0,0 +1,39 @@
+import ruamel.yaml
+from pathlib import Path
+from importlib.resources import open_binary
+
+
+class ConfigIO:
+    @classmethod
+    def open_path(cls, path, *args, **kwargs):
+        """Return open file object for `path`."""
+        path_obj = Path(path)
+        return path_obj.open(*args, **kwargs)
+
+    @classmethod
+    def open_resource(cls, package, resource):
+        """Return open file object for importlib package resource."""
+        return open_binary(package, resource)
+
+    @classmethod
+    def load_yaml_path(cls, path):
+        """Read and return `data` from YAML formatted file `path`."""
+        yaml = ruamel.yaml.YAML()
+        with cls.open_path(path) as f:
+            data = yaml.load(f)
+        return data
+
+    @classmethod
+    def dump_yaml_path(cls, path, data):
+        """Dump `data` in YAML format to `path`."""
+        yaml = ruamel.yaml.YAML()
+        with cls.open_path(path, "w") as f:
+            yaml.dump(data, f)
+
+    @classmethod
+    def load_yaml_resource(cls, package, resource):
+        """Load YAML from importlib package resource."""
+        yaml = ruamel.yaml.YAML()
+        with cls.open_resource(package, resource) as f:
+            data = yaml.load(f)
+        return data