downtime_tracking

downtime_tracking ¤

Downtime tracking by shift and reason.

Essential module for daily downtime analysis: - Downtime by shift - Downtime by reason code - Availability calculations - Downtime trends

DowntimeEvents ¤

DowntimeEvents(
    dataframe: DataFrame,
    *,
    time_column: str = "systime",
    shift_definitions: (
        dict[str, tuple[str, str]] | None
    ) = None,
    specs: Mapping[str, SignalSpec] | None = None
)

Bases: Base

Track machine downtimes by shift and reason.

Each UUID represents one signal: - run_state_uuid: machine state (running/stopped/idle) - reason_uuid: downtime reason code (optional)

Merge keys: [date, shift] for shift-level, [period] for trend data.

Run state comes in one of two forms, and both give the same result:

a string state signal in value_string (running_value="Running", stopped_value="Stopped"), or
a boolean run signal in value_bool (True = running). It is used when the state rows carry value_bool but no value_string, or when running_value / stopped_value is a bool (then matched as given).

Durations are reported in seconds (*_seconds); the *_minutes columns are kept as a convenience.

Pipeline example::

downtime = DowntimeEvents(df)
shift_dt = downtime.downtime_by_shift('machine_state')
# → merge with QualityEvents.nok_by_shift() on [date, shift]
# → merge with ShiftReportEvents.shift_production() on [date, shift]
# → feed into ShiftHandoverEvents.from_shift_data()

Example

import ts_shape df = ts_shape.make_id_signal("state", ["Running", "Stopped", "Running"], ... hold=20, freq="1min", start="2025-01-01 08:00") out = ts_shape.DowntimeEvents(df).downtime_by_shift("state") out[["date", "shift", "downtime_minutes", "availability_pct"]].round(1) date shift downtime_minutes availability_pct 0 2025-01-01 shift_1 20.0 66.1

Initialize downtime tracker.

Parameters:

Name	Type	Description	Default
`dataframe`	`DataFrame`	Input DataFrame with timeseries data	required
`time_column`	`str`	Name of timestamp column (default: 'systime')	`'systime'`
`shift_definitions`	`dict[str, tuple[str, str]] \| None`	Dictionary mapping shift names to (start, end) times Default: 3-shift operation (06:00-14:00, 14:00-22:00, 22:00-06:00)	`None`
`specs`	`Mapping[str, SignalSpec] \| None`	Optional signal semantics (see :mod:`ts_shape.signals`). A state signal's `max_gap` then caps durations by default.	`None`

downtime_by_shift ¤

downtime_by_shift(
    run_state_uuid: str,
    *,
    running_value: str | bool = "Running",
    value_column: str = "value_string",
    max_gap: str | None = None,
    production_day: bool = True
) -> pd.DataFrame

Calculate downtime duration per shift.

Each reading's state holds until the next reading, and that interval is split at the shift boundaries (the ISO 22400-2 time model): a stop from 07:00 to 15:00 is 7 h of downtime in the 06:00-14:00 shift and 1 h in the 14:00-22:00 shift. A shift is reported when any observed time falls in it.

Parameters:

Name	Type	Description	Default
`run_state_uuid`	`str`	UUID for machine state signal	required
`running_value`	`str \| bool`	Value that indicates machine is running. A bool matches a boolean run signal (`value_bool` by default); a boolean run signal is also used, with True as running, when the state rows carry no `value_column` value.	`'Running'`
`value_column`	`str`	Column containing state values	`'value_string'`
`max_gap`	`str \| None`	Optional cap (e.g. "15min"). A duration between two consecutive readings longer than this is treated as a logging gap ("no data") and excluded, instead of being attributed to the preceding state. Without it, a logger outage between a `Running` sample and the next reading is booked as uptime, inflating availability toward 100 % over a period with no data. `None` (default) uses the state signal's spec `max_gap` when `specs` were given and applies no cap otherwise, so every interval between readings, however long, counts toward the preceding state.	`None`
`production_day`	`bool`	Attribute an overnight shift's post-midnight readings to the shift's start date, so a night shift is one `(date, shift)` row (default). False groups by raw calendar date, which splits a night shift across two rows.	`True`

Returns:

Type	Description
`DataFrame`	DataFrame with downtime by shift:
`DataFrame`	start: Start of the shift window (tz-aware in the data's tz)
`DataFrame`	end: End of the shift window (next day for overnight shifts)
`DataFrame`	duration_seconds: Shift window length in seconds
`DataFrame`	date: Production date
`DataFrame`	shift: Shift name
`DataFrame`	total_minutes / total_seconds: Observed shift time (uptime + downtime)
`DataFrame`	downtime_minutes / downtime_seconds: Downtime duration
`DataFrame`	uptime_minutes / uptime_seconds: Running time
`DataFrame`	availability_pct: Uptime percentage
`DataFrame`	excluded_seconds: Time removed by `max_gap` as "no data" (0 without a cap), split at the shift boundaries like the rest; it is in neither uptime nor downtime
`DataFrame`	The `_minutes` columns are a convenience (`_seconds` / 60).

Example

import ts_shape df = ts_shape.make_id_signal( ... "state", ["Running", "Stopped", "Running"], hold=20, freq="1min", ... start="2025-01-01 08:00") out = ts_shape.DowntimeEvents(df).downtime_by_shift("state") out[["shift", "downtime_minutes", "uptime_minutes", "availability_pct"]] shift downtime_minutes uptime_minutes availability_pct 0 shift_1 20.0 39.0 66.101695

downtime_by_reason ¤

downtime_by_reason(
    run_state_uuid: str,
    reason_uuid: str,
    *,
    stopped_value: str | bool = "Stopped",
    value_column_state: str = "value_string",
    value_column_reason: str = "value_string"
) -> pd.DataFrame

Analyze downtime by reason code.

Parameters:

Name	Type	Description	Default
`run_state_uuid`	`str`	UUID for machine state signal	required
`reason_uuid`	`str`	UUID for downtime reason signal	required
`stopped_value`	`str \| bool`	Value indicating machine is stopped. A bool matches a boolean run signal (`value_bool` by default); a boolean run signal is also used, with False as stopped, when the state rows carry no `value_column_state` value.	`'Stopped'`
`value_column_state`	`str`	Column containing state values	`'value_string'`
`value_column_reason`	`str`	Column containing reason codes	`'value_string'`

Returns:

Type	Description
`DataFrame`	DataFrame with downtime by reason:
`DataFrame`	reason: Reason code
`DataFrame`	occurrences: Number of downtime events
`DataFrame`	total_minutes / total_seconds: Total downtime for this reason
`DataFrame`	avg_minutes / avg_seconds: Average downtime per occurrence
`DataFrame`	pct_of_total: Percentage of total downtime
`DataFrame`	The `_minutes` columns are a convenience (`_seconds` / 60).
`DataFrame`	Canonical twin columns: `share_of_total_pct` (after `pct_of_total`); each
`DataFrame`	holds the same values as the deprecated column right before it, which is
`DataFrame`	removed in the next major release.

Example

import pandas as pd, ts_shape states = ["Running", "Stopped", "Running", "Stopped", "Running"] df = pd.concat([ts_shape.make_id_signal("state", states, hold=10, freq="1min"), ... ts_shape.make_id_signal("reason", ["-", "Jam", "-", "Tool", "-"], hold=10, ... freq="1min")]) out = ts_shape.DowntimeEvents(df).downtime_by_reason("state", "reason") out[["reason", "occurrences", "total_minutes", "share_of_total_pct"]] reason occurrences total_minutes share_of_total_pct 0 Jam 10 10.0 50.0 1 Tool 10 10.0 50.0

top_downtime_reasons ¤

top_downtime_reasons(
    run_state_uuid: str,
    reason_uuid: str,
    *,
    top_n: int = 5,
    stopped_value: str | bool = "Stopped",
    value_column_state: str = "value_string",
    value_column_reason: str = "value_string"
) -> pd.DataFrame

Get top N downtime reasons (Pareto analysis).

Parameters:

Name	Type	Description	Default
`run_state_uuid`	`str`	UUID for machine state signal	required
`reason_uuid`	`str`	UUID for downtime reason signal	required
`top_n`	`int`	Number of top reasons to return	`5`
`stopped_value`	`str \| bool`	Value indicating machine is stopped (a string, or a bool for a boolean run signal; see :meth:`downtime_by_reason`)	`'Stopped'`
`value_column_state`	`str`	Column containing state values	`'value_string'`
`value_column_reason`	`str`	Column containing reason codes	`'value_string'`

Returns:

Type	Description
`DataFrame`	DataFrame with top N reasons and cumulative percentage: reason,
`DataFrame`	total_minutes, total_seconds, pct_of_total, cumulative_pct
`DataFrame`	Canonical twin columns: `share_of_total_pct` (after `pct_of_total`); each
`DataFrame`	holds the same values as the deprecated column right before it, which is
`DataFrame`	removed in the next major release.

Example

import pandas as pd, ts_shape states = ["Running", "Stopped", "Running", "Stopped", "Running"] df = pd.concat([ts_shape.make_id_signal("state", states, hold=10, freq="1min"), ... ts_shape.make_id_signal("reason", ["-", "Jam", "-", "Tool", "-"], hold=10, ... freq="1min")]) top = ts_shape.DowntimeEvents(df).top_downtime_reasons("state", "reason", top_n=1) top[["reason", "total_minutes", "cumulative_pct"]] reason total_minutes cumulative_pct 0 Jam 10.0 50.0

availability_trend ¤

availability_trend(
    run_state_uuid: str,
    *,
    running_value: str | bool = "Running",
    value_column: str = "value_string",
    window: str = "1D",
    max_gap: str | None = None
) -> pd.DataFrame

Calculate availability trend over time.

Each reading's state holds until the next reading, and that interval is split at the window boundaries (the ISO 22400-2 time model), so every window holds exactly the up / down time that fell inside it. A window is reported when any observed time falls in it.

Parameters:

Name	Type	Description	Default
`run_state_uuid`	`str`	UUID for machine state signal	required
`running_value`	`str \| bool`	Value that indicates machine is running (a string, or a bool for a boolean run signal; see :meth:`downtime_by_shift`)	`'Running'`
`value_column`	`str`	Column containing state values	`'value_string'`
`window`	`str`	Time window for aggregation (e.g., '1D', '1W')	`'1D'`
`max_gap`	`str \| None`	Optional cap (e.g. "15min"). A gap between consecutive readings longer than this is treated as "no data" and excluded, instead of being attributed to the preceding state (which would inflate availability across a logging outage). `None` (default) applies no cap: every interval between readings, however long, counts toward the preceding state.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with availability trend:
`DataFrame`	start: Period start
`DataFrame`	end: Period end (start + window)
`DataFrame`	duration_seconds: Period length in seconds
`DataFrame`	period: Time period
`DataFrame`	availability_pct: Availability percentage
`DataFrame`	uptime_minutes / uptime_seconds: Total uptime
`DataFrame`	downtime_minutes / downtime_seconds: Total downtime
`DataFrame`	excluded_seconds: Time removed by `max_gap` as "no data" (0 without a cap), split at the window boundaries
`DataFrame`	The `_minutes` columns are a convenience (`_seconds` / 60).

Example

import ts_shape df = ts_shape.make_id_signal( ... "state", ["Running", "Stopped", "Running"], hold=20, freq="1min") out = ts_shape.DowntimeEvents(df).availability_trend("state", window="30min") out[["period", "availability_pct", "downtime_minutes"]] period availability_pct downtime_minutes 0 2025-01-01 00:00:00 66.666667 10.0 1 2025-01-01 00:30:00 65.517241 10.0