Index

signals

Signal semantics: specs, inference and spec-driven primitives.

Describe each signal once -- what kind of value it carries, how it is encoded, how often it is sampled and when a gap means "no data" -- and let the primitives handle counters, intervals, time weighting and alignment correctly::

from ts_shape.signals import infer_specs

specs = infer_specs(df, overrides={"line1:parts": {"unit": "pcs"}})
specs.to_frame()

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

infer_spec

infer_spec(
    group: DataFrame,
    uuid: str,
    time_column: str = "systime",
) -> SignalSpec

Infer the spec of one signal from its rows.

Parameters:

Name	Type	Description	Default
group	DataFrame	The rows of one signal.	required
uuid	str	The signal's uuid, written to the spec.	required
time_column	str	Timestamp column.	'systime'

Returns:

Type	Description
SignalSpec	The inferred :class:SignalSpec; notes explain each field.

Raises:

Type	Description
ColumnNotFoundError	When time_column is missing.

infer_specs

infer_specs(
    dataframe: DataFrame,
    overrides: (
        Mapping[str, Mapping[str, Any]] | None
    ) = None,
    *,
    uuid_column: str = "uuid",
    time_column: str = "systime"
) -> SignalSpecs

Infer a :class:SignalSpec for every uuid in a long-format frame.

Parameters:

Name	Type	Description	Default
dataframe	DataFrame	Long-format data (systime, uuid, value_*).	required
overrides	Mapping[str, Mapping[str, Any]] | None	{uuid: {field: value}} applied after inference -- the place for what the data cannot tell (units, outage limits).	None
uuid_column	str	Column identifying the signal.	'uuid'
time_column	str	Timestamp column.	'systime'

Returns:

Type	Description
SignalSpecs	class:SignalSpecs keyed by uuid (as text); each spec's notes
SignalSpecs	explain the inferred fields.

Raises:

Type	Description
ColumnNotFoundError	When uuid_column or time_column is missing.