Update unit handling

This refactors unit handling so that it keeps supporting old format
but also supports format in
COVESA/vehicle_signal_specification#669.

It only focus on keeping existing functionality.
No functionality for parsing and verifying domains added.

Signed-off-by: Erik Jaegervall <[email protected]>

tests/model/explicit_units_old_syntax.yaml

@@ -0,0 +1,9 @@
+units:
+  puncheon:
+    label: Puncheon
+    description: Volume measure in puncheons (1 puncheon = 318 liters)
+    domain: volume
+  hogshead:
+    label: Hogshead
+    description: Volume measure in hogsheads (1 hogshead = 238 liters)
+    domain: volume

tests/model/test_contants.py

@@ -1,7 +1,7 @@
 import pytest
 import os
 
-from vspec.model.constants import VSSType, VSSDataType, Unit, StringStyle, VSSTreeType, VSSConstant
+from vspec.model.constants import VSSType, VSSDataType, VSSUnitCollection, StringStyle, VSSTreeType, VSSUnit
 
 
 @pytest.mark.parametrize("style_enum, style_str",
@@ -30,19 +30,26 @@ def test_invalid_string_styles():
         StringStyle.from_str("not_a_valid_case")
 
 
-def test_manually_loaded_units():
+@pytest.mark.parametrize("unit_file",
+                         ['explicit_units.yaml',
+                          'explicit_units_old_syntax.yaml'])
+def test_manually_loaded_units(unit_file):
     """
     Test correct parsing of units
     """
-    unit_file = os.path.join(os.path.dirname(__file__), 'explicit_units.yaml')
-    Unit.load_config_file(unit_file)
-    assert Unit.PUNCHEON == Unit.from_str("puncheon")
-    assert Unit.HOGSHEAD == Unit.from_str("hogshead")
+    unit_file = os.path.join(os.path.dirname(__file__), unit_file)
+    VSSUnitCollection.load_config_file(unit_file)
+    assert VSSUnitCollection.get_unit("puncheon") == "puncheon"
+    assert VSSUnitCollection.get_unit("puncheon").definition == \
+           "Volume measure in puncheons (1 puncheon = 318 liters)"
+    assert VSSUnitCollection.get_unit("puncheon").unit == \
+           "Puncheon"
+    assert VSSUnitCollection.get_unit("puncheon").domain == \
+           "volume"
 
 
 def test_invalid_unit():
-    with pytest.raises(Exception):
-        Unit.from_str("not_a_valid_case")
+    assert VSSUnitCollection.get_unit("unknown") is None
 
 
 @pytest.mark.parametrize("type_enum,type_str",
@@ -108,12 +115,12 @@ def test_invalid_vss_tree_types():
         VSSDataType.from_str("not_a_valid_case")
 
 
-def test_vss_constants():
-    """ Test VSSConstant class """
-    item = VSSConstant("mylabel", "myvalue", "mydescription", "mydomain")
-    assert item.value == "myvalue"
-    assert item.label == "mylabel"
-    assert item.description == "mydescription"
+def test_unit():
+    """ Test Unit class """
+    item = VSSUnit("myid", "myunit", "mydefinition", "mydomain")
+    assert item.value == "myid"
+    assert item.unit == "myunit"
+    assert item.definition == "mydefinition"
     assert item.domain == "mydomain"
-    # String subclass so just comparing shall get "value"
-    assert item == "myvalue"
+    # String subclass so just comparing shall get "myid"
+    assert item == "myid"

tests/model/test_vsstree.py

@@ -1,7 +1,7 @@
 import unittest
 import os
 
-from vspec.model.constants import VSSType, VSSDataType, Unit, VSSTreeType
+from vspec.model.constants import VSSType, VSSDataType, VSSUnitCollection, VSSTreeType
 from vspec.model.vsstree import VSSNode
 
 
@@ -35,7 +35,7 @@ def test_complex_construction(self):
                   "aggregate": False,
                   "default": "test-default", "$file_name$": "testfile"}
         unit_file = os.path.join(os.path.dirname(__file__), 'explicit_units.yaml')
-        Unit.load_config_file(unit_file)
+        VSSUnitCollection.load_config_file(unit_file)
         node = VSSNode(
             "test",
             source,
@@ -45,7 +45,7 @@ def test_complex_construction(self):
         self.assertEqual(VSSType.SENSOR, node.type)
         self.assertEqual("26d6e362-a422-11ea-bb37-0242ac130002", node.uuid)
         self.assertEqual(VSSDataType.UINT8, node.datatype)
-        self.assertEqual(Unit.HOGSHEAD, node.unit)
+        self.assertEqual(VSSUnitCollection.get_unit("hogshead"), node.unit)
         self.assertEqual(0, node.min)
         self.assertEqual(100, node.max)
         self.assertEqual(["one", "two"], node.allowed)
@@ -71,7 +71,7 @@ def test_merge_nodes(self):
                   "datatype": "uint8", "unit": "hogshead", "min": 0, "max": 100, "$file_name$": "testfile"}
 
         unit_file = os.path.join(os.path.dirname(__file__), 'explicit_units.yaml')
-        Unit.load_config_file(unit_file)
+        VSSUnitCollection.load_config_file(unit_file)
 
         node_target = VSSNode(
             "MyNode",
@@ -95,7 +95,7 @@ def test_merge_nodes(self):
             node_target.uuid)
         self.assertTrue(node_target.has_datatype())
         self.assertEqual(VSSDataType.UINT8, node_target.datatype)
-        self.assertEqual(Unit.HOGSHEAD, node_target.unit)
+        self.assertEqual(VSSUnitCollection.get_unit("hogshead"), node_target.unit)
         self.assertEqual(0, node_target.min)
         self.assertEqual(100, node_target.max)
 

vspec/__init__.py

@@ -23,7 +23,7 @@
 
 from .model.vsstree import VSSNode
 from .model.exceptions import ImpossibleMergeException, IncompleteElementException
-from .model.constants import VSSTreeType, Unit
+from .model.constants import VSSTreeType, VSSUnitCollection
 
 nestable_types = set(["branch", "struct"])
 
@@ -871,11 +871,11 @@ def load_units(vspec_file: str, unit_files: List[str]):
         vspec_dir = os.path.dirname(os.path.realpath(vspec_file))
         default_vss_unit_file = vspec_dir + os.path.sep + 'units.yaml'
         if os.path.exists(default_vss_unit_file):
-            total_nbr_units = Unit.load_config_file(default_vss_unit_file)
+            total_nbr_units = VSSUnitCollection.load_config_file(default_vss_unit_file)
             logging.info(f"Added {total_nbr_units} units from {default_vss_unit_file}")
     else:
         for unit_file in unit_files:
-            nbr_units = Unit.load_config_file(unit_file)
+            nbr_units = VSSUnitCollection.load_config_file(unit_file)
             if (nbr_units == 0):
                 logging.warning(f"Warning: No units found in {unit_file}")
             else:

vspec/model/constants.py

@@ -13,9 +13,11 @@
 #
 # noinspection PyPackageRequirements
 import re
+import logging
+import sys
 from enum import Enum, EnumMeta
 from typing import (
-    Sequence, Type, TypeVar, Optional, Dict, Tuple, Iterator, TextIO
+    Sequence, Type, TypeVar, Optional, Dict, TextIO
 )
 
 import yaml
@@ -26,17 +28,20 @@
 T = TypeVar("T")
 
 
-class VSSConstant(str):
-    """String subclass that can tag it with description and domain.
+class VSSUnit(str):
+    """String subclass for storing unit information.
     """
-    label: str
-    description: Optional[str] = None
-    domain: Optional[str] = None
-
-    def __new__(cls, label: str, value: str, description: str = "", domain: str = "") -> 'VSSConstant':
-        self = super().__new__(cls, value)
-        self.label = label
-        self.description = description
+    id: str  # Typically abbreviation like "V"
+    unit: Optional[str] = None  # Typically full name like "Volt"
+    definition: Optional[str] = None
+    domain: Optional[str] = None  # Typically quantity, like "Voltage"
+
+    def __new__(cls, id: str, unit: Optional[str] = None, definition: Optional[str] = None,
+                domain: Optional[str] = None) -> 'VSSUnit':
+        self = super().__new__(cls, id)
+        self.id = id
+        self.unit = unit
+        self.definition = definition
         self.domain = domain
         return self
 
@@ -45,61 +50,6 @@ def value(self):
         return self
 
 
-def dict_to_constant_config(name: str, info: Dict[str, str]) -> Tuple[str, VSSConstant]:
-    label = info['label']
-    label = NON_ALPHANUMERIC_WORD.sub('', label).upper()
-    description = info.get('description', '')
-    domain = info.get('domain', '')
-    return label, VSSConstant(info['label'], name, description, domain)
-
-
-def iterate_config_members(config: Dict[str, Dict[str, str]]) -> Iterator[Tuple[str, VSSConstant]]:
-    for u, v in config.items():
-        yield dict_to_constant_config(u, v)
-
-
-class VSSRepositoryMeta(type):
-    """This class defines the enumeration behavior for vss:
-     - Access through Class.ATTRIBUTE
-     - Class.add_config(Dict[str, Dict[str, str]]): Adds values from file
-     - from_str(str): reverse lookup
-     - values(): sequence of values
-    """
-
-    def __new__(mcs, cls, bases, classdict):
-        cls = super().__new__(mcs, cls, bases, classdict)
-
-        if not hasattr(cls, '__reverse_lookup__'):
-            cls.__reverse_lookup__ = {
-                v.value: v for v in cls.__members__.values()
-            }
-        if not hasattr(cls, '__values__'):
-            cls.__values__ = list(cls.__reverse_lookup__.keys())
-
-        return cls
-
-    def __getattr__(cls, key: str) -> str:
-        try:
-            return cls.__members__[key]  # type: ignore[index]
-        except KeyError as e:
-            raise AttributeError(
-                f"type object '{cls.__name__}' has no attribute '{key}'"
-            ) from e
-
-    def add_config(cls, config: Dict[str, Dict[str, str]]):
-        for k, v in iterate_config_members(config):
-            if v.value not in cls.__reverse_lookup__ and k not in cls.__members__:
-                cls.__members__[k] = v  # type: ignore[index]
-                cls.__reverse_lookup__[v.value] = v  # type: ignore[index]
-                cls.__values__.append(v.value)  # type: ignore[attr-defined]
-
-    def from_str(cls: Type[T], value: str) -> T:
-        return cls.__reverse_lookup__[value]  # type: ignore[attr-defined]
-
-    def values(cls: Type[T]) -> Sequence[str]:
-        return cls.__values__  # type: ignore[attr-defined]
-
-
 class EnumMetaWithReverseLookup(EnumMeta):
     """This class extends EnumMeta and adds:
      - from_str(str): reverse lookup
@@ -175,24 +125,58 @@ class VSSDataType(Enum, metaclass=EnumMetaWithReverseLookup):
     STRING_ARRAY = "string[]"
 
 
-class Unit(metaclass=VSSRepositoryMeta):
-    __members__: Dict[str, str] = dict()
+class VSSUnitCollection():
+    units: Dict[str, VSSUnit] = dict()
 
     @staticmethod
     def get_config_dict(yaml_file: TextIO, key: str) -> Dict[str, Dict[str, str]]:
         yaml_config = yaml.safe_load(yaml_file)
-        configs = yaml_config.get(key, {})
+        if (len(yaml_config) == 1) and (key in yaml_config):
+            # Old style unit file
+            configs = yaml_config.get(key, {})
+        else:
+            # New style unit file
+            configs = yaml_config
         return configs
 
-    @staticmethod
-    def load_config_file(config_file: str) -> int:
+    @classmethod
+    def load_config_file(cls, config_file: str) -> int:
         added_configs = 0
         with open(config_file) as my_yaml_file:
-            my_units = Unit.get_config_dict(my_yaml_file, 'units')
+            my_units = cls.get_config_dict(my_yaml_file, 'units')
             added_configs = len(my_units)
-            Unit.add_config(my_units)
+            for k, v in my_units.items():
+                unit = k
+                if "unit" in v:
+                    unit = v["unit"]
+                elif "label" in v:
+                    # Old syntax
+                    unit = v["label"]
+                definition = None
+                if "definition" in v:
+                    definition = v["definition"]
+                elif "description" in v:
+                    # Old syntax
+                    definition = v["description"]
+
+                domain = None
+                if "domain" in v:
+                    domain = v["domain"]
+                else:
+                    logging.error("No domain found for unit %s", k)
+                    sys.exit(-1)
+
+                unit_node = VSSUnit(k, unit, definition, domain)
+                cls.units[k] = unit_node
         return added_configs
 
+    @classmethod
+    def get_unit(cls, id: str) -> Optional[VSSUnit]:
+        if id in cls.units:
+            return cls.units[id]
+        else:
+            return None
+
 
 class VSSTreeType(Enum, metaclass=EnumMetaWithReverseLookup):
     SIGNAL_TREE = "signal_tree"

vspec/model/vsstree.py

@@ -9,7 +9,7 @@
 # SPDX-License-Identifier: MPL-2.0
 
 from anytree import Node, Resolver, ChildResolverError, RenderTree  # type: ignore[import]
-from .constants import VSSType, VSSDataType, Unit, VSSConstant
+from .constants import VSSType, VSSDataType, VSSUnitCollection, VSSUnit
 from .exceptions import NameStyleValidationException, \
     ImpossibleMergeException, IncompleteElementException
 from typing import Any, Optional, Set, List
@@ -46,7 +46,7 @@ class VSSNode(Node):
     # neither in core or extended,
     whitelisted_extended_attributes: List[str] = []
 
-    unit: Optional[VSSConstant]
+    unit: Optional[VSSUnit]
 
     min = ""
     max = ""
@@ -147,9 +147,8 @@ def extractCoreAttribute(name: str):
                 sys.exit(-1)
 
             unit = self.source_dict["unit"]
-            try:
-                self.unit = Unit.from_str(unit)
-            except KeyError:
+            self.unit = VSSUnitCollection.get_unit(unit)
+            if self.unit is None:
                 logging.error(f"Unknown unit {unit} for signal {self.qualified_name()}. Terminating.")
                 sys.exit(-1)