lesavka/scripts/manual/local_av_stimulus.py

#!/usr/bin/env python3
"""Serve a local A/V stimulus page for the mirrored upstream sync probe."""

from __future__ import annotations

import argparse
import http.server
import json
import socketserver
import threading
import time
import urllib.parse
from pathlib import Path

DEFAULT_EVENT_WIDTH_CODES = "1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Serve Lesavka local A/V sync stimulus")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=18444)
    parser.add_argument("--status", default="/tmp/lesavka-local-av-stimulus-status.json")
    parser.add_argument("--duration-seconds", type=int, default=20)
    parser.add_argument("--warmup-seconds", type=int, default=4)
    parser.add_argument("--pulse-period-ms", type=int, default=1000)
    parser.add_argument("--pulse-width-ms", type=int, default=120)
    parser.add_argument("--marker-tick-period", type=int, default=5)
    parser.add_argument("--audio-gain", type=float, default=0.55)
    parser.add_argument("--event-width-codes", default=DEFAULT_EVENT_WIDTH_CODES)
    args = parser.parse_args()
    args.event_width_codes = parse_event_width_codes(args.event_width_codes)
    args.audio_gain = max(0.0, min(1.0, args.audio_gain))
    return args


def parse_event_width_codes(raw: str) -> list[int]:
    codes = [int(part.strip()) for part in raw.split(",") if part.strip()]
    if not codes:
        raise SystemExit("--event-width-codes must contain at least one integer")
    if any(code < 1 for code in codes):
        raise SystemExit("--event-width-codes values must be positive")
    return codes


class StimulusState:
    def __init__(self, status_path: Path, args: argparse.Namespace) -> None:
        self.status_path = status_path
        self.args = args
        self.lock = threading.RLock()
        self.start_token = 0
        self.preview_token = 0
        self.preview_seconds = 0
        self.status = {
            "booted_at": time.time(),
            "ready": False,
            "started": False,
            "completed": False,
            "last_error": None,
            "page_message": "booting",
            "stimulus_mode": "idle",
            "audio_state": "not-created",
            "observed_start_token": None,
            "completed_start_token": None,
            "observed_preview_token": None,
            "completed_preview_token": None,
            "pulse_active": False,
            "pulse_index": 0,
            "pulse_width_code": 0,
            "last_update": time.time(),
        }
        self.write_status()

    def write_status(self) -> None:
        self.status_path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.status_path.with_suffix(".tmp")
        tmp.write_text(json.dumps(self.status, indent=2, sort_keys=True), encoding="utf-8")
        tmp.replace(self.status_path)

    def update(self, payload: dict) -> None:
        with self.lock:
            self.status.update(payload)
            self.status["last_update"] = time.time()
            self.write_status()

    def snapshot(self) -> dict:
        with self.lock:
            snap = dict(self.status)
            snap.update({
                "start_token": self.start_token,
                "preview_token": self.preview_token,
                "preview_seconds": self.preview_seconds,
                "duration_seconds": self.args.duration_seconds,
                "warmup_seconds": self.args.warmup_seconds,
                "pulse_period_ms": self.args.pulse_period_ms,
                "pulse_width_ms": self.args.pulse_width_ms,
                "marker_tick_period": self.args.marker_tick_period,
                "audio_gain": self.args.audio_gain,
                "event_width_codes": self.args.event_width_codes,
            })
            return snap

    def request_start(self) -> dict:
        with self.lock:
            self.start_token += 1
            self.status.update({
                "started": False,
                "completed": False,
                "last_error": None,
                "page_message": "start requested",
                "stimulus_mode": "start",
                "start_requested_at": time.time(),
            })
            self.write_status()
            return self.snapshot()

    def request_preview(self, seconds: int) -> dict:
        with self.lock:
            self.preview_token += 1
            self.preview_seconds = max(1, min(30, seconds))
            self.status.update({
                "started": False,
                "completed": False,
                "last_error": None,
                "page_message": "preview requested",
                "stimulus_mode": "preview",
                "preview_requested_at": time.time(),
            })
            self.write_status()
            return self.snapshot()


def page_html() -> str:
    return """<!doctype html>
<html>
<head>
<meta charset=\"utf-8\">
<title>Lesavka Local A/V Stimulus</title>
<style>
html, body { margin: 0; width: 100%; height: 100%; overflow: hidden; background: #02040a; color: #eaf2ff; font: 20px/1.4 system-ui, sans-serif; }
#stage { position: fixed; inset: 0; display: grid; place-items: center; background: #000; transition: none; }
#stage.active { background: var(--pulse-color, #ff2d2d); color: #02040a; }
#card { max-width: 900px; padding: 28px; border-radius: 24px; background: rgba(0, 0, 0, 0.82); border: 1px solid rgba(52, 65, 86, 0.34); text-align: center; }
#stage.active #card { background: transparent; border-color: transparent; }
#big { font-size: clamp(48px, 9vw, 140px); font-weight: 900; letter-spacing: 0.06em; color: #111827; }
#stage.active #big { color: rgba(0,0,0,0.16); }
#status { white-space: pre-wrap; font: 15px/1.45 ui-monospace, monospace; color: #1f2937; opacity: 0.72; }
#stage.active #status { color: rgba(0,0,0,0.22); opacity: 0.68; }
</style>
</head>
<body>
<div id=\"stage\"><div id=\"card\"><div id=\"big\">LESAVKA</div><div id=\"status\">booting...</div></div></div>
<script>
const stage = document.getElementById('stage');
const statusEl = document.getElementById('status');
let startToken = 0;
let running = false;
let audioCtx = null;
let oscillator = null;
let gain = null;
let startedAt = 0;
let stimulusMode = 'idle';
let lastPulse = { active: false, pulseIndex: 0, widthCode: 0 };
let previewToken = 0;
const pulseColors = {
  1: '#b81d24',
  2: '#007a3d',
  3: '#1456b8',
  4: '#b56b00',
  5: '#d81b60',
  6: '#00bcd4',
  7: '#cddc39',
  8: '#7e57c2',
  9: '#ff7043',
  10: '#26a69a',
  11: '#ff4081',
  12: '#5c6bc0',
  13: '#ffeb3b',
  14: '#69f0ae',
  15: '#ab47bc',
  16: '#03a9f4'
};
const pulseFrequencies = {
  1: 620,
  2: 780,
  3: 940,
  4: 1120,
  5: 1320,
  6: 1540,
  7: 1780,
  8: 2040,
  9: 2320,
  10: 2620,
  11: 2960,
  12: 3340,
  13: 3760,
  14: 4220,
  15: 4740,
  16: 5320
};

function setStatus(message) { statusEl.textContent = message; }
async function postJson(path, payload) {
  await fetch(path, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(payload) });
}
function ensureAudio() {
  if (audioCtx) return;
  audioCtx = new AudioContext();
  oscillator = audioCtx.createOscillator();
  oscillator.type = 'sine';
  oscillator.frequency.value = 880;
  gain = audioCtx.createGain();
  gain.gain.value = 0;
  oscillator.connect(gain).connect(audioCtx.destination);
  oscillator.start();
}
function eventAt(elapsedMs, command) {
  const warmupMs = command.warmup_seconds * 1000;
  if (elapsedMs < warmupMs || elapsedMs > command.duration_seconds * 1000) return { active: false, pulseIndex: 0, widthCode: 1 };
  const sinceWarmup = elapsedMs - warmupMs;
  const pulseIndex = Math.floor(sinceWarmup / command.pulse_period_ms);
  const offset = sinceWarmup % command.pulse_period_ms;
  const codes = command.event_width_codes && command.event_width_codes.length ? command.event_width_codes : [1];
  const widthCode = codes[pulseIndex % codes.length];
  const width = Math.min(command.pulse_period_ms - 1, command.pulse_width_ms);
  return { active: offset < width, pulseIndex, widthCode };
}
async function runStimulus(command, mode, token) {
  if (running) {
    await postJson('/status', { last_error: `${mode} token ${token} ignored because stimulus is already running`, page_message: 'stimulus already running' });
    return false;
  }
  running = true;
  stimulusMode = mode;
  try {
    ensureAudio();
    await audioCtx.resume();
    if (audioCtx.state !== 'running') {
      throw new Error(`AudioContext did not start; state=${audioCtx.state}`);
    }
  } catch (err) {
    running = false;
    stimulusMode = 'idle';
    stage.classList.remove('active');
    if (gain) gain.gain.setTargetAtTime(0, audioCtx ? audioCtx.currentTime : 0, 0.005);
    await postJson('/status', {
      ready: true,
      started: false,
      completed: false,
      last_error: String(err && (err.stack || err)),
      stimulus_mode: mode,
      audio_state: audioCtx ? audioCtx.state : 'not-created',
      page_message: `stimulus ${mode} failed`,
    });
    return false;
  }
  startedAt = performance.now();
  const tokenPayload = mode === 'preview' ? { observed_preview_token: token } : { observed_start_token: token };
  await postJson('/status', {
    ready: true,
    started: true,
    completed: false,
    last_error: null,
    stimulus_mode: mode,
    audio_state: audioCtx.state,
    page_message: mode === 'preview' ? 'stimulus preview running' : 'stimulus running',
    ...tokenPayload,
  });
  const tick = async () => {
    const elapsed = performance.now() - startedAt;
    const event = eventAt(elapsed, command);
    lastPulse = event;
    const active = event.active;
    const pulseIndex = event.pulseIndex;
    const widthCode = event.widthCode;
    stage.style.setProperty('--pulse-color', pulseColors[widthCode] || pulseColors[1]);
    stage.classList.toggle('active', active);
    oscillator.frequency.setTargetAtTime(pulseFrequencies[widthCode] || pulseFrequencies[1], audioCtx.currentTime, 0.003);
    const audioGain = Math.max(0, Math.min(1, Number(command.audio_gain ?? 0.55)));
    gain.gain.setTargetAtTime(active ? audioGain : 0.0, audioCtx.currentTime, 0.005);
    setStatus(`running\nelapsed=${(elapsed / 1000).toFixed(2)}s\nactive=${active}\nevent=${pulseIndex}\nwidth_code=${widthCode}\nPoint the real webcam at this window and keep the real microphone hearing the tone.`);
    if (elapsed <= command.duration_seconds * 1000 + 500) {
      requestAnimationFrame(tick);
    } else {
      stage.classList.remove('active');
      gain.gain.setTargetAtTime(0, audioCtx.currentTime, 0.005);
      running = false;
      stimulusMode = 'idle';
      lastPulse = { active: false, pulseIndex, widthCode };
      const completedPayload = mode === 'preview' ? { completed_preview_token: token } : { completed_start_token: token };
      await postJson('/status', {
        completed: true,
        stimulus_mode: mode,
        audio_state: audioCtx.state,
        page_message: mode === 'preview' ? 'stimulus preview completed' : 'stimulus completed',
        ...completedPayload,
      });
      setStatus('completed');
    }
  };
  requestAnimationFrame(tick);
  return true;
}
async function pollCommand() {
  try {
    const command = await fetch('/command').then(r => r.json());
    if (command.preview_token !== previewToken) {
      const requestedPreviewToken = command.preview_token;
      previewToken = requestedPreviewToken;
      const previewSeconds = Number(command.preview_seconds || 4);
      await runStimulus({ ...command, duration_seconds: previewSeconds, warmup_seconds: 0 }, 'preview', requestedPreviewToken);
      return;
    }
    if (command.start_token !== startToken) {
      const requestedStartToken = command.start_token;
      const didStart = await runStimulus(command, 'start', requestedStartToken);
      if (didStart) startToken = requestedStartToken;
    }
  } catch (err) {
    await postJson('/status', { last_error: String(err), page_message: 'command poll failed' });
  }
}
setInterval(pollCommand, 200);
setInterval(() => postJson('/status', {
  ready: true,
  page_message: running ? `${stimulusMode} heartbeat` : 'ready heartbeat',
  stimulus_mode: stimulusMode,
  audio_state: audioCtx ? audioCtx.state : 'not-created',
  pulse_active: Boolean(lastPulse.active),
  pulse_index: Number(lastPulse.pulseIndex || 0),
  pulse_width_code: Number(lastPulse.widthCode || 0),
}), 1000);
void postJson('/status', { ready: true, page_message: 'page ready' });
setStatus(`ready
Point the real webcam at this window.
Keep speakers audible to the selected microphone.`);
</script>
</body>
</html>"""


class StimulusHandler(http.server.BaseHTTPRequestHandler):
    state: StimulusState

    def _send(self, code: int, body: bytes, content_type: str = "application/json") -> None:
        self.send_response(code)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path in ("/", "/index.html"):
            self.state.update({"page_message": "html served"})
            self._send(200, page_html().encode("utf-8"), "text/html; charset=utf-8")
            return
        if parsed.path == "/command":
            self._send(200, json.dumps(self.state.snapshot()).encode("utf-8"))
            return
        if parsed.path == "/status":
            self._send(200, json.dumps(self.state.snapshot()).encode("utf-8"))
            return
        self._send(404, b"not found", "text/plain; charset=utf-8")

    def do_POST(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        length = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(length)
        if parsed.path == "/status":
            payload = json.loads(body.decode("utf-8"))
            self.state.update(payload)
            self._send(200, json.dumps(self.state.snapshot()).encode("utf-8"))
            return
        if parsed.path == "/start":
            self._send(200, json.dumps(self.state.request_start()).encode("utf-8"))
            return
        if parsed.path == "/preview":
            query = urllib.parse.parse_qs(parsed.query)
            seconds = 4
            if query.get("seconds"):
                try:
                    seconds = int(query["seconds"][0])
                except ValueError:
                    seconds = 4
            self._send(200, json.dumps(self.state.request_preview(seconds)).encode("utf-8"))
            return
        self._send(404, b"not found", "text/plain; charset=utf-8")

    def log_message(self, fmt: str, *args) -> None:
        pass


class ReusableTcpServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    allow_reuse_address = True
    daemon_threads = True


def main() -> None:
    args = parse_args()
    state = StimulusState(Path(args.status), args)

    class Handler(StimulusHandler):
        pass

    Handler.state = state
    with ReusableTcpServer((args.host, args.port), Handler) as httpd:
        httpd.serve_forever()


if __name__ == "__main__":
    main()