model/normal_split.py

"""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()