File size: 2,019 Bytes
fa840fe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 | #!/usr/bin/env python
"""Generate a synthetic sensor CSV for manual testing / demos.
Usage:
python scripts/generate_sample_csv.py --rows 100000 --out sample_sensors.csv
python scripts/generate_sample_csv.py --rows 5000000 --out big.csv # stress test
Produces realistic temperature/vibration/pressure/rotational_speed readings with
a configurable fraction of injected anomalies (spikes / level shifts).
"""
from __future__ import annotations
import argparse
import numpy as np
import pandas as pd
def generate(rows: int, anomaly_frac: float, seed: int = 42) -> pd.DataFrame:
rng = np.random.default_rng(seed)
temperature = rng.normal(70, 1.5, rows)
vibration = rng.normal(0.5, 0.05, rows)
pressure = rng.normal(30, 0.8, rows)
rotational_speed = rng.normal(1500, 20, rows)
n_anom = int(rows * anomaly_frac)
idx = rng.choice(rows, size=n_anom, replace=False)
temperature[idx] += rng.normal(20, 5, n_anom)
vibration[idx] += rng.normal(1.0, 0.3, n_anom)
pressure[idx] -= rng.normal(8, 2, n_anom)
timestamps = pd.date_range("2026-01-01", periods=rows, freq="s")
return pd.DataFrame(
{
"timestamp": timestamps,
"temperature": temperature.round(3),
"vibration": vibration.round(4),
"pressure": pressure.round(3),
"rotational_speed": rotational_speed.round(1),
}
)
def main() -> None:
parser = argparse.ArgumentParser(description="Generate a sample sensor CSV.")
parser.add_argument("--rows", type=int, default=100_000)
parser.add_argument("--anomaly-frac", type=float, default=0.02)
parser.add_argument("--out", type=str, default="sample_sensors.csv")
parser.add_argument("--seed", type=int, default=42)
args = parser.parse_args()
df = generate(args.rows, args.anomaly_frac, args.seed)
df.to_csv(args.out, index=False)
print(f"Wrote {len(df):,} rows ({args.anomaly_frac:.1%} anomalies) -> {args.out}")
if __name__ == "__main__":
main()
|