refactor: Deprecate RuleChannelReferences

python/examples/ingestion_with_yaml_config/main.py

@@ -5,7 +5,7 @@
 from dotenv import load_dotenv
 from sift_py.grpc.transport import SiftChannelConfig, use_sift_channel
 from sift_py.ingestion.service import IngestionService
-from sift_py.rule.service import RuleChannelReference, RuleService
+from sift_py.rule.service import RuleService
 from sift_py.yaml.rule import load_named_expression_modules
 from simulator import Simulator
 from telemetry_config import nostromos_lv_426
@@ -61,10 +61,6 @@
                 RULE_MODULES_DIR.joinpath("nostromo.yml"),
             ],
             named_expressions=named_expressions,
-            channel_references=[
-                RuleChannelReference("overvoltage", {"$1": "vehicle_state", "$2": "voltage"}),
-                RuleChannelReference("undervoltage", {"$1": "vehicle_state", "$2": "voltage"}),
-            ],
         )
 
         # Create an optional run as part of this ingestion

python/examples/ingestion_with_yaml_config/rule_modules/voltage.yml

@@ -4,6 +4,9 @@ rules:
     expression: $1 == "Accelerating" && $2 > 80
     rule_client_key: overvoltage-rule
     type: review
+    channel_references:
+      - $1: vehicle_state
+      - $2: voltage
     asset_names:
       - NostromoLV426
 
@@ -12,5 +15,8 @@ rules:
     expression: $1 == "Accelerating" && $2 < 40
     rule_client_key: undervoltage-rule
     type: review
+    channel_references:
+      - $1: vehicle_state
+      - $2: voltage
     asset_names:
       - NostromoLV426

python/lib/sift_py/rule/service.py

@@ -64,14 +64,9 @@ def load_rules_from_yaml(
         self,
         paths: List[Path],
         named_expressions: Optional[Dict[str, str]] = None,
-        channel_references: Optional[List[RuleChannelReference]] = None,
     ) -> List[RuleConfig]:
         """
         Loads rules from a YAML spec, and creates or updates the rules in the Sift API.
-        If the rule does not contain channel references in its YAML definition, provide a dict of rule names mapped
-        to a list of channel references. Otherwise if the YAML definition contains channel references, the `channel_references_map`
-        should be omitted. If channel references are present in both the YAML definition and provided in the `channel_references_map`,
-        or if neither are provided for a given rule, an exception will be thrown.
         For more on rule YAML definitions, see `sift_py.ingestion.config.yaml.spec.RuleYamlSpec`.
         """
         module_rules = load_rule_modules(paths)
@@ -80,26 +75,14 @@ def load_rules_from_yaml(
         for rule_yaml in module_rules:
             rule_name = rule_yaml["name"]
 
-            # First validate channel references
+            # First parse channel references
             yaml_channel_references = rule_yaml.get("channel_references", [])
-            arg_channel_references: Dict[str, Any] = {}
-
-            if channel_references:
-                for rule_channel_refs in channel_references:
-                    if rule_channel_refs.rule_name == rule_name:
-                        arg_channel_references = rule_channel_refs.channel_references
-
-            if yaml_channel_references and arg_channel_references:
-                raise ValueError(
-                    f"Rule of name '{rule_yaml['name']}' cannot have both YAML and channel_references argument provided. "
-                    "Please provide only one or the other."
-                )
 
             rule_channel_references: List[
                 Union[ExpressionChannelReference, ExpressionChannelReferenceChannelConfig]
             ] = []
 
-            def parse_channel_refs(channel_ref: Dict[str, Any]):
+            for channel_ref in yaml_channel_references:
                 for ref, channel_config in channel_ref.items():
                     if isinstance(channel_config, dict):
                         name = channel_config.get("name", "")
@@ -125,12 +108,6 @@ def parse_channel_refs(channel_ref: Dict[str, Any]):
                         }
                     )
 
-            if yaml_channel_references:
-                for channel_ref in yaml_channel_references:
-                    parse_channel_refs(channel_ref)
-            elif arg_channel_references:
-                parse_channel_refs(arg_channel_references)
-
             if not rule_channel_references:
                 raise ValueError(f"Rule of name '{rule_yaml['name']}' requires channel_references")