cycle_time_tracking

cycle_time_tracking ¤

Cycle time tracking by part number.

Simple, practical module for cycle time analysis: - Cycle time per part number - Statistics (min, avg, max, std) - Slow cycle detection - Trend analysis

CycleTimeEvents ¤

CycleTimeEvents(
    dataframe: DataFrame,
    *,
    time_column: str = "systime",
    event_uuid: str = "prod:cycle_time"
)

Bases: Base

Track cycle times by part number.

Each UUID represents one signal: - part_id_uuid: string signal with current part number - cycle_trigger_uuid: boolean/integer signal for cycle completion

Example

import pandas as pd, ts_shape df = pd.concat([ts_shape.make_id_signal("part", ["A", "B"], hold=30), ... ts_shape.make_timeseries(["cycle"], n_points=60, value_column="value_bool")]) stats = ts_shape.CycleTimeEvents(df).cycle_time_statistics("part", "cycle") stats[["part_number", "count", "avg_seconds"]].round(1) part_number count avg_seconds 0 A 7 111.4 1 B 5 198.0

Initialize cycle time tracker.

Parameters:

Name	Type	Description	Default
`dataframe`	`DataFrame`	Input DataFrame with timeseries data	required
`time_column`	`str`	Name of timestamp column (default: 'systime')	`'systime'`
`event_uuid`	`str`	UUID tagged on derived cycle events (the `uuid` column of point outputs; default: 'prod:cycle_time')	`'prod:cycle_time'`

cycle_time_by_part ¤

cycle_time_by_part(
    part_id_uuid: str,
    cycle_trigger_uuid: str,
    *,
    value_column_part: str = "value_string",
    value_column_trigger: str = "value_bool",
    max_cycle_gap: str | None = None
) -> pd.DataFrame

Calculate cycle time for each part number.

Parameters:

Name	Type	Description	Default
`part_id_uuid`	`str`	UUID for part number signal	required
`cycle_trigger_uuid`	`str`	UUID for cycle completion trigger	required
`value_column_part`	`str`	Column containing part numbers	`'value_string'`
`value_column_trigger`	`str`	Column containing cycle triggers	`'value_bool'`
`max_cycle_gap`	`str \| None`	Optional cap (e.g. "5min"). Intervals between consecutive completions longer than this are dropped as inter-cycle downtime rather than reported as a cycle. Without it, a stoppage between completions (a break, a jam) becomes one giant "cycle" that inflates max/avg and is falsely flagged as a slow cycle.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with columns:
`DataFrame`	systime: Cycle completion time
`DataFrame`	uuid: Event UUID of this tracker
`DataFrame`	source_uuid: The cycle trigger UUID
`DataFrame`	part_number: Part number/ID
`DataFrame`	cycle_time_seconds: Cycle time in seconds

Example

import pandas as pd, ts_shape df = pd.concat([ts_shape.make_id_signal("part", ["A", "B"], hold=30), ... ts_shape.make_timeseries(["cycle"], n_points=60, value_column="value_bool")]) out = ts_shape.CycleTimeEvents(df).cycle_time_by_part("part", "cycle") out[["part_number", "cycle_time_seconds"]].head(3) part_number cycle_time_seconds 0 A 60.0 1 A 120.0 2 A 120.0

cycle_time_statistics ¤

cycle_time_statistics(
    part_id_uuid: str,
    cycle_trigger_uuid: str,
    *,
    value_column_part: str = "value_string",
    value_column_trigger: str = "value_bool",
    max_cycle_gap: str | None = None
) -> pd.DataFrame

Calculate statistics: min, avg, max, std cycle time by part.

Parameters:

Name	Type	Description	Default
`part_id_uuid`	`str`	UUID for part number signal	required
`cycle_trigger_uuid`	`str`	UUID for cycle completion trigger	required
`value_column_part`	`str`	Column containing part numbers	`'value_string'`
`value_column_trigger`	`str`	Column containing cycle triggers	`'value_bool'`
`max_cycle_gap`	`str \| None`	Optional cap (e.g. "5min") forwarded to :meth:`cycle_time_by_part`: intervals between consecutive completions longer than this are dropped as inter-cycle downtime. Default None keeps every interval.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with statistics per part:
`DataFrame`	part_number: Part number/ID
`DataFrame`	count: Number of cycles
`DataFrame`	min_seconds: Minimum cycle time
`DataFrame`	avg_seconds: Average cycle time
`DataFrame`	max_seconds: Maximum cycle time
`DataFrame`	std_seconds: Standard deviation
`DataFrame`	median_seconds: Median cycle time (robust target)

Example

import pandas as pd, ts_shape df = pd.concat([ts_shape.make_id_signal("part", ["A", "B"], hold=30), ... ts_shape.make_timeseries(["cycle"], n_points=60, value_column="value_bool")]) stats = ts_shape.CycleTimeEvents(df).cycle_time_statistics("part", "cycle") stats[["part_number", "count", "median_seconds", "max_seconds"]] part_number count median_seconds max_seconds 0 A 7 120.0 180.0 1 B 5 150.0 420.0

detect_slow_cycles ¤

detect_slow_cycles(
    part_id_uuid: str,
    cycle_trigger_uuid: str,
    *,
    threshold_factor: float = 1.5,
    value_column_part: str = "value_string",
    value_column_trigger: str = "value_bool",
    max_cycle_gap: str | None = None
) -> pd.DataFrame

Identify cycles that exceed normal time by threshold factor.

Parameters:

Name	Type	Description	Default
`part_id_uuid`	`str`	UUID for part number signal	required
`cycle_trigger_uuid`	`str`	UUID for cycle completion trigger	required
`threshold_factor`	`float`	Cycles slower than median * factor are flagged (default: 1.5)	`1.5`
`value_column_part`	`str`	Column containing part numbers	`'value_string'`
`value_column_trigger`	`str`	Column containing cycle triggers	`'value_bool'`
`max_cycle_gap`	`str \| None`	Optional cap (e.g. "5min") forwarded to :meth:`cycle_time_by_part`: intervals between consecutive completions longer than this are dropped as inter-cycle downtime. Default None keeps every interval.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with slow cycles:
`DataFrame`	systime: Cycle completion time
`DataFrame`	uuid: Event UUID of this tracker
`DataFrame`	source_uuid: The cycle trigger UUID
`DataFrame`	part_number: Part number/ID
`DataFrame`	cycle_time_seconds: Actual cycle time
`DataFrame`	median_seconds: Expected (median) cycle time
`DataFrame`	deviation_factor: How many times slower than expected
`DataFrame`	is_slow: Flag (always True in returned data)

Example

import pandas as pd, ts_shape df = pd.concat([ts_shape.make_id_signal("part", ["A", "B"], hold=30), ... ts_shape.make_timeseries(["cycle"], n_points=60, value_column="value_bool")]) slow = ts_shape.CycleTimeEvents(df).detect_slow_cycles("part", "cycle") slow[["part_number", "cycle_time_seconds", "deviation_factor"]] part_number cycle_time_seconds deviation_factor 0 A 180.0 1.5 1 B 240.0 1.6 2 B 420.0 2.8

cycle_time_trend ¤

cycle_time_trend(
    part_id_uuid: str,
    cycle_trigger_uuid: str,
    part_number: str,
    *,
    window_size: int = 20,
    value_column_part: str = "value_string",
    value_column_trigger: str = "value_bool",
    max_cycle_gap: str | None = None
) -> pd.DataFrame

Analyze cycle time trends for a specific part.

Parameters:

Name	Type	Description	Default
`part_id_uuid`	`str`	UUID for part number signal	required
`cycle_trigger_uuid`	`str`	UUID for cycle completion trigger	required
`part_number`	`str`	Specific part number to analyze	required
`window_size`	`int`	Number of cycles for moving average (default: 20)	`20`
`value_column_part`	`str`	Column containing part numbers	`'value_string'`
`value_column_trigger`	`str`	Column containing cycle triggers	`'value_bool'`
`max_cycle_gap`	`str \| None`	Optional cap (e.g. "5min") forwarded to :meth:`cycle_time_by_part`: intervals between consecutive completions longer than this are dropped as inter-cycle downtime. Default None keeps every interval.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with trend data:
`DataFrame`	systime: Cycle completion time
`DataFrame`	cycle_time_seconds: Actual cycle time
`DataFrame`	moving_avg: Moving average cycle time
`DataFrame`	trend: 'improving', 'stable', or 'degrading'

Example

import pandas as pd, ts_shape df = pd.concat([ts_shape.make_id_signal("part", ["A", "B"], hold=30), ... ts_shape.make_timeseries(["cycle"], n_points=60, value_column="value_bool")]) trend = ts_shape.CycleTimeEvents(df).cycle_time_trend( ... "part", "cycle", "A", window_size=3) trend[["cycle_time_seconds", "moving_avg", "trend"]].tail(3) cycle_time_seconds moving_avg trend 4 60.0 120.0 improving 5 120.0 120.0 stable 6 120.0 100.0 improving

hourly_cycle_time_summary ¤

hourly_cycle_time_summary(
    part_id_uuid: str,
    cycle_trigger_uuid: str,
    *,
    value_column_part: str = "value_string",
    value_column_trigger: str = "value_bool",
    max_cycle_gap: str | None = None
) -> pd.DataFrame

Hourly summary of cycle times by part.

Parameters:

Name	Type	Description	Default
`part_id_uuid`	`str`	UUID for part number signal	required
`cycle_trigger_uuid`	`str`	UUID for cycle completion trigger	required
`value_column_part`	`str`	Column containing part numbers	`'value_string'`
`value_column_trigger`	`str`	Column containing cycle triggers	`'value_bool'`
`max_cycle_gap`	`str \| None`	Optional cap (e.g. "5min") forwarded to :meth:`cycle_time_by_part`: intervals between consecutive completions longer than this are dropped as inter-cycle downtime. Default None keeps every interval.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with hourly statistics:
`DataFrame`	hour: Hour timestamp
`DataFrame`	part_number: Part number/ID
`DataFrame`	cycles_completed: Number of cycles
`DataFrame`	avg_cycle_time: Average cycle time
`DataFrame`	min_cycle_time: Fastest cycle
`DataFrame`	max_cycle_time: Slowest cycle

Example

import pandas as pd, ts_shape df = pd.concat([ts_shape.make_id_signal("part", ["A", "B"], hold=30), ... ts_shape.make_timeseries(["cycle"], n_points=60, value_column="value_bool")]) hourly = ts_shape.CycleTimeEvents(df).hourly_cycle_time_summary("part", "cycle") hourly[["part_number", "cycles_completed", "avg_cycle_time"]] part_number cycles_completed avg_cycle_time 0 A 7 111.428571 1 B 5 198.000000