sift_py.yaml.rule

  1from __future__ import annotations
  2
  3import re
  4from pathlib import Path
  5from typing import Any, Dict, List, Literal, Union, cast
  6
  7import yaml
  8from typing_extensions import NotRequired, TypedDict
  9
 10from sift_py.ingestion.config.yaml.error import YamlConfigError
 11from sift_py.rule.config import RuleActionAnnotationKind
 12from sift_py.yaml.channel import (
 13    ChannelConfigYamlSpec,
 14    _validate_channel_reference,
 15)
 16from sift_py.yaml.utils import _handle_subdir, _type_fqn
 17
 18_SUB_EXPRESSION_REGEX = re.compile(r"^\$[a-zA-Z_]+$")
 19
 20
 21def load_named_expression_modules(paths: List[Path]) -> Dict[str, str]:
 22    """
 23    Takes in a list of paths to YAML files which contains named expressions and processes them into a `dict`.
 24    The key is the name of the expression and the value is the expression itself. For more information on
 25    named expression modules see `sift_py/yaml/rule.py`.
 26    """
 27
 28    named_expressions = {}
 29
 30    for path in paths:
 31        named_expr_module = _read_named_expression_module_yaml(path)
 32
 33        for name, expr in named_expr_module.items():
 34            if name in named_expressions:
 35                raise YamlConfigError(
 36                    f"Encountered expressions with identical names being loaded, '{name}'."
 37                )
 38            named_expressions[name] = expr
 39
 40    return named_expressions
 41
 42
 43def load_rule_modules(paths: List[Path]) -> List[RuleYamlSpec]:
 44    """
 45    Takes in a list of paths which may either be directories or files containing rule module YAML files,
 46    and processes them into a `list`. For more information on rule modules see
 47    RulemoduleYamlSpec in `sift_py/yaml/rule.py`.
 48    """
 49
 50    rule_modules: List[RuleYamlSpec] = []
 51
 52    def update_rule_modules(rule_module_path: Path):
 53        rule_module = _read_rule_module_yaml(rule_module_path)
 54        rule_modules.extend(rule_module)
 55
 56    for path in paths:
 57        if path.is_dir():
 58            _handle_subdir(path, update_rule_modules)
 59        elif path.is_file():
 60            update_rule_modules(path)
 61
 62    return rule_modules
 63
 64
 65def _read_named_expression_module_yaml(path: Path) -> Dict[str, str]:
 66    with open(path, "r") as f:
 67        named_expressions = cast(Dict[Any, Any], yaml.safe_load(f.read()))
 68
 69        for key, value in named_expressions.items():
 70            if not isinstance(key, str):
 71                raise YamlConfigError(
 72                    f"Expected '{key}' to be a string in named expression module '{path}'."
 73                )
 74            if not isinstance(value, str):
 75                raise YamlConfigError(
 76                    f"Expected expression of '{key}' to be a string in named expression module '{path}'."
 77                )
 78
 79        return cast(Dict[str, str], named_expressions)
 80
 81
 82def _read_rule_module_yaml(path: Path) -> List[RuleYamlSpec]:
 83    with open(path, "r") as f:
 84        module_rules = cast(Dict[Any, Any], yaml.safe_load(f.read()))
 85        rules = module_rules.get("rules")
 86        if not isinstance(rules, list):
 87            raise YamlConfigError(
 88                f"Expected '{rules}' to be a list in rule module yaml: '{path}'"
 89                f"{_type_fqn(RuleYamlSpec)}"
 90            )
 91
 92        for rule in cast(List[Any], rules):
 93            _validate_rule(rule)
 94
 95        return cast(List[RuleYamlSpec], rules)
 96
 97
 98def _validate_rule(val: Any):
 99    rule = cast(Dict[Any, Any], val)
100
101    name = rule.get("name")
102
103    if not isinstance(name, str):
104        raise YamlConfigError._invalid_property(name, "- name", "str", ["rules"])
105
106    channel_references = rule.get("channel_references")
107
108    if channel_references is not None:
109        if not isinstance(channel_references, list):
110            raise YamlConfigError._invalid_property(
111                channel_references,
112                "- channel_references",
113                f"List[Dict[str, {_type_fqn(ChannelConfigYamlSpec)}]]",
114                ["rules"],
115            )
116
117        for channel_reference in cast(List[Any], channel_references):
118            _validate_channel_reference(channel_reference)
119
120    rule_client_key = rule.get("rule_client_key")
121    description = rule.get("description")
122    expression = rule.get("expression")
123    rule_type = rule.get("type")
124    assignee = rule.get("assignee")
125    tags = rule.get("tags")
126    sub_expressions = rule.get("sub_expressions")
127    asset_names = rule.get("asset_names")
128    tag_names = rule.get("tag_names")
129
130    if rule_client_key is not None and not isinstance(rule_client_key, str):
131        raise YamlConfigError._invalid_property(
132            rule_client_key, "- rule_client_key", "str", ["rules"]
133        )
134
135    if description is not None and not isinstance(description, str):
136        raise YamlConfigError._invalid_property(description, "- description", "str", ["rules"])
137
138    if isinstance(expression, dict):
139        expression_name = cast(Dict[Any, Any], expression).get("name")
140
141        if not isinstance(expression_name, str):
142            raise YamlConfigError._invalid_property(
143                expression_name,
144                "name",
145                "str",
146                ["rules", "- expression"],
147            )
148
149    elif not isinstance(expression, str):
150        raise YamlConfigError._invalid_property(
151            expression,
152            "- expression",
153            "<class 'str'> | <class 'dict'>",
154            ["rules"],
155        )
156
157    valid_rule_types = [kind.value for kind in RuleActionAnnotationKind]
158
159    if rule_type not in valid_rule_types:
160        raise YamlConfigError._invalid_property(
161            rule_type,
162            "- type",
163            " | ".join(valid_rule_types),
164            ["rules"],
165        )
166
167    if assignee is not None and not isinstance(assignee, str):
168        raise YamlConfigError._invalid_property(
169            assignee,
170            "- assignee",
171            "str",
172            ["rules"],
173        )
174
175    if tags is not None and not isinstance(tags, list):
176        raise YamlConfigError._invalid_property(
177            tags,
178            "- tags",
179            "List[str]",
180            ["rules"],
181        )
182
183    if sub_expressions is not None:
184        if not isinstance(channel_references, list):
185            raise YamlConfigError._invalid_property(
186                channel_references,
187                "- sub_expressions",
188                "List[Dict[str, List[Dict[str, str]]]]",
189                ["rules"],
190            )
191
192        for sub_expression in cast(List[Any], sub_expressions):
193            _validate_sub_expression(sub_expression)
194
195    if asset_names is not None and not isinstance(asset_names, list):
196        raise YamlConfigError._invalid_property(
197            asset_names,
198            "- asset_names",
199            "List[str]",
200            ["rules"],
201        )
202
203    if tag_names is not None and not isinstance(tag_names, list):
204        raise YamlConfigError._invalid_property(
205            tag_names,
206            "- tag_names",
207            "List[str]",
208            ["rules"],
209        )
210
211
212def _validate_sub_expression(val: Any):
213    sub_expression = cast(Dict[Any, Any], val)
214
215    for key in sub_expression.keys():
216        if not isinstance(key, str):
217            raise YamlConfigError._invalid_property(
218                sub_expression,
219                "- <str>",
220                "Dict[str, Any]",
221                ["rules", "- sub_expressions"],
222            )
223
224        if _SUB_EXPRESSION_REGEX.match(key) is None:
225            raise YamlConfigError(
226                f"Invalid sub-expression key, '{key}'. Characters must be in the character set [a-zA-Z_] and prefixed with a '$'."
227            )
228
229
230class RuleModuleYamlSpec(TypedDict):
231    """
232    The formal definition of what a rule module looks like in YAML.
233
234    `rules`: A list of rules that belong to the module.
235    """
236
237    rules: List[RuleYamlSpec]
238
239
240class RuleYamlSpec(TypedDict):
241    """
242    The formal definition of what a single rule looks like in YAML.
243
244    `name`: Name of the rule.
245    `rule_client_key`: User-defined string-key that uniquely identifies this rule config.
246    `description`: Description of rule.
247    `expression`:
248        Either an expression-string or a `sift_py.ingestion.config.yaml.spec.NamedExpressionYamlSpec` referencing a named expression.
249    `type`: Determines the action to perform if a rule gets evaluated to true.
250    `assignee`: If `type` is `review`, determines who to notify. Expects an email.
251    `tags`: Tags to associate with the rule.
252    `channel_references`: A list of channel references that maps to an actual channel. More below.
253    `sub_expressions`: A list of sub-expressions which is a mapping of place-holders to sub-expressions. Only used if using named expressions.
254    `asset_names`: A list of asset names that this rule should be applied to. ONLY VALID if defining rules outside of a telemetry config.
255    `tag_names`: A list of tag names that this rule should be applied to. ONLY VALID if defining rules outside of a telemetry config.
256
257    Channel references:
258    A channel reference is a string containing a numerical value prefixed with "$". Examples include "$1", "$2", "$11", and so on.
259    The channel reference is mapped to an actual channel config. In YAML it would look something like this:
260
261    ```yaml
262    channel_references:
263      - $1: *vehicle_state_channel
264      - $2: *voltage_channel
265    ```
266
267    Sub-expressions:
268    A sub-expression is made up of two components: A reference and the actual sub-expression. The sub-expression reference is
269    a string with a "$" prepended to another string comprised of characters in the following character set: `[a-zA-Z0-9_]`.
270    This reference should be mapped to the actual sub-expression. For example, say you have kinematic equations in `kinematics.yml`,
271    and the equation you're interested in using looks like the following:
272
273    ```yaml
274    kinetic_energy_gt:
275      0.5 * $mass * $1 * $1 > $threshold
276    ```
277
278    To properly use `kinetic_energy_gt` in your rule, it would look like the following:
279
280    ```yaml
281    rules:
282      - name: kinetic_energy
283        description: Tracks high energy output while in motion
284        type: review
285        assignee: bob@example.com
286        expression:
287          name: kinetic_energy_gt
288        channel_references:
289          - $1: *velocity_channel
290        sub_expressions:
291          - $mass: 10
292          - $threshold: 470
293        tags:
294            - nostromo
295    ```
296    """
297
298    name: str
299    rule_client_key: NotRequired[str]
300    description: NotRequired[str]
301    expression: Union[str, NamedExpressionYamlSpec]
302    type: Union[Literal["phase"], Literal["review"]]
303    assignee: NotRequired[str]
304    tags: NotRequired[List[str]]
305    channel_references: NotRequired[List[Dict[str, ChannelConfigYamlSpec]]]
306    sub_expressions: NotRequired[List[Dict[str, str]]]
307    asset_names: NotRequired[List[str]]
308    tag_names: NotRequired[List[str]]
309
310
311class NamedExpressionYamlSpec(TypedDict):
312    """
313    A named expression. This class is the formal definition of what a named expression
314    should look like in YAML. The value of `name` may contain a mix of channel references
315    and channel identifiers.
316
317    For a formal definition of channel references and channel identifiers see the following:
318    `sift_py.ingestion.config.yaml.spec.RuleYamlSpec`.
319    """
320
321    name: str