ADD: Simulate measurements

simulate_measurements.py

@@ -0,0 +1,144 @@
+"""
+Simulates weather measurements for development purposes.
+Uses SQLModel directly so all model defaults are respected.
+Generates realistic indoor/outdoor measurements for Germany over a full year.
+
+IMPORTANT: Stop your FastAPI server before running this script.
+"""
+
+import math
+import random
+import sys
+import os
+from datetime import datetime, timezone, timedelta
+
+# make sure models are importable - adjust path if needed
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+from sqlmodel import Session, create_engine, select
+from models.measurement import IndoorMeasurement, OutdoorMeasurement
+from models.station import Station
+
+DB_PATH = "database.db"
+
+STATIONS = [
+    {"mac": "AA:BB:CC:DD:EE:01", "name": "Wohnzimmer",    "indoor_only": True},
+    {"mac": "AA:BB:CC:DD:EE:02", "name": "Schlafzimmer",  "indoor_only": True},
+    {"mac": "AA:BB:CC:DD:EE:03", "name": "Wetterstation", "indoor_only": False},
+]
+
+
+# ── Sensor simulation ─────────────────────────────────────────────────────────
+
+def outdoor_temperature(dt: datetime) -> float:
+    day_of_year = dt.timetuple().tm_yday
+    hour = dt.hour + dt.minute / 60
+    seasonal = 11 + 11 * math.sin((day_of_year - 80) / 365 * 2 * math.pi)
+    daily = 4 * math.sin((hour - 5) / 24 * 2 * math.pi)
+    return round(seasonal + daily + random.gauss(0, 0.5), 1)
+
+def indoor_temperature(outdoor_temp: float) -> float:
+    base = 20 + (outdoor_temp - 11) * 0.15
+    return round(base + random.gauss(0, 0.2), 1)
+
+def outdoor_humidity(temp: float) -> float:
+    base = 80 - (temp - 5) * 0.8
+    return round(max(30.0, min(99.0, base + random.gauss(0, 3))), 1)
+
+def indoor_humidity() -> float:
+    return round(random.gauss(52, 3), 1)
+
+def outdoor_pressure(dt: datetime) -> float:
+    day_of_year = dt.timetuple().tm_yday
+    slow_wave = 5 * math.sin(day_of_year / 365 * 2 * math.pi)
+    return round(1013 + slow_wave + random.gauss(0, 1.5), 1)
+
+
+# ── DB helpers ────────────────────────────────────────────────────────────────
+
+def get_or_create_station(session: Session, mac: str, name: str) -> Station:
+    station = session.exec(select(Station).where(Station.mac == mac)).first()
+    if not station:
+        station = Station(mac=mac, name=name)  # created_at via default_factory
+        session.add(station)
+        session.flush()  # assigns id without committing
+    return station
+
+
+def generate_indoor_rows(station_id: int, start: datetime, end: datetime) -> list[IndoorMeasurement]:
+    rows = []
+    current = start
+    while current <= end:
+        out_temp = outdoor_temperature(current)
+        rows.append(IndoorMeasurement(
+            station_id=station_id,
+            timestamp=current,  # explicit historical timestamp
+            temperature=indoor_temperature(out_temp),
+            humidity=indoor_humidity(),
+        ))
+        current += timedelta(minutes=5)
+    return rows
+
+
+def generate_outdoor_rows(station_id: int, start: datetime, end: datetime) -> list[OutdoorMeasurement]:
+    rows = []
+    current = start
+    while current <= end:
+        out_temp = outdoor_temperature(current)
+        rows.append(OutdoorMeasurement(
+            station_id=station_id,
+            timestamp=current,  # explicit historical timestamp
+            temperature=out_temp,
+            humidity=outdoor_humidity(out_temp),
+            pressure=outdoor_pressure(current),
+        ))
+        current += timedelta(minutes=2)
+    return rows
+
+
+# ── Main ──────────────────────────────────────────────────────────────────────
+
+def simulate():
+    start = datetime(2024, 1, 1, 0, 0, tzinfo=timezone.utc)
+    end   = datetime(2024, 12, 31, 23, 59, tzinfo=timezone.utc)
+
+    engine = create_engine(f"sqlite:///{DB_PATH}")
+
+    with Session(engine) as session:
+        try:
+            for station_cfg in STATIONS:
+                mac, name = station_cfg["mac"], station_cfg["name"]
+                station = get_or_create_station(session, mac, name)
+                print(f"\nStation: {name} (mac={mac})")
+
+                print("  Generating indoor rows...", end=" ", flush=True)
+                indoor_rows = generate_indoor_rows(station.id, start, end)
+                print(f"{len(indoor_rows)} rows")
+
+                print("  Inserting indoor rows...", end=" ", flush=True)
+                session.add_all(indoor_rows)
+                print("done")
+
+                if not station_cfg["indoor_only"]:
+                    print("  Generating outdoor rows...", end=" ", flush=True)
+                    outdoor_rows = generate_outdoor_rows(station.id, start, end)
+                    print(f"{len(outdoor_rows)} rows")
+
+                    print("  Inserting outdoor rows...", end=" ", flush=True)
+                    session.add_all(outdoor_rows)
+                    print("done")
+
+            print("\nCommitting...", end=" ", flush=True)
+            session.commit()
+            print("done")
+
+        except Exception as e:
+            session.rollback()
+            print(f"\n[ERROR] Rolled back: {e}")
+            raise
+
+    print("\nSimulation complete.")
+
+
+if __name__ == "__main__":
+    simulate()