shift_reporting

shift_reporting ¤

Shift-based production reporting.

Simple module for shift summaries: - Production by shift - Shift performance comparison - Shift targets and actuals

ShiftReportEvents ¤

ShiftReportEvents(
    dataframe: DataFrame,
    *,
    time_column: str = "systime",
    shift_definitions: (
        dict[str, tuple[str, str]] | None
    ) = None
)

Bases: Base

Simple shift-based production reporting.

Each UUID represents one signal: - counter_uuid: production counter - part_id_uuid: part number (optional)

Merge keys: [date, shift] for shift-level outputs.

Pipeline example::

reporter = ShiftReportEvents(df)
prod = reporter.shift_production('counter')
# → merge with QualityEvents.nok_by_shift() on [date, shift]
# → merge with DowntimeEvents.downtime_by_shift() on [date, shift]
# → feed combined DataFrame into ShiftHandoverEvents.from_shift_data()

Example usage

reporter = ShiftReportEvents(df, shift_definitions={ "day": ("06:00", "14:00"), "afternoon": ("14:00", "22:00"), "night": ("22:00", "06:00"), })

Production per shift¤

shift_prod = reporter.shift_production( counter_uuid='counter_signal', part_id_uuid='part_number_signal' )

Compare shifts¤

comparison = reporter.shift_comparison(counter_uuid='counter_signal')

Initialize shift reporter.

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`

Example shift_definitions

{ "shift_1": ("06:00", "14:00"), "shift_2": ("14:00", "22:00"), "shift_3": ("22:00", "06:00"), }

shift_production ¤

shift_production(
    counter_uuid: str,
    part_id_uuid: str | None = None,
    *,
    value_column_counter: str = "value_integer",
    value_column_part: str = "value_string",
    date: str | None = None,
    handle_resets: bool = True,
    production_day: bool = True
) -> pd.DataFrame

Production quantity per shift.

The reset-aware increments are computed once over the whole counter signal and each one is booked to the shift of the reading that reports it, so the shift quantities add up to the signal total: the parts made between the last reading of one shift and the first reading of the next are counted in the next shift (in the example, 310 parts over 06:00-21:30 are 150 + 160, not 150 + 150). A shift boundary is also where counters commonly reset (shift change, part change, PLC restart); a decreasing counter is treated as a reset and the post-reset production is counted.

Parameters:

Name	Type	Description	Default
`counter_uuid`	`str`	Production counter UUID	required
`part_id_uuid`	`str \| None`	Part number UUID (optional, for part-specific production)	`None`
`value_column_counter`	`str`	Column containing counter values	`'value_integer'`
`value_column_part`	`str`	Column containing part numbers	`'value_string'`
`date`	`str \| None`	Specific date in 'YYYY-MM-DD' format (optional). Selects the readings of that calendar date; their increments are still computed on the full signal, so the first reading of the day carries the production since the reading before it.	`None`
`handle_resets`	`bool`	When True (default), treat a decreasing counter as a reset and count the production after the reset instead of losing it. The result gains a `resets` column with the number of resets observed in each shift (a drop at the shift's first reading counts in that shift). When False, the quantity is `max(0, last_count - first_count)` of the shift's own readings: a shift containing a reset reports too little and the production between two shifts' readings is lost.	`True`
`production_day`	`bool`	When True, the post-midnight portion of an overnight shift (e.g. `22:00 -> 06:00`) is attributed to the shift's start date, so a night shift is one contiguous `(date, shift)` group instead of being split across two calendar dates (the default). When False, grouping is by raw calendar date. Only affects schedules with an overnight shift.	`True`

Returns:

Type	Description
`DataFrame`	DataFrame with production by shift:
`DataFrame`	start: Start of the shift window (`date` at the shift start time from `shift_definitions`, 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`	part_number: Part number (if part_id_uuid provided)
`DataFrame`	quantity: Parts produced: the reset-aware increments reported by the shift's readings (each since the previous reading)
`DataFrame`	resets: Number of counter resets in the shift (only when `handle_resets=True`)

Example

import ts_shape df = ts_shape.make_timeseries(["counter"], n_points=32, freq="30min", ... start="2025-01-01 06:00", noise=0, drift=310, value_column="value_integer") out = ts_shape.ShiftReportEvents(df).shift_production("counter") out[["date", "shift", "quantity"]] date shift quantity 0 2025-01-01 shift_1 150 1 2025-01-01 shift_2 160

shift_comparison ¤

shift_comparison(
    counter_uuid: str,
    *,
    value_column_counter: str = "value_integer",
    days: int = 7
) -> pd.DataFrame

Compare shift performance over recent days.

Parameters:

Name	Type	Description	Default
`counter_uuid`	`str`	Production counter UUID	required
`value_column_counter`	`str`	Column containing counter values	`'value_integer'`
`days`	`int`	Number of recent days to analyze (default: 7)	`7`

Returns:

Type	Description
`DataFrame`	DataFrame with shift comparison:
`DataFrame`	shift: Shift name
`DataFrame`	avg_quantity: Average production per shift
`DataFrame`	min_quantity: Minimum production
`DataFrame`	max_quantity: Maximum production
`DataFrame`	std_quantity: Standard deviation
`DataFrame`	days_count: Number of days included

Example

import ts_shape df = ts_shape.make_timeseries(["counter"], n_points=96, freq="30min", ... start="2025-01-01 06:00", baseline=10, noise=3, value_column="value_integer") df["value_integer"] = df["value_integer"].cumsum() # running part counter out = ts_shape.ShiftReportEvents(df).shift_comparison("counter") out[["shift", "avg_quantity", "min_quantity", "max_quantity"]] shift avg_quantity min_quantity max_quantity 0 shift_1 150.0 144 156 1 shift_2 163.5 157 170 2 shift_3 156.5 147 166

shift_targets ¤

shift_targets(
    counter_uuid: str,
    targets: dict[str, float],
    *,
    value_column_counter: str = "value_integer",
    date: str | None = None
) -> pd.DataFrame

Compare actual production to shift targets.

Parameters:

Name	Type	Description	Default
`counter_uuid`	`str`	Production counter UUID	required
`targets`	`dict[str, float]`	Dictionary mapping shift names to target quantities	required
`value_column_counter`	`str`	Column containing counter values	`'value_integer'`
`date`	`str \| None`	Specific date in 'YYYY-MM-DD' format (optional)	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with target comparison:
`DataFrame`	date: Production date
`DataFrame`	shift: Shift name
`DataFrame`	actual: Actual production
`DataFrame`	target: Target production
`DataFrame`	variance: Difference (actual - target)
`DataFrame`	achievement_pct: Percentage of target achieved

Example

import ts_shape df = ts_shape.make_timeseries(["counter"], n_points=32, freq="30min", ... start="2025-01-01 06:00", noise=0, drift=310, value_column="value_integer") rep = ts_shape.ShiftReportEvents(df) out = rep.shift_targets("counter", targets={"shift_1": 150, "shift_2": 200}) out[["shift", "actual", "target", "variance", "achievement_pct"]] shift actual target variance achievement_pct 0 shift_1 150 150 0 100.0 1 shift_2 160 200 -40 80.0

best_and_worst_shifts ¤

best_and_worst_shifts(
    counter_uuid: str,
    *,
    value_column_counter: str = ...,
    days: int = ...,
    as_frame: Literal[True] = ...
) -> pd.DataFrame

best_and_worst_shifts(
    counter_uuid: str,
    *,
    value_column_counter: str = ...,
    days: int = ...,
    as_frame: Literal[False]
) -> dict[str, pd.DataFrame]

best_and_worst_shifts(
    counter_uuid: str,
    *,
    value_column_counter: str = "value_integer",
    days: int = 30,
    as_frame: bool = True
) -> dict[str, pd.DataFrame] | pd.DataFrame

Identify best and worst performing shifts.

Parameters:

Name	Type	Description	Default
`counter_uuid`	`str`	Production counter UUID	required
`value_column_counter`	`str`	Column containing counter values	`'value_integer'`
`days`	`int`	Number of recent days to analyze (default: 30)	`30`
`as_frame`	`bool`	When True (default), return a single DataFrame (see below). `as_frame=False` returns the dictionary form; it is deprecated, emits a `DeprecationWarning` and will be removed in the next major release.	`True`

Returns:

Type	Description
`dict[str, DataFrame] \| DataFrame`	With `as_frame=True` (default), one DataFrame with columns:
`dict[str, DataFrame] \| DataFrame`	start, end, duration_seconds: the shift window (`date` at the shift start time, tz-aware in the data's tz; an overnight shift ends on the next day)
`dict[str, DataFrame] \| DataFrame`	rank_type: `'best'` or `'worst'`
`dict[str, DataFrame] \| DataFrame`	date, shift, quantity
`dict[str, DataFrame] \| DataFrame`	The best rows come first (highest quantity first), then the worst
`dict[str, DataFrame] \| DataFrame`	rows (lowest quantity first). A shift can appear in both groups
`dict[str, DataFrame] \| DataFrame`	when there are fewer than ten shifts.
`dict[str, DataFrame] \| DataFrame`	With `as_frame=False` (deprecated), a dictionary with:
`dict[str, DataFrame] \| DataFrame`	'best': Top 5 best shifts (columns date, shift, quantity)
`dict[str, DataFrame] \| DataFrame`	'worst': Top 5 worst shifts (columns date, shift, quantity)

Example

import ts_shape df = ts_shape.make_timeseries(["counter"], n_points=96, freq="30min", ... start="2025-01-01 06:00", baseline=10, noise=3, value_column="value_integer") df["value_integer"] = df["value_integer"].cumsum() # running part counter ranked = ts_shape.ShiftReportEvents(df).best_and_worst_shifts("counter") best = ranked[ranked["rank_type"] == "best"] best[["date", "shift", "quantity"]].head(3) date shift quantity 0 2025-01-01 shift_2 170 1 2025-01-01 shift_3 166 2 2025-01-02 shift_2 157