sift_py.yaml.channel

  1from __future__ import annotations
  2
  3import re
  4from typing import Any, Dict, List, Literal, Union, cast
  5
  6from typing_extensions import NotRequired, TypedDict
  7
  8from sift_py.error import _component_deprecation_warning
  9from sift_py.ingestion.channel import ChannelDataTypeStrRep
 10from sift_py.ingestion.config.yaml.error import YamlConfigError
 11from sift_py.yaml.utils import _type_fqn
 12
 13_CHANNEL_REFERENCE_REGEX = re.compile(r"^\$\d+$")
 14
 15
 16def _validate_channel_anchor(val: Any):
 17    if not isinstance(val, str):
 18        raise YamlConfigError._invalid_property(
 19            val,
 20            "<str>",
 21            "&str",
 22            ["channels"],
 23        )
 24
 25
 26def _validate_channel(val: Any):
 27    channel = cast(Dict[Any, Any], val)
 28
 29    name = channel.get("name")
 30
 31    if not isinstance(name, str):
 32        raise YamlConfigError._invalid_property(name, "- name", "str", ["channels"])
 33
 34    description = channel.get("description")
 35
 36    if description is not None and not isinstance(description, str):
 37        raise YamlConfigError._invalid_property(description, "- description", "str", ["channels"])
 38
 39    unit = channel.get("unit")
 40
 41    if unit is not None and not isinstance(unit, str):
 42        raise YamlConfigError._invalid_property(unit, "- unit", "str", ["channels"])
 43    component = channel.get("component")
 44    if component is not None:
 45        _component_deprecation_warning()
 46        if not isinstance(component, str):
 47            raise YamlConfigError._invalid_property(component, "- component", "str", ["channels"])
 48
 49    data_type = channel.get("data_type")
 50    valid_data_type_values = [v.value for v in ChannelDataTypeStrRep]
 51
 52    if not data_type in valid_data_type_values:
 53        raise YamlConfigError._invalid_property(
 54            data_type,
 55            "- data_type",
 56            " | ".join(valid_data_type_values),
 57            ["channels"],
 58        )
 59
 60    if data_type == ChannelDataTypeStrRep.ENUM.value:
 61        enum_types = channel.get("enum_types")
 62
 63        if not isinstance(enum_types, list):
 64            raise YamlConfigError._invalid_property(
 65                enum_types,
 66                "- enum_types",
 67                f"List<{_type_fqn(ChannelEnumTypeYamlSpec)}>",
 68                ["channels"],
 69            )
 70
 71        for enum_type in cast(List[Any], enum_types):
 72            _validate_enum_type(enum_type)
 73
 74    elif data_type == ChannelDataTypeStrRep.BIT_FIELD.value:
 75        bit_field_elements = channel.get("bit_field_elements")
 76
 77        if not isinstance(bit_field_elements, list):
 78            raise YamlConfigError._invalid_property(
 79                bit_field_elements,
 80                "- bit_field_elements",
 81                f"List<{_type_fqn(ChannelBitFieldElementYamlSpec)}>",
 82            )
 83
 84        for bit_field_element in cast(List[Any], bit_field_elements):
 85            _validate_bit_field_element(bit_field_element)
 86
 87    else:
 88        enum_types = channel.get("enum_types")
 89
 90        if enum_types is not None:
 91            raise YamlConfigError(
 92                f"Channel of data-type '{data_type}' should not have 'enum_types' set."
 93            )
 94
 95        bit_field_elements = channel.get("bit_field_elements")
 96
 97        if bit_field_elements is not None:
 98            raise YamlConfigError(
 99                f"Channel of data-type '{data_type}' should not have 'bit_field_elements' set."
100            )
101
102
103def _validate_enum_type(val: Any):
104    enum_type = cast(Dict[Any, Any], val)
105
106    name = enum_type.get("name")
107
108    if not isinstance(name, str):
109        raise YamlConfigError._invalid_property(
110            name,
111            "- name",
112            "str",
113            ["channels", "- enum_type"],
114        )
115
116    key = enum_type.get("key")
117
118    if not isinstance(key, int):
119        raise YamlConfigError._invalid_property(
120            key,
121            "- key",
122            "int",
123            ["channels", "- enum_type"],
124        )
125
126
127def _validate_bit_field_element(val: Any):
128    bit_field_element = cast(Dict[Any, Any], val)
129
130    name = bit_field_element.get("name")
131
132    if not isinstance(name, str):
133        raise YamlConfigError._invalid_property(
134            name, "- name", "str", ["channels", "- bit_field_elements"]
135        )
136
137    index = bit_field_element.get("index")
138
139    if not isinstance(index, int):
140        raise YamlConfigError._invalid_property(
141            name, "- index", "int", ["channels", "- bit_field_elements"]
142        )
143
144    bit_count = bit_field_element.get("bit_count")
145
146    if not isinstance(bit_count, int):
147        raise YamlConfigError._invalid_property(
148            name, "- bit_count", "int", ["channels", "- bit_field_elements"]
149        )
150
151
152def _validate_channel_reference(val: Any):
153    channel_reference = cast(Dict[Any, Any], val)
154
155    for key, value in channel_reference.items():
156        if not isinstance(key, str):
157            raise YamlConfigError._invalid_property(
158                channel_reference,
159                "- <str>",
160                f"Dict[str, {_type_fqn(ChannelConfigYamlSpec)}]",
161                ["rules", "- channel_references"],
162            )
163
164        if _CHANNEL_REFERENCE_REGEX.match(key) is None:
165            raise YamlConfigError(
166                f"Invalid channel reference key '{key}'. Expected an integer prefixed with '$' e.g. '$1', '$2', and so on."
167            )
168
169        if isinstance(value, dict):  # Do this for YamlConfigChannelSpec but not str
170            try:
171                _validate_channel(value)
172            except YamlConfigError as err:
173                raise YamlConfigError(f"Rule '{key}' contains an invalid channel reference:\n{err}")
174
175
176class ChannelConfigYamlSpec(TypedDict):
177    """
178    Formal spec that defines what a channel should look like in YAML.
179
180    `name`: Name of channel.
181    `description`: Optional channel description.
182    `unit`: Unit of measurement.
183    `component`: Name of component that channel belongs to.
184    `data_type`: Type of the data associated with the channel.
185    `enum_types`: Required if `data_type` is `enum.
186    `bit_field_elements`: Required if `data_type` is `bit_field`.
187    """
188
189    name: str
190    description: NotRequired[str]
191    unit: NotRequired[str]
192    component: NotRequired[str]
193    data_type: Union[
194        Literal["double"],
195        Literal["string"],
196        Literal["enum"],
197        Literal["bit_field"],
198        Literal["bool"],
199        Literal["float"],
200        Literal["int32"],
201        Literal["int64"],
202        Literal["uint32"],
203        Literal["uint64"],
204    ]
205    enum_types: NotRequired[List[ChannelEnumTypeYamlSpec]]
206    bit_field_elements: NotRequired[List[ChannelBitFieldElementYamlSpec]]
207
208
209class ChannelEnumTypeYamlSpec(TypedDict):
210    """
211    Formal spec that defines what a channel enum type should look like in YAML.
212    """
213
214    name: str
215    key: int
216
217
218class ChannelBitFieldElementYamlSpec(TypedDict):
219    """
220    Formal spec that defines what a bit-field element should look like in YAML.
221    """
222
223    name: str
224    index: int
225    bit_count: int