pytraf — UDP heartbeat + simulated jamming + AAP remediation

Independent scripts, each runnable on its own, plus pytraf.py to orchestrate them together:

• heartbeat_sender.py — sends a UDP heartbeat packet (seq + timestamp) every --interval seconds.
• heartbeat_receiver.py — listens for heartbeats, logs every receipt, and runs a watchdog that logs heartbeat_missed / heartbeat_recovered when packets stop arriving.
• jam_control.py — flips a local jam_state.json flag that the sender checks. When active, the sender drops (and optionally corrupts) a configurable fraction of its own outgoing packets — this simulates jamming without touching real network/RF layers or anyone else's traffic.
• detector.py — tails the receiver's log, computes a rolling miss-ratio, and when it crosses a threshold, POSTs to an Ansible Automation Platform job template to launch remediation.
• build_dataset.py / train_model.py / ml_detector.py — the ML pipeline that replaces detector.py's fixed threshold with a trained classifier (see the training section below).
• pytraf.py — master CLI that starts/stops the above together instead of juggling multiple terminals. See below.

All logs are newline-delimited JSON (logs/*.jsonl) so a real anomaly-detection or LLM-based model can be pointed at them instead of the simple threshold heuristic in detector.py.

Project architecture diagram

flowchart LR
    subgraph Orchestration
        P["pytraf.py\nmaster controller"]
    end

    subgraph Runtime
        J["jam_control.py\nwrites jam_state.json"]
        S["heartbeat_sender.py\nUDP sender"]
        R["heartbeat_receiver.py\nUDP receiver"]
        D["detector.py\nheuristic detector"]
        M["ml_detector.py\ntrained-model detector"]
    end

    subgraph ML_Pipeline
        F["features.py\nshared feature logic"]
        B["build_dataset.py\nbuilds dataset.csv"]
        T["train_model.py\ntrains model.joblib"]
    end

    subgraph State_and_Logs
        JS["jam_state.json\nactive/drop/corrupt flags"]
        SR["logs/sender.jsonl"]
        RR["logs/receiver.jsonl"]
        DR["logs/detector.jsonl"]
        DS["dataset.csv"]
        MM["model.joblib"]
    end

    subgraph External
        AAP["AAP / Ansible Automation Platform\nHTTP job launch"]
    end

    P -->|starts/stops| J
    P -->|starts/stops| S
    P -->|starts/stops| R
    P -->|starts/stops| D
    P -->|starts/stops| M
    P -->|collect + train| B
    P -->|collect + train| T

    J -->|writes active state| JS
    JS -->|read by sender| S

    S -->|UDP heartbeat packets| R
    R -->|receiver events| RR
    S -->|sent/jammed events| SR

    RR -->|tail/follow| D
    RR -->|tail/follow| M
    D -->|detects jamming| DR
    M -->|detects jamming| DR

    D -->|POST trigger_reason| AAP
    M -->|POST trigger_reason| AAP

    F -->|shared feature defs| B
    F -->|shared feature defs| M

    SR -->|socket/label source| B
    RR -->|event stream| B
    B -->|labeled rows| DS
    DS -->|train input| T
    T -->|trained model bundle| MM
    MM -->|loaded by live detector| M

Loading

Terminal UI (pytraf.py tui)

There is now a simple standard-terminal dashboard you can run without installing a separate TUI framework:

python pytraf.py tui

The menu supports:

• launching the live demo
• collecting a training run
• training a model
• toggling jamming on/off
• viewing status
• quitting cleanly

This is a lightweight text dashboard designed for a regular terminal window, including Windows terminals.

Master script (pytraf.py)

pip install -r requirements.txt

# Start receiver + sender + heuristic detector together, Ctrl+C stops all three
python pytraf.py run --dry-run

# Same, but also auto-jam in the background every 1-2 minutes
python pytraf.py run --dry-run --jam-auto

# Same, but using the trained ML model instead of the threshold heuristic
python pytraf.py run --detector ml --dry-run --jam-auto

# Collect 5 minutes of labeled data (receiver+sender+auto-jam), then stop
python pytraf.py collect --minutes 5

# Build the dataset and train the model from the logs just collected
python pytraf.py train

# Both of the above in one shot
python pytraf.py pipeline --minutes 5

# Passthrough to jam_control.py
python pytraf.py jam on --duration 30 --drop-rate 0.9

# Quick look at jam state, log sizes, and whether a model exists
python pytraf.py status

run and collect terminate every child process on Ctrl+C and force jam_state.json back to inactive on the way out, so nothing is left jamming after the script exits. Drop --dry-run once AAP_HOST/AAP_TOKEN/AAP_JOB_TEMPLATE_ID are set (see below) to have detected jamming actually launch the AAP job.

Each script below still works standalone — pytraf.py just saves you from opening several terminals and forgetting to turn jamming back off.

Quick start, one component at a time (three terminals)

pip install -r requirements.txt

# Terminal 1
python heartbeat_receiver.py --port 9999 --interval 1

# Terminal 2
python heartbeat_sender.py --host 127.0.0.1 --port 9999 --interval 1

# Terminal 3
python detector.py --window 15 --threshold 0.5 --dry-run

With --dry-run, the detector prints what it would send to AAP instead of calling it — use this until you're ready to wire up real credentials.

Triggering jamming

# one-shot burst: 30s of ~90% packet drop
python jam_control.py on --duration 30 --drop-rate 0.9

# force it off early
python jam_control.py off

# hands-off: randomly jam every 1-2 minutes for 10-30s at a time
python jam_control.py auto --every-min 60 --every-max 120 --burst-min 10 --burst-max 30

Wiring up real AAP remediation

Set these environment variables (never hardcode credentials in the scripts):

export AAP_HOST="https://aap.example.com"
export AAP_TOKEN="<personal access token>"
export AAP_JOB_TEMPLATE_ID="42"
export AAP_VERIFY_SSL="true"

Then drop --dry-run from the detector.py invocation. It calls:

POST {AAP_HOST}/api/v2/job_templates/{AAP_JOB_TEMPLATE_ID}/launch/
Authorization: Bearer {AAP_TOKEN}
{"extra_vars": {"trigger_reason": {...}}}

detector.py --cooldown (default 120s) prevents re-triggering the job repeatedly while an outage is ongoing.

Tuning detection

• --window: how many seconds of receiver events to look at (default 15s)
• --threshold: fraction of missed/corrupted heartbeats in that window before declaring jamming (default 0.5)
• --miss-threshold on the receiver: how many missed intervals before a gap counts as a "miss" event (default 2.5x the heartbeat interval)

Training a small model to replace the threshold heuristic

detector.py uses a fixed miss-ratio threshold. ml_detector.py swaps that for a small classifier (logistic regression / shallow random forest) trained on your own logs — the sender already stamps jam_active on every packet it sends, so labels come for free; no manual labeling needed.

# 1. Collect a run with both jammed and clean periods (a few minutes)
python heartbeat_receiver.py --port 9999 --interval 1
python heartbeat_sender.py --host 127.0.0.1 --port 9999 --interval 1
python jam_control.py auto --every-min 60 --every-max 120 --burst-min 10 --burst-max 30

# 2. Turn the logs into a labeled dataset
python build_dataset.py --window-size 20 --out dataset.csv

# 3. Train and pick the better of logistic regression / random forest
python train_model.py --dataset dataset.csv --model-out model.joblib

# 4. Run the ML-backed detector instead of detector.py
python ml_detector.py --model model.joblib --window-size 20 --prob-threshold 0.5 --dry-run

features.py is shared by build_dataset.py (offline) and ml_detector.py (live) so the model always sees the same feature definitions it was trained on. --window-size must match between the two. Features are all receiver-observable (loss rate, corrupt rate, current miss streak, average latency, latency jitter over the last N events) — nothing from the sender's internal jam state leaks into inference, only into the training labels.

If accuracy is poor, the usual fixes are: collect a longer/more varied run (more jamming styles — try varying --drop-rate/--corrupt-rate), or tune --window-size (bigger window = smoother but slower to react).

Notes

• Nothing here does real RF or network-level jamming — it's an application-layer simulation between your own sender and receiver, safe to run against localhost or your own lab hosts.
• Not tested by me in this session (no Python interpreter available in this environment) — please run the quick start above and confirm behavior before relying on it.