dp3.common.attrspec

AttrType

Bases: Flag

Enum of attribute types

PLAIN = 1 OBSERVATIONS = 2 TIMESERIES = 4

from_str classmethod

from_str(type_str: str)

Convert string representation like "plain" to AttrType.

Source code in dp3/common/attrspec.py

@classmethod
def from_str(cls, type_str: str):
    """
    Convert string representation like "plain" to AttrType.
    """
    try:
        return cls(cls[type_str.upper()])
    except Exception as e:
        raise ValueError(f"Invalid `type` of attribute '{type_str}'") from e

ObservationsHistoryParams

Bases: BaseModel

History parameters field of observations attribute

TimeseriesTSParams

Bases: BaseModel

Timeseries parameters field of timeseries attribute

TimeseriesSeries

Bases: BaseModel

Series of timeseries attribute

AttrSpecGeneric

Bases: SpecModel

Base of attribute specification

Parent of other AttrSpec classes.

Attributes:

Name	Type	Description
ttl	Optional[ParsedTimedelta]	Optional extension of TTL of the entity - will be ignored if lifetime setting does not match.

AttrSpecClassic

Bases: AttrSpecGeneric

Parent of non-timeseries AttrSpec classes.

is_iterable property

is_iterable: bool

Returns whether specified attribute is iterable.

element_type property

element_type: DataType

Returns the element type for iterable data types.

is_relation property

is_relation: bool

Returns whether specified attribute is a link.

relation_to property

relation_to: str

Returns linked entity id. Raises ValueError if attribute is not a link.

is_mirrored property

is_mirrored: bool

Returns whether specified attribute is a mirrored link.

mirror_as property

mirror_as: str

Returns:

Type	Description
str	name of the mirrored attribute.

Raises: ValueError: If attribute is not a mirrored link.

AttrSpecPlain

AttrSpecPlain(**data)

Bases: AttrSpecClassic

Plain attribute specification

Source code in dp3/common/attrspec.py

def __init__(self, **data):
    super().__init__(**data)

    entity_spec = get_entity_context()["self"]
    self._dp_model = create_model(
        f"DataPointPlain_{self.id}",
        __base__=DataPointPlainBase,
        eid=(entity_spec.id_data_type.data_type, ...),
        v=(self.data_type.data_type, ...),
    )

AttrSpecReadOnly

AttrSpecReadOnly(**data)

Bases: AttrSpecPlain

Read-only plain attribute specification. Used for internal attributes.

Source code in dp3/common/attrspec.py

def __init__(self, **data):
    super().__init__(**data)

    entity_spec = get_entity_context()["self"]
    self._dp_model = create_model(
        f"DataPointReadOnly_{self.id}",
        __base__=DataPointPlainBase,
        eid=(entity_spec.id_data_type.data_type, ...),
        v=(ReadOnly, ...),
    )

AttrSpecObservations

AttrSpecObservations(**data)

Bases: AttrSpecClassic

Observations attribute specification

Source code in dp3/common/attrspec.py

def __init__(self, **data):
    super().__init__(**data)

    value_validator = self.data_type.data_type
    entity_spec = get_entity_context()["self"]
    self._dp_model = create_model(
        f"DataPointObservations_{self.id}",
        __base__=DataPointObservationsBase,
        eid=(entity_spec.id_data_type.data_type, ...),
        v=(value_validator, ...),
    )

AttrSpecTimeseries

AttrSpecTimeseries(**data)

Bases: AttrSpecGeneric

Timeseries attribute specification

Source code in dp3/common/attrspec.py

def __init__(self, **data):
    super().__init__(**data)

    entity_spec = get_entity_context()["self"]

    # Typing of `v` field
    dp_value_typing = {}
    for s in self.series:
        data_type = self.series[s].data_type.data_type
        dp_value_typing[s] = ((list[data_type]), ...)

    # Add root validator
    if self.timeseries_type == "regular":
        root_validator = dp_ts_root_validator_regular_wrapper(self.timeseries_params.time_step)
    elif self.timeseries_type == "irregular":
        root_validator = dp_ts_root_validator_irregular
    elif self.timeseries_type == "irregular_intervals":
        root_validator = dp_ts_root_validator_irregular_intervals
    else:
        raise ValueError(f"Unknown timeseries type '{self.timeseries_type}'")

    # Validators
    dp_validators = {
        "v_validator": field_validator("v")(dp_ts_v_validator),
        "root_validator": model_validator(mode="after")(root_validator),
    }

    self._dp_model = create_model(
        f"DataPointTimeseries_{self.id}",
        __base__=DataPointTimeseriesBase,
        __validators__=dp_validators,
        eid=(entity_spec.id_data_type.data_type, ...),
        v=(create_model(f"DataPointTimeseriesValue_{self.id}", **dp_value_typing), ...),
    )

AttrSpec

AttrSpec(id: str, spec: dict[str, Any]) -> AttrSpecType

Factory for AttrSpec classes

Source code in dp3/common/attrspec.py

def AttrSpec(id: str, spec: dict[str, Any]) -> AttrSpecType:
    """Factory for `AttrSpec` classes"""

    assert isinstance(spec, dict), "Attribute specification must be a dict"
    if "type" not in spec:
        raise ValueError("Missing mandatory attribute `type`")
    attr_type = AttrType.from_str(spec.get("type"))
    subclasses = {
        AttrType.PLAIN: AttrSpecPlain,
        AttrType.OBSERVATIONS: AttrSpecObservations,
        AttrType.TIMESERIES: AttrSpecTimeseries,
    }

    spec["id"] = id  # Fill in the id if not present, else overwrite
    return subclasses[attr_type](**spec)