|
13 | 13 | # |
14 | 14 | # noinspection PyPackageRequirements |
15 | 15 | import re |
| 16 | +import logging |
| 17 | +import sys |
16 | 18 | from enum import Enum, EnumMeta |
17 | 19 | from typing import ( |
18 | | - Sequence, Type, TypeVar, Optional, Dict, Tuple, Iterator, TextIO |
| 20 | + Sequence, Type, TypeVar, Optional, Dict, TextIO |
19 | 21 | ) |
20 | 22 |
|
21 | 23 | import yaml |
|
26 | 28 | T = TypeVar("T") |
27 | 29 |
|
28 | 30 |
|
29 | | -class VSSConstant(str): |
30 | | - """String subclass that can tag it with description and domain. |
| 31 | +class VSSUnit(str): |
| 32 | + """String subclass for storing unit information. |
31 | 33 | """ |
32 | | - label: str |
33 | | - description: Optional[str] = None |
34 | | - domain: Optional[str] = None |
35 | | - |
36 | | - def __new__(cls, label: str, value: str, description: str = "", domain: str = "") -> 'VSSConstant': |
37 | | - self = super().__new__(cls, value) |
38 | | - self.label = label |
39 | | - self.description = description |
40 | | - self.domain = domain |
| 34 | + id: str # Typically abbreviation like "V" |
| 35 | + unit: Optional[str] = None # Typically full name like "Volt" |
| 36 | + definition: Optional[str] = None |
| 37 | + quantity: Optional[str] = None # Typically quantity, like "Voltage" |
| 38 | + |
| 39 | + def __new__(cls, id: str, unit: Optional[str] = None, definition: Optional[str] = None, |
| 40 | + quantity: Optional[str] = None) -> 'VSSUnit': |
| 41 | + self = super().__new__(cls, id) |
| 42 | + self.id = id |
| 43 | + self.unit = unit |
| 44 | + self.definition = definition |
| 45 | + self.quantity = quantity |
41 | 46 | return self |
42 | 47 |
|
43 | 48 | @property |
44 | 49 | def value(self): |
45 | 50 | return self |
46 | 51 |
|
47 | 52 |
|
48 | | -def dict_to_constant_config(name: str, info: Dict[str, str]) -> Tuple[str, VSSConstant]: |
49 | | - label = info['label'] |
50 | | - label = NON_ALPHANUMERIC_WORD.sub('', label).upper() |
51 | | - description = info.get('description', '') |
52 | | - domain = info.get('domain', '') |
53 | | - return label, VSSConstant(info['label'], name, description, domain) |
54 | | - |
55 | | - |
56 | | -def iterate_config_members(config: Dict[str, Dict[str, str]]) -> Iterator[Tuple[str, VSSConstant]]: |
57 | | - for u, v in config.items(): |
58 | | - yield dict_to_constant_config(u, v) |
59 | | - |
60 | | - |
61 | | -class VSSRepositoryMeta(type): |
62 | | - """This class defines the enumeration behavior for vss: |
63 | | - - Access through Class.ATTRIBUTE |
64 | | - - Class.add_config(Dict[str, Dict[str, str]]): Adds values from file |
65 | | - - from_str(str): reverse lookup |
66 | | - - values(): sequence of values |
67 | | - """ |
68 | | - |
69 | | - def __new__(mcs, cls, bases, classdict): |
70 | | - cls = super().__new__(mcs, cls, bases, classdict) |
71 | | - |
72 | | - if not hasattr(cls, '__reverse_lookup__'): |
73 | | - cls.__reverse_lookup__ = { |
74 | | - v.value: v for v in cls.__members__.values() |
75 | | - } |
76 | | - if not hasattr(cls, '__values__'): |
77 | | - cls.__values__ = list(cls.__reverse_lookup__.keys()) |
78 | | - |
79 | | - return cls |
80 | | - |
81 | | - def __getattr__(cls, key: str) -> str: |
82 | | - try: |
83 | | - return cls.__members__[key] # type: ignore[index] |
84 | | - except KeyError as e: |
85 | | - raise AttributeError( |
86 | | - f"type object '{cls.__name__}' has no attribute '{key}'" |
87 | | - ) from e |
88 | | - |
89 | | - def add_config(cls, config: Dict[str, Dict[str, str]]): |
90 | | - for k, v in iterate_config_members(config): |
91 | | - if v.value not in cls.__reverse_lookup__ and k not in cls.__members__: |
92 | | - cls.__members__[k] = v # type: ignore[index] |
93 | | - cls.__reverse_lookup__[v.value] = v # type: ignore[index] |
94 | | - cls.__values__.append(v.value) # type: ignore[attr-defined] |
95 | | - |
96 | | - def from_str(cls: Type[T], value: str) -> T: |
97 | | - return cls.__reverse_lookup__[value] # type: ignore[attr-defined] |
98 | | - |
99 | | - def values(cls: Type[T]) -> Sequence[str]: |
100 | | - return cls.__values__ # type: ignore[attr-defined] |
101 | | - |
102 | | - |
103 | 53 | class EnumMetaWithReverseLookup(EnumMeta): |
104 | 54 | """This class extends EnumMeta and adds: |
105 | 55 | - from_str(str): reverse lookup |
@@ -175,24 +125,61 @@ class VSSDataType(Enum, metaclass=EnumMetaWithReverseLookup): |
175 | 125 | STRING_ARRAY = "string[]" |
176 | 126 |
|
177 | 127 |
|
178 | | -class Unit(metaclass=VSSRepositoryMeta): |
179 | | - __members__: Dict[str, str] = dict() |
| 128 | +class VSSUnitCollection(): |
| 129 | + units: Dict[str, VSSUnit] = dict() |
180 | 130 |
|
181 | 131 | @staticmethod |
182 | 132 | def get_config_dict(yaml_file: TextIO, key: str) -> Dict[str, Dict[str, str]]: |
183 | 133 | yaml_config = yaml.safe_load(yaml_file) |
184 | | - configs = yaml_config.get(key, {}) |
| 134 | + if (len(yaml_config) == 1) and (key in yaml_config): |
| 135 | + # Old style unit file |
| 136 | + configs = yaml_config.get(key, {}) |
| 137 | + else: |
| 138 | + # New style unit file |
| 139 | + configs = yaml_config |
185 | 140 | return configs |
186 | 141 |
|
187 | | - @staticmethod |
188 | | - def load_config_file(config_file: str) -> int: |
| 142 | + @classmethod |
| 143 | + def load_config_file(cls, config_file: str) -> int: |
189 | 144 | added_configs = 0 |
190 | 145 | with open(config_file) as my_yaml_file: |
191 | | - my_units = Unit.get_config_dict(my_yaml_file, 'units') |
| 146 | + my_units = cls.get_config_dict(my_yaml_file, 'units') |
192 | 147 | added_configs = len(my_units) |
193 | | - Unit.add_config(my_units) |
| 148 | + for k, v in my_units.items(): |
| 149 | + unit = k |
| 150 | + if "unit" in v: |
| 151 | + unit = v["unit"] |
| 152 | + elif "label" in v: |
| 153 | + # Old syntax |
| 154 | + unit = v["label"] |
| 155 | + definition = None |
| 156 | + if "definition" in v: |
| 157 | + definition = v["definition"] |
| 158 | + elif "description" in v: |
| 159 | + # Old syntax |
| 160 | + definition = v["description"] |
| 161 | + |
| 162 | + quantity = None |
| 163 | + if "quantity" in v: |
| 164 | + quantity = v["quantity"] |
| 165 | + elif "domain" in v: |
| 166 | + # Old syntax |
| 167 | + quantity = v["domain"] |
| 168 | + else: |
| 169 | + logging.error("No quantity (domain) found for unit %s", k) |
| 170 | + sys.exit(-1) |
| 171 | + |
| 172 | + unit_node = VSSUnit(k, unit, definition, quantity) |
| 173 | + cls.units[k] = unit_node |
194 | 174 | return added_configs |
195 | 175 |
|
| 176 | + @classmethod |
| 177 | + def get_unit(cls, id: str) -> Optional[VSSUnit]: |
| 178 | + if id in cls.units: |
| 179 | + return cls.units[id] |
| 180 | + else: |
| 181 | + return None |
| 182 | + |
196 | 183 |
|
197 | 184 | class VSSTreeType(Enum, metaclass=EnumMetaWithReverseLookup): |
198 | 185 | SIGNAL_TREE = "signal_tree" |
|
0 commit comments