iot_patterns

sqllocks_spindle.simulation.iot_patterns

IoT telemetry stream patterns — sensor drift, missing readings, alert storms, fleet sim.

Generates realistic IoT streaming data anomalies that can be layered on top of base IoT domain data produced by the generator.

Usage::

from sqllocks_spindle.simulation.iot_patterns import IoTTelemetrySimulator, IoTTelemetryConfig

cfg = IoTTelemetryConfig(fleet_size=50, duration_hours=24)
result = IoTTelemetrySimulator(readings_df=readings, devices_df=devices, config=cfg).run()

Classes

IoTTelemetryConfig dataclass

Configuration for :class:IoTTelemetrySimulator.

Parameters:

Name	Type	Description	Default
fleet_size	int	Number of active devices in the simulation.	50
duration_hours	float	Total simulation window in hours.	24.0
reading_interval_seconds	float	Nominal seconds between consecutive readings.	60.0
drift_enabled	bool	Whether to apply gradual sensor drift.	True
drift_rate	float	Per-reading drift magnitude (additive random-walk step size).	0.001
drift_probability	float	Fraction of sensors that will experience drift.	0.1
missing_enabled	bool	Whether to inject missing (NaN) readings.	True
missing_probability	float	Per-reading chance of a value being set to NaN.	0.05
alert_storm_enabled	bool	Whether to generate alert storm bursts.	True
alert_storm_probability	float	Per-hour chance of an alert storm starting.	0.02
alert_storm_duration_minutes	float	How long each alert storm lasts.	15.0
alert_storm_rate_multiplier	float	Alert frequency multiplier during a storm.	10.0
battery_drain_enabled	bool	Whether to simulate battery drain over time.	True
battery_drain_rate	float	Battery percentage lost per hour.	0.1
seed	int	Random seed for reproducibility.	42

IoTTelemetryResult dataclass

Result of :meth:IoTTelemetrySimulator.run.

Attributes:

Name	Type	Description
readings	DataFrame	Modified readings DataFrame with drift and missing values applied.
alerts	DataFrame	Generated alert events including alert-storm bursts.
fleet_status	DataFrame	Per-device status snapshot at the end of the simulation.
stats	dict[str, Any]	Summary statistics dictionary.

IoTTelemetrySimulator

Simulate realistic IoT telemetry anomalies on top of generated data.

Takes pre-generated readings and devices DataFrames (from the IoT domain generator) and applies configurable anomaly layers: sensor drift, missing readings, alert storms, and battery drain.

Parameters:

Name	Type	Description	Default
readings_df	DataFrame | None	Base readings DataFrame. Expected to contain at least device_id (or sensor_id), a numeric value column, and a reading_time (or created_at) timestamp column.	None
devices_df	DataFrame | None	Devices DataFrame. Expected to contain at least device_id and optionally battery_level.	None
config	IoTTelemetryConfig | None	:class:IoTTelemetryConfig.	None

Example::

sim = IoTTelemetrySimulator(readings_df=readings, devices_df=devices)
result = sim.run()
result.readings.head()

Methods:

run()

Execute all enabled simulation layers and return results.