Skip to content

spec

spec ¤

Signal semantics: what a uuid's values mean.

Every analysis on industrial time series silently depends on a few facts about each signal:

  • kind -- is it a continuous measurement, a cumulative counter, a state (run/idle, recipe, alarm), or a sparse event stream?
  • encoding -- is it sampled periodically, or change-encoded (one row per change, the is_delta convention)?
  • timing -- its nominal sampling period and the gap after which the signal must be treated as "no data" rather than "value held".

Getting any of these wrong produces silently wrong results: counters that lose production at a reset, intervals that collapse to zero length on change-encoded data, averages biased by sampling density, durations that bridge logging outages. :class:SignalSpec records these facts once per uuid so that the primitives in :mod:ts_shape.signals.primitives (and the detectors built on them) get them right by construction.

SignalSpec dataclass ¤

SignalSpec(
    uuid: str,
    kind: Kind = "continuous",
    encoding: Encoding = "sampled",
    value_column: str = "value_double",
    unit: str | None = None,
    nominal_period: Timedelta | None = None,
    max_gap: Timedelta | None = None,
    tz: str | None = None,
    reset_policy: Literal[
        "reset_to_zero", "none"
    ] = "reset_to_zero",
    notes: dict[str, str] = dict(),
)

Semantics of one signal (one uuid).

Attributes:

Name Type Description
uuid str

The signal identifier.
kind Kind

"continuous" (a measured level), "counter" (cumulative, may reset), "state" (a held discrete value) or "event" (sparse occurrences without a held value).
encoding Encoding

"sampled" (rows at a regular period, values repeat) or "change_encoded" (one row per change; a value holds until the next row).
value_column str

The value_* column that carries the signal.
unit str | None

Engineering unit, informational (e.g. "kWh", "pcs").
nominal_period Timedelta | None

Typical time between samples.
max_gap Timedelta | None

Longest interval across which a value may be held. Longer spans are "no data". None means never cap.
tz str | None

Timezone name of systime (None for naive timestamps).
reset_policy Literal['reset_to_zero', 'none']

For counters: "reset_to_zero" (a drop restarts the count; the default) or "none" (decreases are real, e.g. a level that is not cumulative).
notes dict[str, str]

Free-form provenance, e.g. why a field was inferred.
Example

from ts_shape import SignalSpec spec = SignalSpec("line1:parts", kind="counter", ... value_column="value_integer", unit="pcs") spec.kind, spec.reset_policy ('counter', 'reset_to_zero') spec.with_(unit="parts").unit 'parts' SignalSpec.from_dict(spec.to_dict()) == spec True

is_change_encoded property ¤

is_change_encoded: bool

Whether rows are emitted only when the value changes.

holds_value property ¤

holds_value: bool

Whether a value persists until the next row (state / counter / level).

interval_end property ¤

interval_end: str

interval_end mode for :func:ts_shape.utils.intervals.state_intervals.

A held value lasts until the next change, for both encodings: on sampled data closing at the last sample undercounts every interval by one period, on change-encoded data it collapses them to zero.

with_ ¤

with_(**changes: Any) -> SignalSpec

Return a copy with changes applied (validated).

Parameters:

Name Type Description Default
**changes Any

New field values.

 {}

Returns:

Type Description
SignalSpec

The new spec.

Raises:

Type Description
ValueError

For unknown field names or invalid values.

to_dict ¤

to_dict() -> dict[str, Any]

Plain, serializable mapping (Timedeltas as strings).

from_dict classmethod ¤

from_dict(data: Mapping[str, Any]) -> SignalSpec

Build a spec from a plain mapping of field names to values.

Parameters:

Name Type Description Default
data Mapping[str, Any]

Mapping of SignalSpec field names to values, e.g. the output of to_dict.

 required

Returns:

Type Description
SignalSpec

The validated SignalSpec.

Raises:

Type Description
ValueError

If data contains unknown field names or invalid field values.

SignalSpecs ¤

SignalSpecs(specs: Mapping[str, SignalSpec] | None = None)

Bases: Mapping[str, SignalSpec]

A read-only collection of :class:SignalSpec, keyed by uuid.

Build it with :func:ts_shape.signals.infer_specs, then override what the data cannot tell::

specs = infer_specs(df).override({
    "line1:parts": {"kind": "counter", "unit": "pcs"},
    "line1:state": {"max_gap": "15min"},
})

override ¤

override(
    changes: Mapping[str, Mapping[str, Any]],
) -> SignalSpecs

Return new specs with per-uuid field overrides applied.

A uuid not yet present is added with the given fields.

Parameters:

Name Type Description Default
changes Mapping[str, Mapping[str, Any]]

Mapping of uuid to {field: value}; the fields replace those of the existing spec (see SignalSpec.with_).

 required

Returns:

Name Type Description
New SignalSpecs

class:SignalSpecs.

Raises:

Type Description
ValueError

For unknown field names or invalid values.

to_frame ¤

to_frame() -> pd.DataFrame

One row per signal, columns = spec fields (notes excluded).

to_dict ¤

to_dict() -> dict[str, dict[str, Any]]

Serialize to {uuid: {field: value}} (the YAML/JSON layout).

The uuid key is omitted from each inner mapping and notes are excluded.

Returns:

Type Description
dict[str, dict[str, Any]]

Mapping of uuid to that signal's spec fields.

from_dict classmethod ¤

from_dict(
    data: Mapping[str, Mapping[str, Any]],
) -> SignalSpecs

Build from {uuid: {field: value}} (the YAML/JSON layout).

Parameters:

Name Type Description Default
data Mapping[str, Mapping[str, Any]]

Mapping of uuid to {field: value}, e.g. the output of to_dict. Field names must be SignalSpec fields.

 required

from_yaml classmethod ¤

from_yaml(path: str | Path) -> SignalSpecs

Load {uuid: {field: value}} from a YAML file (needs PyYAML).

Parameters:

Name Type Description Default
path str | Path

Path to a YAML file with the {uuid: {field: value}} layout; an empty file gives empty specs.

 required