| """Generate ~1000 windows of healthy-shape synthetic telemetry. |
| |
| Used by Pattern 8 (model/train_anomaly.py) to calibrate the |
| 95th-percentile-of-normal threshold (D-ANOM-02). |
| |
| Mirrors Phase 1's columnar Parquet writer convention exactly so |
| load_anomaly_features() works on the output without modification. |
| """ |
| from __future__ import annotations |
|
|
| from pathlib import Path |
| from typing import Any |
|
|
| import pyarrow as pa |
| import pyarrow.parquet as pq |
| from numpy.random import PCG64, Generator, SeedSequence |
|
|
| from model.seeds import phase2_seeds |
| from model.synth.state_machines.normal_baseline import ( |
| _normal_baseline_window, |
| ) |
|
|
| N_NORMAL: int = 1000 |
| |
|
|
|
|
| |
| |
| |
| _COLUMNS: tuple[str, ...] = ( |
| "timestamp", "os", "network_mode", "rssi_dbm", "bssid", "bssid_mode", |
| "channel", |
| "ping_continuity_window_ms", "ping_continuity_avg_rtt_ms", |
| "ping_continuity_packet_loss_pct", "ping_continuity_jitter_ms", |
| "latency_jitter_ms", "dns_resolution_ms", |
| "dhcp_event_class", "auth_event_class", "captive_portal_detected", |
| "mac_randomization_state", "driver_state", |
| "per_packet_retry_count", "rts_cts_rate", "beacon_rssi_dbm", |
| "neighbor_ap_count_5ghz", |
| "window_ms", |
| "class", |
| ) |
|
|
|
|
| def _flatten_frame(frame: dict[str, Any]) -> dict[str, Any]: |
| """Flatten ping_continuity sub-dict into 4 columnar fields.""" |
| out = dict(frame) |
| pc = out.pop("ping_continuity") |
| out["ping_continuity_window_ms"] = pc["window_ms"] |
| out["ping_continuity_avg_rtt_ms"] = pc["avg_rtt_ms"] |
| out["ping_continuity_packet_loss_pct"] = pc["packet_loss_pct"] |
| out["ping_continuity_jitter_ms"] = pc["jitter_ms"] |
| return out |
|
|
|
|
| def generate_normal_split(seed: int, out_path: Path) -> None: |
| """Generate N_NORMAL windows of healthy-shape frames into out_path Parquet.""" |
| rng = Generator(PCG64(SeedSequence(seed))) |
|
|
| columns: dict[str, list[Any]] = {col: [] for col in _COLUMNS} |
| for _ in range(N_NORMAL): |
| window = _normal_baseline_window(rng) |
| for frame in window: |
| flat = _flatten_frame(frame) |
| for col in _COLUMNS: |
| columns[col].append(flat[col]) |
|
|
| out_path.parent.mkdir(parents=True, exist_ok=True) |
| pq.write_table(pa.Table.from_pydict(columns), out_path) |
|
|
|
|
| def main() -> None: |
| """`python -m model.normal_split` entry -- used by `make synth-normal`. |
| |
| Uses model.seeds.phase2_seeds()["normal_split_synth"] (D-REPRO-02 sub-stream). |
| """ |
| seed = phase2_seeds()["normal_split_synth"] |
| out = Path("data/normal.parquet") |
| generate_normal_split(seed, out) |
| print(f"wrote {out}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
