sift_py.calculated_channels.config

  1from __future__ import annotations
  2
  3from typing import List, Optional, Union
  4
  5from pydantic import BaseModel, ConfigDict, field_validator, model_validator
  6from typing_extensions import NotRequired, TypedDict
  7
  8from sift_py.rule.config import (
  9    ExpressionChannelReference,
 10    ExpressionChannelReferenceChannelConfig,
 11    _channel_references_from_dicts,
 12)
 13
 14
 15class CalculatedChannelConfig(BaseModel):
 16    """
 17    Configuration for a calculated channel.
 18
 19    - `name`: Name of the calculated channel.
 20    - `description`: Description of the calculated channel.
 21    - `expression`: The expression to calculate the value of the calculated channel.
 22    - `channel_references`: A list of channel references that are used in the expression. Each reference can either
 23      be an `ExpressionChannelReference` or `ExpressionChannelReferenceChannelConfig`.
 24    - `units`: Units of the calculated channel.
 25    - `client_key`: A user defined string that uniquely identifies the calculated channel. Name is unique, but may be changed.
 26    - `asset_names`: A list of asset names to make the calculation available for.
 27    - `tags`: A list of tags on assets to make the calculation available for. NOTE: Not yet supported.
 28    - `all_assets`: A flag that, when set to `True`, associates the calculated channel with all assets.
 29    """
 30
 31    model_config = ConfigDict(arbitrary_types_allowed=True)
 32
 33    name: str
 34    description: str = ""
 35    expression: str
 36    channel_references: List[
 37        Union[ExpressionChannelReference, ExpressionChannelReferenceChannelConfig]
 38    ]
 39    units: Optional[str] = None
 40    calculated_channel_id: Optional[str] = None
 41    client_key: Optional[str] = None
 42    asset_names: Optional[List[str]] = None
 43    tag_names: Optional[List[str]] = None
 44    all_assets: bool = False
 45
 46    @field_validator("tag_names", mode="before")
 47    @classmethod
 48    def check_for_unsupported_tags(cls, v):
 49        if v:
 50            raise ValueError("`tag_names` is not yet supported.")
 51        return v
 52
 53    @field_validator("channel_references", mode="before")
 54    @classmethod
 55    def convert_channel_references(
 56        cls, raw: List[Union[ExpressionChannelReference, ExpressionChannelReferenceChannelConfig]]
 57    ) -> List[ExpressionChannelReference]:
 58        if not isinstance(raw, list):
 59            raise ValueError("`channel_references` must be a list.")
 60
 61        for ref in raw:
 62            if not isinstance(ref, dict) or (
 63                "channel_identifier" not in ref and "channel_config" not in ref
 64            ):
 65                raise ValueError(
 66                    "`channel_references` must be a list of `ExpressionChannelReference` or `ExpressionChannelReferenceChannelConfig`."
 67                )
 68
 69        return _channel_references_from_dicts(raw)
 70
 71    @model_validator(mode="after")
 72    def validate_assets(self):
 73        if not self.asset_names and not self.tag_names and not self.all_assets:
 74            raise ValueError(
 75                "At least one of `asset_names`, `tag_names` must be specified or `all_assets` must be set to `True`."
 76            )
 77        if self.all_assets and (self.asset_names or self.tag_names):
 78            raise ValueError(
 79                "`all_assets` cannot be `True` if `asset_names` or `tag_names` are specified."
 80            )
 81        return self
 82
 83
 84class CalculatedChannelUpdate(TypedDict):
 85    """
 86    Represents a dictionary for updating properties of a calculated channel. All fields are optional
 87    and only the provided fields will be updated.
 88
 89    - `name`: Updated name of the calculated channel.
 90    - `description`: Updated description of the calculated channel.
 91    - `units`: String representing the units for the calculated channel.
 92    - `expression`: Updated expression used to calculate channel values.
 93    - `channel_references`: A list of channel references which can either be `ExpressionChannelReference`
 94       or `ExpressionChannelReferenceChannelConfig` used in the expression.
 95    - `asset_names`: List of assets associated with the calculation.
 96    - `tags`: List of tags for associating the calculated channel to assets.
 97    - `all_assets`: Boolean flag indicating if the calculated channel applies to all assets.
 98    - `archived`: Boolean flag indicating if the calculated channel is archived.
 99    """
100
101    name: NotRequired[str]
102    description: NotRequired[str]
103    units: NotRequired[str]
104    expression: NotRequired[str]
105    channel_references: NotRequired[
106        List[Union[ExpressionChannelReference, ExpressionChannelReferenceChannelConfig]]
107    ]
108    asset_names: NotRequired[List[str]]
109    tag_names: NotRequired[List[str]]
110    all_assets: NotRequired[bool]
111    archived: NotRequired[bool]