File size: 2,962 Bytes
c57b128 | 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 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """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 # 1000 windows x ~30 frames = ~30k rows.
# Stable 95th-percentile estimate (~1500 samples in tail).
# Mirror Phase 1's _COLUMNS (24-column flattened schema, EXACT order from
# model/synth/generate.py::_COLUMNS). Hard-coded here to avoid importing private
# symbols. NOTE: bssid_mode comes BEFORE channel in the canonical order.
_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()
|