diff --git a/Mimir.slnx b/Mimir.slnx index 84ccf97..901ce47 100644 --- a/Mimir.slnx +++ b/Mimir.slnx @@ -3,5 +3,7 @@ + + diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index 2b3abf3..ea070b1 100644 --- a/docs/implementation-plan.md +++ b/docs/implementation-plan.md @@ -39,6 +39,13 @@ a named invariant that the native runtime cannot protect yet. have USB-attached Moves; Starfire can feed local tracking streams, while Muninn on Nightwing publishes remote tracking streams. Odin owns discovery and schema projection, not pose capture or Mimir fusion. +- `scripts/start-nightwing-move-tracking.ps1` is the narrow live Nightwing + bring-up path: it starts the `/eve/periwinkle` receiver/recorder on Starfire, + stages the Nightwing Eye/Move Python worker, lights configured USB Moves via + HID report `0x06`, and publishes `mimir.psmove_light_state.v1` plus + `mimir.move_controller_observation_state.v1` blob observations from both PS3 + Eyes. The blob stream is optical witness evidence for later pose fusion, not + the final 6DoF pose owner. - `MimirProcessStreamSource` for bridge/network command edges. - `MimirFrameEventProcessStreamSource` for temporary JSON-line frame metadata from native probes into the same rolling buffers Fensalir inspects. One probe diff --git a/notes/current-system-map.md b/notes/current-system-map.md index 936b4e4..139c133 100644 --- a/notes/current-system-map.md +++ b/notes/current-system-map.md @@ -120,6 +120,17 @@ Smoke proof: `dotnet run --project .\src\Mimir.BufferSmoke\Mimir.BufferSmoke.csp -- --move-tracking-contract-smoke` consumes one Starfire Move and one Nightwing Move observation into two tracking buffers. +Live Nightwing bring-up uses `scripts/start-nightwing-move-tracking.ps1`. +Starfire starts `Mimir.EveSensorReceiver` on `/eve/periwinkle`, stages +`nw_eye_cap.py`, `nw_move_hint.py`, and `nightwing_typed_witness_publisher.py` +onto Nightwing, and launches the publisher with `--track-eyes` plus explicit +`--move-light name=/dev/hidrawN:#rrggbb` HID ownership. The worker keeps the +USB Move sphere lit with PS Move report `0x06`, emits +`mimir.psmove_light_state.v1`, and publishes compact +`mimir.move_controller_observation_state.v1` blob observations from both PS3 +Eyes. Those Eye/Move observations are the live optical witness feeding the later +pose/fusion owner; they are not themselves the final 6DoF pose authority. + ## Program Composition [[docs/mimir-program-composition|Mimir Program Composition]] is the live diff --git a/scripts/start-nightwing-move-tracking.ps1 b/scripts/start-nightwing-move-tracking.ps1 new file mode 100644 index 0000000..f3397d5 --- /dev/null +++ b/scripts/start-nightwing-move-tracking.ps1 @@ -0,0 +1,158 @@ +param( + [string]$NightwingHost = "nightwing", + [int]$Port = 8796, + [string]$StarfireHost = "192.168.1.66", + [string]$Path = "/eve/periwinkle", + [string]$EyeDevices = "/dev/video2,/dev/video3", + [string[]]$MoveLight = @( + "move-usb=/dev/hidraw1:#35ff6c" + ), + [string]$RecorderRoot = "C:\Users\Meta\Videos\Mimir\VerseCaptures", + [switch]$SkipRecorder, + [switch]$DryRun +) + +$ErrorActionPreference = "Stop" +$repo = Split-Path -Parent $PSScriptRoot +$runId = Get-Date -Format "yyyyMMdd-HHmmss" +$runDir = Join-Path $repo "artifacts\runtime\nightwing-move-tracking-$runId" +$witnessUrl = "ws://$StarfireHost`:$Port$Path" +$subscribeUrl = "ws://127.0.0.1:$Port$Path/subscribe" + +New-Item -ItemType Directory -Force -Path $runDir | Out-Null +New-Item -ItemType Directory -Force -Path $RecorderRoot | Out-Null + +function Start-LoggedProcess { + param( + [string]$Name, + [string]$FilePath, + [string[]]$ArgumentList + ) + + $stdout = Join-Path $runDir "$Name.out.log" + $stderr = Join-Path $runDir "$Name.err.log" + if ($DryRun) { + Write-Host "DRY $Name`: $FilePath $($ArgumentList -join ' ')" + return $null + } + + $process = Start-Process ` + -FilePath $FilePath ` + -ArgumentList $ArgumentList ` + -WorkingDirectory $repo ` + -WindowStyle Hidden ` + -RedirectStandardOutput $stdout ` + -RedirectStandardError $stderr ` + -PassThru + Set-Content -Path (Join-Path $runDir "$Name.pid") -Value $process.Id + Write-Host "$Name pid=$($process.Id) stdout=$stdout stderr=$stderr" + return $process +} + +function Wait-Health { + param([string]$Url) + if ($DryRun) { + Write-Host "DRY wait health: $Url" + return + } + + $deadline = (Get-Date).AddSeconds(45) + do { + try { + $response = Invoke-WebRequest -UseBasicParsing -Uri $Url -TimeoutSec 3 + if ($response.StatusCode -ge 200 -and $response.StatusCode -lt 300) { + Write-Host "health ok: $Url" + return + } + } catch { + Start-Sleep -Milliseconds 500 + } + } while ((Get-Date) -lt $deadline) + + throw "Timed out waiting for $Url" +} + +Write-Host "Nightwing Move tracking run: $runDir" +Write-Host "Witness URL: $witnessUrl" + +if (-not $DryRun) { + Get-CimInstance Win32_Process | + Where-Object { + ($_.CommandLine -match "Mimir.EveSensorReceiver" -and $_.CommandLine -match "--port $Port") -or + ($_.CommandLine -match "Mimir.VerseRecorder" -and $_.CommandLine -match "$Port$Path/subscribe") + } | + ForEach-Object { + try { Stop-Process -Id $_.ProcessId -Force -ErrorAction Stop } catch {} + } +} + +$receiver = Start-LoggedProcess ` + -Name "periwinkle-receiver" ` + -FilePath "dotnet" ` + -ArgumentList @( + "run", + "--project", + (Join-Path $repo "src\Mimir.EveSensorReceiver\Mimir.EveSensorReceiver.csproj"), + "--", + "--port", "$Port", + "--path", $Path, + "--subscribe-path", "$Path/subscribe", + "--source-id", "nightwing-witness", + "--type", "cultmesh-observation" + ) + +if ($receiver) { + Wait-Health -Url "http://127.0.0.1:$Port/health" +} + +if (-not $SkipRecorder) { + Start-LoggedProcess ` + -Name "verse-recorder" ` + -FilePath "dotnet" ` + -ArgumentList @( + "run", + "--project", + (Join-Path $repo "src\Mimir.VerseRecorder\Mimir.VerseRecorder.csproj"), + "--", + "--url", $subscribeUrl, + "--out-dir", $RecorderRoot + ) | Out-Null +} + +if ($DryRun) { + Write-Host "DRY Nightwing stage and start on $NightwingHost" +} else { + scp (Join-Path $repo "tools\nw_eye_cap.py") "${NightwingHost}:/tmp/nw_eye_cap.py" | Out-Null + scp (Join-Path $repo "tools\nw_move_hint.py") "${NightwingHost}:/tmp/nw_move_hint.py" | Out-Null + scp (Join-Path $repo "tools\nightwing_typed_witness_publisher.py") "${NightwingHost}:/tmp/nightwing_typed_witness_publisher.py" | Out-Null + + $remoteLog = "~/.local/state/gamecult/mimir-nightwing-move-tracking-$runId.log" + $remotePid = "~/.local/state/gamecult/mimir-nightwing-move-tracking-$runId.pid" + $moveArgs = ($MoveLight | ForEach-Object { "--move-light '$($_ -replace "'", "'\''")'" }) -join " " + $remoteScript = @" +mkdir -p ~/.local/state/gamecult +pkill -f '/tmp/nightwing_typed_witness_publisher.py.*--track-eyes' || true +nohup python3 /tmp/nightwing_typed_witness_publisher.py --url '$witnessUrl' --track-eyes --track-builtin-camera --track-builtin-mic --interval 0.10 --eye-devices '$EyeDevices' --eye-window-seconds 0.08 --tracking-stride 6 $moveArgs > $remoteLog 2>&1 < /dev/null & +sleep 0.5 +pgrep -n -f '/tmp/nightwing_typed_witness_publisher.py.*--track-eyes' > $remotePid +cat $remotePid +"@ + $remoteB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($remoteScript)) + $nightwingPid = ssh $NightwingHost "printf '%s' '$remoteB64' | base64 -d | bash" + Set-Content -Path (Join-Path $runDir "nightwing-witness.pid") -Value ($nightwingPid | Select-Object -First 1) + Set-Content -Path (Join-Path $runDir "nightwing-witness.logpath.txt") -Value $remoteLog + Write-Host "nightwing-witness pid=$($nightwingPid | Select-Object -First 1) log=$remoteLog" +} + +$manifest = [ordered]@{ + kind = "mimir.nightwing_move_tracking_supervisor.v1" + runId = $runId + runDir = $runDir + witnessUrl = $witnessUrl + subscribeUrl = $subscribeUrl + eyeDevices = $EyeDevices + moveLights = $MoveLight + startedAt = (Get-Date).ToString("o") +} +$manifest | ConvertTo-Json -Depth 5 | Set-Content -Path (Join-Path $runDir "supervisor.json") +Write-Host "Supervisor manifest: $(Join-Path $runDir "supervisor.json")" diff --git a/src/Mimir.EveSensorReceiver/Mimir.EveSensorReceiver.csproj b/src/Mimir.EveSensorReceiver/Mimir.EveSensorReceiver.csproj new file mode 100644 index 0000000..48be7bc --- /dev/null +++ b/src/Mimir.EveSensorReceiver/Mimir.EveSensorReceiver.csproj @@ -0,0 +1,17 @@ + + + + Exe + net10.0 + Mimir.EveSensorReceiver + Mimir.EveSensorReceiver + enable + enable + + + + + + + + diff --git a/src/Mimir.EveSensorReceiver/Program.cs b/src/Mimir.EveSensorReceiver/Program.cs new file mode 100644 index 0000000..7cdfc78 --- /dev/null +++ b/src/Mimir.EveSensorReceiver/Program.cs @@ -0,0 +1,501 @@ +using System.Net; +using System.Net.Sockets; +using System.Collections.Concurrent; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using MessagePack; +using Mimir.Runtime.Synchronization; + +var options = EveSensorReceiverOptions.Parse(args); +using var receiver = new EveSensorReceiver(options); +await receiver.RunAsync(); + +internal sealed record EveSensorReceiverOptions( + int Port, + string Path, + string SourceId, + string ExpectedType, + string SubscribePath) +{ + public static EveSensorReceiverOptions Parse(string[] args) + { + return new EveSensorReceiverOptions( + ParseInt(args, "--port", 8793), + ParseString(args, "--path", "/eve/camera"), + ParseString(args, "--source-id", "eve-camera"), + ParseString(args, "--type", "video-frame"), + ParseString(args, "--subscribe-path", ParseString(args, "--path", "/eve/camera") + "/subscribe")); + } + + private static string ParseString(IReadOnlyList args, string name, string fallback) + { + for (var index = 0; index < args.Count - 1; index++) + { + if (string.Equals(args[index], name, StringComparison.OrdinalIgnoreCase)) + { + return args[index + 1]; + } + } + + return fallback; + } + + private static int ParseInt(IReadOnlyList args, string name, int fallback) + { + return int.TryParse(ParseString(args, name, ""), out var value) ? value : fallback; + } +} + +internal sealed class EveSensorReceiver(EveSensorReceiverOptions options) : IDisposable +{ + private readonly TcpListener listener = new(IPAddress.Any, options.Port); + private readonly CancellationTokenSource stopping = new(); + private readonly ConcurrentDictionary subscribers = new(); + private int nextSubscriberId; + + public async Task RunAsync() + { + listener.Start(); + Console.Error.WriteLine($"Eve sensor receiver listening on ws://0.0.0.0:{options.Port}{options.Path} source={options.SourceId}"); + while (!stopping.IsCancellationRequested) + { + var client = await listener.AcceptTcpClientAsync(stopping.Token).ConfigureAwait(false); + _ = Task.Run(() => HandleAsync(client)); + } + } + + private async Task HandleAsync(TcpClient tcpClient) + { + await using var stream = tcpClient.GetStream(); + var request = await ReadHttpRequestAsync(stream).ConfigureAwait(false); + if (request.Path == "/health") + { + var health = JsonSerializer.Serialize(new + { + ok = true, + options.SourceId, + options.ExpectedType, + options.Path, + options.SubscribePath, + Subscribers = subscribers.Count, + }); + await WriteHttpResponseAsync(stream, "200 OK", "application/json", Encoding.UTF8.GetBytes(health)).ConfigureAwait(false); + return; + } + + if (string.Equals(request.Path, options.SubscribePath, StringComparison.Ordinal) + && request.Headers.TryGetValue("Sec-WebSocket-Key", out var subscriberKey)) + { + await WriteWebSocketHandshakeAsync(stream, subscriberKey).ConfigureAwait(false); + await SubscribeAsync(stream).ConfigureAwait(false); + return; + } + + if (!string.Equals(request.Path, options.Path, StringComparison.Ordinal) + || !request.Headers.TryGetValue("Sec-WebSocket-Key", out var key)) + { + await WriteHttpResponseAsync(stream, "404 Not Found", "text/plain", Encoding.UTF8.GetBytes("not found")).ConfigureAwait(false); + return; + } + + await WriteWebSocketHandshakeAsync(stream, key).ConfigureAwait(false); + while (!stopping.IsCancellationRequested) + { + var frame = await ReceiveFrameAsync(stream).ConfigureAwait(false); + if (frame.Opcode == 0x8) + { + await WriteCloseFrameAsync(stream).ConfigureAwait(false); + break; + } + + if (frame.Opcode == 0x2) + { + if (TryValidateCultMeshObservation(frame.Payload, out var normalizedObservation) + || TryValidateCultMeshMediaObservation(frame.Payload, out normalizedObservation)) + { + Console.WriteLine(normalizedObservation); + Console.Out.Flush(); + await BroadcastAsync(normalizedObservation).ConfigureAwait(false); + } + + continue; + } + + if (frame.Opcode != 0x1) + { + continue; + } + + var text = Encoding.UTF8.GetString(frame.Payload); + if (TryValidateFrameEvent(text, out var normalized)) + { + Console.WriteLine(normalized); + Console.Out.Flush(); + await BroadcastAsync(normalized).ConfigureAwait(false); + } + } + } + + private async Task SubscribeAsync(NetworkStream stream) + { + var id = Interlocked.Increment(ref nextSubscriberId); + var subscriber = new WebSocketSubscriber(id, stream); + subscribers[id] = subscriber; + Console.Error.WriteLine($"Eve sensor receiver subscriber {id} connected on {options.SubscribePath}."); + try + { + await subscriber.SendTextAsync(JsonSerializer.Serialize(new + { + type = "mimir-verse-subscription", + status = "connected", + options.Path, + options.SubscribePath, + options.SourceId, + options.ExpectedType, + })).ConfigureAwait(false); + + while (!stopping.IsCancellationRequested) + { + var frame = await ReceiveFrameAsync(stream).ConfigureAwait(false); + if (frame.Opcode == 0x8) + { + await WriteCloseFrameAsync(stream).ConfigureAwait(false); + break; + } + } + } + catch (IOException) + { + } + catch (SocketException) + { + } + finally + { + subscribers.TryRemove(id, out _); + Console.Error.WriteLine($"Eve sensor receiver subscriber {id} disconnected."); + } + } + + private async Task BroadcastAsync(string normalized) + { + foreach (var pair in subscribers.ToArray()) + { + try + { + await pair.Value.SendTextAsync(normalized).ConfigureAwait(false); + } + catch (IOException) + { + subscribers.TryRemove(pair.Key, out _); + } + catch (SocketException) + { + subscribers.TryRemove(pair.Key, out _); + } + catch (ObjectDisposedException) + { + subscribers.TryRemove(pair.Key, out _); + } + } + } + + private bool TryValidateFrameEvent(string text, out string normalized) + { + normalized = ""; + try + { + using var document = JsonDocument.Parse(text); + var root = document.RootElement; + if (!root.TryGetProperty("type", out var type) + || !string.Equals(type.GetString(), options.ExpectedType, StringComparison.OrdinalIgnoreCase)) + { + return false; + } + + if (!string.Equals(options.ExpectedType, "cultmesh-observation", StringComparison.OrdinalIgnoreCase) + && (!root.TryGetProperty("sourceId", out var sourceId) + || !string.Equals(sourceId.GetString(), options.SourceId, StringComparison.Ordinal))) + { + return false; + } + + normalized = root.GetRawText(); + return true; + } + catch (JsonException) + { + return false; + } + } + + private bool TryValidateCultMeshObservation(byte[] payload, out string normalized) + { + normalized = ""; + try + { + var observation = MessagePackSerializer.Deserialize(payload); + if (!string.Equals(observation.StreamId, options.SourceId, StringComparison.Ordinal) && + !string.Equals(options.ExpectedType, "cultmesh-observation", StringComparison.OrdinalIgnoreCase)) + { + return false; + } + + normalized = JsonSerializer.Serialize(new + { + type = "cultmesh-observation", + document = "mimir.eve_sensor_observation.v1", + observation.ObservationId, + observation.DeviceId, + observation.StreamId, + observation.Kind, + observation.Sequence, + observation.SensorTimestampNs, + observation.ElapsedRealtimeNs, + observation.WallClockUtc, + observation.ClockDomainId, + observation.Values, + observation.Action, + observation.PointerCount, + observation.X, + observation.Y, + observation.Accuracy, + }); + return true; + } + catch + { + return false; + } + } + + private bool TryValidateCultMeshMediaObservation(byte[] payload, out string normalized) + { + normalized = ""; + try + { + var observation = MessagePackSerializer.Deserialize(payload); + if (!string.Equals(observation.StreamId, options.SourceId, StringComparison.Ordinal) && + !string.Equals(options.ExpectedType, "cultmesh-observation", StringComparison.OrdinalIgnoreCase)) + { + return false; + } + + normalized = JsonSerializer.Serialize(new + { + type = "cultmesh-media-observation", + document = "mimir.eve_media_observation.v1", + observation.ObservationId, + observation.DeviceId, + observation.StreamId, + observation.Kind, + observation.Sequence, + observation.SensorTimestampNs, + observation.ElapsedRealtimeNs, + observation.WallClockUtc, + observation.ClockDomainId, + observation.Format, + observation.Width, + observation.Height, + observation.SampleRate, + observation.Channels, + observation.FrameCount, + observation.PayloadEncoding, + PayloadBytes = observation.Payload.Length, + }); + return true; + } + catch + { + return false; + } + } + + private static async Task ReadHttpRequestAsync(NetworkStream stream) + { + var bytes = new List(); + var lastFour = new Queue(4); + while (true) + { + var value = stream.ReadByte(); + if (value < 0) + { + throw new IOException("Client closed before HTTP headers completed."); + } + + var b = (byte)value; + bytes.Add(b); + lastFour.Enqueue(b); + if (lastFour.Count > 4) + { + lastFour.Dequeue(); + } + + if (lastFour.Count == 4 && lastFour.SequenceEqual(new byte[] { 13, 10, 13, 10 })) + { + break; + } + } + + var text = Encoding.ASCII.GetString(bytes.ToArray()); + var lines = text.Split("\r\n", StringSplitOptions.None); + var first = lines[0].Split(' '); + var headers = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var line in lines.Skip(1)) + { + var colon = line.IndexOf(':'); + if (colon > 0) + { + headers[line[..colon].Trim()] = line[(colon + 1)..].Trim(); + } + } + + await Task.CompletedTask.ConfigureAwait(false); + return new HttpRequest(first.Length > 1 ? first[1] : "/", headers); + } + + private static async Task WriteHttpResponseAsync(NetworkStream stream, string status, string contentType, byte[] body) + { + var header = Encoding.ASCII.GetBytes( + $"HTTP/1.1 {status}\r\nContent-Type: {contentType}\r\nContent-Length: {body.Length}\r\nConnection: close\r\n\r\n"); + await stream.WriteAsync(header).ConfigureAwait(false); + await stream.WriteAsync(body).ConfigureAwait(false); + } + + private static async Task WriteWebSocketHandshakeAsync(NetworkStream stream, string key) + { + var acceptBytes = SHA1.HashData(Encoding.ASCII.GetBytes(key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11")); + var accept = Convert.ToBase64String(acceptBytes); + var response = Encoding.ASCII.GetBytes( + "HTTP/1.1 101 Switching Protocols\r\n" + + "Upgrade: websocket\r\n" + + "Connection: Upgrade\r\n" + + $"Sec-WebSocket-Accept: {accept}\r\n\r\n"); + await stream.WriteAsync(response).ConfigureAwait(false); + } + + private static async Task ReceiveFrameAsync(NetworkStream stream) + { + var header = new byte[2]; + await ReadExactAsync(stream, header).ConfigureAwait(false); + var opcode = header[0] & 0x0f; + var masked = (header[1] & 0x80) != 0; + var length = header[1] & 0x7f; + if (length == 126) + { + var extended = new byte[2]; + await ReadExactAsync(stream, extended).ConfigureAwait(false); + length = (extended[0] << 8) | extended[1]; + } + else if (length == 127) + { + var extended = new byte[8]; + await ReadExactAsync(stream, extended).ConfigureAwait(false); + var longLength = 0L; + foreach (var b in extended) + { + longLength = (longLength << 8) | b; + } + + if (longLength > int.MaxValue) + { + throw new InvalidDataException("WebSocket frame is too large."); + } + + length = (int)longLength; + } + + var mask = new byte[4]; + if (masked) + { + await ReadExactAsync(stream, mask).ConfigureAwait(false); + } + + var payload = new byte[length]; + await ReadExactAsync(stream, payload).ConfigureAwait(false); + if (masked) + { + for (var index = 0; index < payload.Length; index++) + { + payload[index] ^= mask[index % 4]; + } + } + + return new WebSocketFrame(opcode, payload); + } + + private static Task WriteCloseFrameAsync(NetworkStream stream) + { + return stream.WriteAsync(new byte[] { 0x88, 0x00 }).AsTask(); + } + + private static async Task ReadExactAsync(NetworkStream stream, byte[] buffer) + { + var offset = 0; + while (offset < buffer.Length) + { + var read = await stream.ReadAsync(buffer.AsMemory(offset, buffer.Length - offset)).ConfigureAwait(false); + if (read == 0) + { + throw new IOException("WebSocket closed."); + } + + offset += read; + } + } + + public void Dispose() + { + stopping.Cancel(); + listener.Stop(); + stopping.Dispose(); + } +} + +internal sealed record HttpRequest(string Path, IReadOnlyDictionary Headers); + +internal sealed record WebSocketFrame(int Opcode, byte[] Payload); + +internal sealed class WebSocketSubscriber(int id, NetworkStream stream) +{ + private readonly SemaphoreSlim writeLock = new(1, 1); + + public int Id => id; + + public async Task SendTextAsync(string text) + { + var payload = Encoding.UTF8.GetBytes(text); + var header = new List { 0x81 }; + if (payload.Length < 126) + { + header.Add((byte)payload.Length); + } + else if (payload.Length <= ushort.MaxValue) + { + header.Add(126); + header.Add((byte)(payload.Length >> 8)); + header.Add((byte)(payload.Length & 0xff)); + } + else + { + header.Add(127); + for (var shift = 56; shift >= 0; shift -= 8) + { + header.Add((byte)(((ulong)payload.Length >> shift) & 0xff)); + } + } + + await writeLock.WaitAsync().ConfigureAwait(false); + try + { + await stream.WriteAsync(header.ToArray()).ConfigureAwait(false); + await stream.WriteAsync(payload).ConfigureAwait(false); + await stream.FlushAsync().ConfigureAwait(false); + } + finally + { + writeLock.Release(); + } + } +} diff --git a/src/Mimir.Runtime/Synchronization/MimirCultMeshContracts.cs b/src/Mimir.Runtime/Synchronization/MimirCultMeshContracts.cs index 3276d43..4de48b9 100644 --- a/src/Mimir.Runtime/Synchronization/MimirCultMeshContracts.cs +++ b/src/Mimir.Runtime/Synchronization/MimirCultMeshContracts.cs @@ -171,6 +171,50 @@ public sealed record MimirEveOperatorSurfaceDocument( [property: Key(6)] bool CanShowStats, [property: Key(7)] string[] CommandTopics); +[CultDocument("mimir.eve_sensor_observation", "mimir.eve_sensor_observation.v1")] +[MessagePackObject] +public sealed record MimirEveSensorObservationDocument( + [property: Key(0)] + [property: CultName] + string ObservationId, + [property: Key(1)] string DeviceId, + [property: Key(2)] string StreamId, + [property: Key(3)] string Kind, + [property: Key(4)] long Sequence, + [property: Key(5)] long SensorTimestampNs, + [property: Key(6)] long ElapsedRealtimeNs, + [property: Key(7)] string WallClockUtc, + [property: Key(8)] string ClockDomainId, + [property: Key(9)] double[] Values, + [property: Key(10)] string? Action, + [property: Key(11)] int? PointerCount, + [property: Key(12)] double? X, + [property: Key(13)] double? Y, + [property: Key(14)] int? Accuracy); + +[CultDocument("mimir.eve_media_observation", "mimir.eve_media_observation.v1")] +[MessagePackObject] +public sealed record MimirEveMediaObservationDocument( + [property: Key(0)] + [property: CultName] + string ObservationId, + [property: Key(1)] string DeviceId, + [property: Key(2)] string StreamId, + [property: Key(3)] string Kind, + [property: Key(4)] long Sequence, + [property: Key(5)] long SensorTimestampNs, + [property: Key(6)] long ElapsedRealtimeNs, + [property: Key(7)] string WallClockUtc, + [property: Key(8)] string ClockDomainId, + [property: Key(9)] string Format, + [property: Key(10)] int? Width, + [property: Key(11)] int? Height, + [property: Key(12)] int? SampleRate, + [property: Key(13)] int? Channels, + [property: Key(14)] int? FrameCount, + [property: Key(15)] string PayloadEncoding, + [property: Key(16)] byte[] Payload); + public static class MimirCultMeshContractFactory { public static MimirBioacousticCodebookState CreateCodebookState( diff --git a/src/Mimir.VerseRecorder/Mimir.VerseRecorder.csproj b/src/Mimir.VerseRecorder/Mimir.VerseRecorder.csproj new file mode 100644 index 0000000..05d41e6 --- /dev/null +++ b/src/Mimir.VerseRecorder/Mimir.VerseRecorder.csproj @@ -0,0 +1,12 @@ + + + + Exe + net10.0 + Mimir.VerseRecorder + Mimir.VerseRecorder + enable + enable + + + diff --git a/src/Mimir.VerseRecorder/Program.cs b/src/Mimir.VerseRecorder/Program.cs new file mode 100644 index 0000000..28a274c --- /dev/null +++ b/src/Mimir.VerseRecorder/Program.cs @@ -0,0 +1,151 @@ +using System.Net.WebSockets; +using System.Text; +using System.Text.Json; + +var options = VerseRecorderOptions.Parse(args); +Directory.CreateDirectory(options.OutputDirectory); +var runId = DateTimeOffset.Now.ToString("yyyyMMdd-HHmmss"); +var runDirectory = Path.Combine(options.OutputDirectory, runId); +Directory.CreateDirectory(runDirectory); +var jsonlPath = Path.Combine(runDirectory, "observations.jsonl"); +var sessionPath = Path.Combine(runDirectory, "session.json"); + +await File.WriteAllTextAsync( + sessionPath, + JsonSerializer.Serialize( + new + { + kind = "mimir.verse_recorder_session.v1", + runId, + options.Url, + startedAt = DateTimeOffset.Now.ToString("O"), + options.Seconds, + observations = jsonlPath, + }, + new JsonSerializerOptions { WriteIndented = true })); + +Console.Error.WriteLine($"Mimir Verse recorder subscribing to {options.Url}"); +Console.Error.WriteLine($"Writing {jsonlPath}"); + +using var stopping = new CancellationTokenSource(); +if (options.Seconds > 0) +{ + stopping.CancelAfter(TimeSpan.FromSeconds(options.Seconds)); +} + +Console.CancelKeyPress += (_, eventArgs) => +{ + eventArgs.Cancel = true; + stopping.Cancel(); +}; + +var count = 0L; +var lastMeter = DateTimeOffset.UtcNow; +await using var output = new StreamWriter(new FileStream(jsonlPath, FileMode.Create, FileAccess.Write, FileShare.Read), Encoding.UTF8); + +while (!stopping.IsCancellationRequested) +{ + using var socket = new ClientWebSocket(); + try + { + await socket.ConnectAsync(options.Url, stopping.Token); + while (socket.State == WebSocketState.Open && !stopping.IsCancellationRequested) + { + var text = await ReceiveTextAsync(socket, stopping.Token); + if (text == null) + { + break; + } + + await output.WriteLineAsync(text); + count++; + if (DateTimeOffset.UtcNow - lastMeter >= TimeSpan.FromSeconds(options.MeterSeconds)) + { + await output.FlushAsync(); + Console.Error.WriteLine($"verse-recorder observations={count} path={jsonlPath}"); + lastMeter = DateTimeOffset.UtcNow; + } + } + } + catch (OperationCanceledException) when (stopping.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + Console.Error.WriteLine($"verse-recorder reconnect after {ex.GetType().Name}: {ex.Message}"); + try + { + await Task.Delay(TimeSpan.FromSeconds(options.ReconnectSeconds), stopping.Token); + } + catch (OperationCanceledException) + { + break; + } + } +} + +await output.FlushAsync(); +Console.Error.WriteLine($"verse-recorder complete observations={count} path={jsonlPath}"); + +static async Task ReceiveTextAsync(ClientWebSocket socket, CancellationToken stopping) +{ + var buffer = new byte[64 * 1024]; + using var memory = new MemoryStream(); + while (true) + { + var result = await socket.ReceiveAsync(buffer, stopping); + if (result.MessageType == WebSocketMessageType.Close) + { + return null; + } + + memory.Write(buffer, 0, result.Count); + if (result.EndOfMessage) + { + return Encoding.UTF8.GetString(memory.ToArray()); + } + } +} + +internal sealed record VerseRecorderOptions( + Uri Url, + string OutputDirectory, + double Seconds, + double ReconnectSeconds, + double MeterSeconds) +{ + public static VerseRecorderOptions Parse(string[] args) + { + var defaultOutput = Path.Combine( + Environment.GetFolderPath(Environment.SpecialFolder.MyVideos), + "Mimir", + "VerseCaptures"); + return new VerseRecorderOptions( + new Uri(ParseString(args, "--url", "ws://127.0.0.1:8796/eve/periwinkle/subscribe")), + ParseString(args, "--out-dir", defaultOutput), + ParseDouble(args, "--seconds", 0.0), + ParseDouble(args, "--reconnect-seconds", 2.0), + ParseDouble(args, "--meter-seconds", 5.0)); + } + + private static string ParseString(IReadOnlyList args, string name, string fallback) + { + for (var index = 0; index < args.Count - 1; index++) + { + if (string.Equals(args[index], name, StringComparison.OrdinalIgnoreCase)) + { + return args[index + 1]; + } + } + + return fallback; + } + + private static double ParseDouble(IReadOnlyList args, string name, double fallback) + { + return double.TryParse(ParseString(args, name, ""), out var value) + ? value + : fallback; + } +} diff --git a/state/evidence.jsonl b/state/evidence.jsonl index ab7ff83..dda3f5f 100644 --- a/state/evidence.jsonl +++ b/state/evidence.jsonl @@ -124,3 +124,4 @@ {"ts": "2026-06-09T02:17:25+01:00", "type": "decision", "status": "ok", "note": "Mimir program composition authority moved out of OBS: Muninn hosts publish local stream capabilities/media; Mimir owns selected-stream subscription, scene graph, Eve operator surfaces, and program publication intent; Yggdrasil-facing publisher daemon consumes Mimir output; OBS is compatibility-only. Added typed mimir.program_scene.v1, mimir.program_output.v1, and mimir.eve_operator_surface.v1 contracts and BufferSmoke proof coverage."} {"ts": "2026-06-09T02:54:54+01:00", "type": "verification", "status": "ok", "note": "Imported the current OBS scene into Mimir-owned typed program state with Mimir.BufferSmoke --import-obs-program-scene. state/mimir-program-composition.cc now carries mimir.program_scene.v1, mimir.program_output.v1, and mimir.eve_operator_surface.v1 for eight current layers: Starfire monitor, Discord crop/chroma key, Raven monitor, Kiyo Pro, Mimir program texture, Muninn stream placeholder, SP browser, and hidden bigscreen."} {"ts": "2026-06-10T16:41:42+01:00", "type": "implementation", "status": "ok", "note": "Added typed PS Move tracking observations as Mimir runtime streams: MimirStreamKind.Tracking, mimir.move_tracking_observation.v1, local Starfire and remote Nightwing Move capture profiles, Muninn producer identity, Odin discovery provenance, runtime buffer readout, and BufferSmoke proofs for Starfire/Nightwing USB-attached Moves entering Mimir buffers."} +{"ts": "2026-06-10T17:38:22+01:00", "type": "verification", "status": "ok", "note": "Restored and launched the narrow Nightwing Move tracking witness path. scripts/start-nightwing-move-tracking.ps1 started Mimir.EveSensorReceiver on Starfire port 8796, staged Nightwing Eye/Move Python tools, launched Nightwing worker PID 1131578 with --move-light move-usb=/dev/hidraw1:#35ff6c, and VerseRecorder C:\\Users\\Meta\\Videos\\Mimir\\VerseCaptures\\20260610-173549 recorded mimir.psmove_light_state.v1 HID success plus mimir.move_controller_observation_state.v1 blobs from both ps3-eye-0 and ps3-eye-1."} diff --git a/state/map.yaml b/state/map.yaml index 2c038ae..23e3099 100644 --- a/state/map.yaml +++ b/state/map.yaml @@ -85,6 +85,12 @@ accepted_design: are local tracking streams, Nightwing Moves are emitted by Muninn on Nightwing, Odin owns discovery/schema projection, and Mimir owns buffering and fusion. + - `scripts/start-nightwing-move-tracking.ps1` restores the live Nightwing + Eye/Move witness path: Starfire owns the `/eve/periwinkle` receiver and + recorder, Nightwing owns PS Move HID report `0x06` LED writes and PS3 Eye + blob extraction, and the stream currently emits `mimir.psmove_light_state.v1` + plus `mimir.move_controller_observation_state.v1` optical witness records + before later fusion into full tracking poses. - `MimirFrameEventProcessStreamSource` is a temporary diagnostic adapter for JSON-line video frame and audio block metadata from native probes into runtime rolling buffers; it does not carry pixels/audio payloads or own the diff --git a/tools/nightwing_typed_witness_publisher.py b/tools/nightwing_typed_witness_publisher.py new file mode 100644 index 0000000..8b4648f --- /dev/null +++ b/tools/nightwing_typed_witness_publisher.py @@ -0,0 +1,493 @@ +#!/usr/bin/env python3 +"""Publish Nightwing witness status as typed Mimir CultMesh observations. + +This sends MessagePack `mimir.eve_sensor_observation.v1` documents to +Mimir.EveSensorReceiver. It is deliberately narrow: Nightwing owns local sensor +presence/timing receipts, while Mimir and Odin own synchronization and UI +projection. +""" + +from __future__ import annotations + +import argparse +import base64 +import datetime as dt +import glob +import math +import os +import random +import socket +import struct +import subprocess +import time +import urllib.parse +from dataclasses import dataclass + +try: + import nw_eye_cap + import nw_move_hint +except Exception: + nw_eye_cap = None + nw_move_hint = None + + +def pack_nil() -> bytes: + return b"\xc0" + + +def pack_bool(value: bool) -> bytes: + return b"\xc3" if value else b"\xc2" + + +def pack_int(value: int) -> bytes: + if 0 <= value <= 0x7f: + return bytes([value]) + if -32 <= value < 0: + return bytes([0xe0 | (value + 32)]) + if -(1 << 31) <= value < (1 << 31): + return b"\xd2" + struct.pack(">i", value) + return b"\xd3" + struct.pack(">q", value) + + +def pack_float(value: float) -> bytes: + return b"\xcb" + struct.pack(">d", float(value)) + + +def pack_str(value: str) -> bytes: + data = value.encode("utf-8") + if len(data) <= 31: + return bytes([0xa0 | len(data)]) + data + if len(data) <= 0xff: + return b"\xd9" + bytes([len(data)]) + data + if len(data) <= 0xffff: + return b"\xda" + struct.pack(">H", len(data)) + data + return b"\xdb" + struct.pack(">I", len(data)) + data + + +def pack_array(items: list[bytes]) -> bytes: + if len(items) <= 15: + return bytes([0x90 | len(items)]) + b"".join(items) + if len(items) <= 0xffff: + return b"\xdc" + struct.pack(">H", len(items)) + b"".join(items) + return b"\xdd" + struct.pack(">I", len(items)) + b"".join(items) + + +def pack_double_array(values: list[float]) -> bytes: + return pack_array([pack_float(value) for value in values]) + + +def pack_optional_string(value: str | None) -> bytes: + return pack_nil() if value is None else pack_str(value) + + +def pack_optional_int(value: int | None) -> bytes: + return pack_nil() if value is None else pack_int(value) + + +def sensor_document( + observation_id: str, + device_id: str, + stream_id: str, + kind: str, + sequence: int, + values: list[float], + accuracy: int | None = None, +) -> bytes: + now_ns = time.time_ns() + wall = dt.datetime.now(dt.timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") + return pack_array([ + pack_str(observation_id), + pack_str(device_id), + pack_str(stream_id), + pack_str(kind), + pack_int(sequence), + pack_int(now_ns), + pack_int(time.monotonic_ns()), + pack_str(wall), + pack_str("nightwing-monotonic"), + pack_double_array(values), + pack_optional_string(None), + pack_optional_int(None), + pack_nil(), + pack_nil(), + pack_optional_int(accuracy), + ]) + + +class WebSocket: + def __init__(self, url: str) -> None: + parsed = urllib.parse.urlparse(url) + if parsed.scheme != "ws": + raise ValueError("Only ws:// is supported") + self.host = parsed.hostname or "127.0.0.1" + self.port = parsed.port or 80 + self.path = parsed.path or "/" + self.sock: socket.socket | None = None + + def connect(self) -> None: + key = base64.b64encode(os.urandom(16)).decode("ascii") + sock = socket.create_connection((self.host, self.port), timeout=5) + request = ( + f"GET {self.path} HTTP/1.1\r\n" + f"Host: {self.host}:{self.port}\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\n" + "Sec-WebSocket-Version: 13\r\n" + "\r\n" + ) + sock.sendall(request.encode("ascii")) + response = b"" + while b"\r\n\r\n" not in response: + chunk = sock.recv(4096) + if not chunk: + raise ConnectionError("WebSocket handshake closed") + response += chunk + if b" 101 " not in response.splitlines()[0]: + raise ConnectionError(response.decode("utf-8", errors="replace").splitlines()[0]) + self.sock = sock + + def send_binary(self, payload: bytes) -> None: + self._send_frame(0x2, payload) + + def close(self) -> None: + if self.sock: + self.sock.close() + self.sock = None + + def _send_frame(self, opcode: int, payload: bytes) -> None: + if not self.sock: + raise ConnectionError("WebSocket not connected") + mask = random.randbytes(4) if hasattr(random, "randbytes") else os.urandom(4) + length = len(payload) + head = bytearray([0x80 | opcode]) + if length < 126: + head.append(0x80 | length) + elif length <= 0xffff: + head.extend([0x80 | 126]) + head.extend(struct.pack(">H", length)) + else: + head.extend([0x80 | 127]) + head.extend(struct.pack(">Q", length)) + masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload)) + self.sock.sendall(bytes(head) + mask + masked) + + +def video_devices() -> list[tuple[str, int, str]]: + devices: list[tuple[str, int, str]] = [] + for path in sorted(glob.glob("/sys/class/video4linux/video*")): + node = os.path.basename(path) + try: + index = int(node.replace("video", "")) + except ValueError: + continue + try: + with open(os.path.join(path, "name"), "r", encoding="utf-8", errors="replace") as handle: + name = handle.read().strip() + except OSError: + name = "video" + devices.append((node, index, name)) + return devices + + +@dataclass(frozen=True) +class MoveLight: + name: str + hidraw: str + rgb: tuple[int, int, int] + + +def parse_rgb(value: str) -> tuple[int, int, int]: + text = value.strip() + if text.startswith("#"): + text = text[1:] + if len(text) != 6: + raise ValueError(f"Expected #rrggbb, got {value!r}") + return (int(text[0:2], 16), int(text[2:4], 16), int(text[4:6], 16)) + + +def parse_move_light(value: str) -> MoveLight: + try: + name, rest = value.split("=", 1) + hidraw, rgb_text = rest.split(":", 1) + except ValueError as ex: + raise argparse.ArgumentTypeError("Move light must be name=/dev/hidrawN:#rrggbb") from ex + name = name.strip() + hidraw = hidraw.strip() + if not name or not hidraw: + raise argparse.ArgumentTypeError("Move light name and hidraw path are required") + try: + rgb = parse_rgb(rgb_text) + except ValueError as ex: + raise argparse.ArgumentTypeError(str(ex)) from ex + return MoveLight(name=name, hidraw=hidraw, rgb=rgb) + + +def write_move_light(move: MoveLight, rgb: tuple[int, int, int] | None = None) -> bool: + r, g, b = rgb if rgb is not None else move.rgb + try: + with open(move.hidraw, "wb", buffering=0) as device: + device.write(bytes([0x06, 0, r, g, b, 0, 0, 0, 0])) + return True + except OSError: + return False + + +def move_light_observations(args: argparse.Namespace, sequence: int) -> list[bytes]: + docs: list[bytes] = [] + for move in args.move_light: + ok = write_move_light(move) + r, g, b = move.rgb + docs.append(sensor_document( + observation_id=f"nightwing:{move.name}:move-light:{sequence}", + device_id="nightwing", + stream_id=f"nightwing-{move.name}", + kind="mimir.psmove_light_state.v1", + sequence=sequence, + values=[1.0 if ok else 0.0, float(r), float(g), float(b)], + accuracy=3 if ok else 0, + )) + return docs + + +def observations(sequence: int) -> list[bytes]: + docs: list[bytes] = [] + for node, index, name in video_devices(): + present = 1.0 if os.path.exists(f"/dev/{node}") else 0.0 + is_eye = 1.0 if "gspca" in name.lower() or "playstation" in name.lower() else 0.0 + docs.append(sensor_document( + observation_id=f"nightwing:{node}:{sequence}", + device_id="nightwing", + stream_id=f"nightwing-{node}", + kind="camera-device-status", + sequence=sequence, + values=[present, float(index), is_eye], + accuracy=3 if present else 0, + )) + docs.append(sensor_document( + observation_id=f"nightwing:leap-tracking:{sequence}", + device_id="nightwing", + stream_id="nightwing-leap-tracking", + kind="leap-tracking-status", + sequence=sequence, + values=[0.0], + accuracy=0, + )) + return docs + + +def builtin_camera_observations(args: argparse.Namespace, sequence: int) -> list[bytes]: + docs: list[bytes] = [] + for node, index, name in video_devices(): + lname = name.lower() + is_eye = "gspca" in lname or "playstation" in lname or "eye" in lname + if is_eye: + continue + docs.append(sensor_document( + observation_id=f"nightwing:{node}:builtin-camera:{sequence}", + device_id="nightwing", + stream_id=f"nightwing-{node}", + kind="mimir.nightwing_builtin_camera_witness.v1", + sequence=sequence, + values=[1.0, float(index), float(len(name)), float(args.interval)], + accuracy=2, + )) + if not docs: + docs.append(sensor_document( + observation_id=f"nightwing:builtin-camera:missing:{sequence}", + device_id="nightwing", + stream_id="nightwing-builtin-camera", + kind="mimir.nightwing_builtin_camera_witness.v1", + sequence=sequence, + values=[0.0], + accuracy=0, + )) + return docs + + +def builtin_mic_observation(args: argparse.Namespace, sequence: int) -> bytes: + values = [0.0, 0.0, 0.0, float(args.mic_sample_rate)] + accuracy = 0 + try: + result = subprocess.run( + [ + "timeout", + str(max(0.05, args.mic_window_seconds)), + "arecord", + "-D", + args.mic_device, + "-f", + "S16_LE", + "-r", + str(args.mic_sample_rate), + "-c", + "1", + "-t", + "raw", + "-q", + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + check=False, + timeout=max(0.2, args.mic_window_seconds + 0.2), + ) + data = result.stdout + count = len(data) // 2 + if count > 0: + samples = struct.unpack("<" + "h" * count, data[:count * 2]) + rms = math.sqrt(sum(float(sample) * float(sample) for sample in samples) / count) / 32768.0 + peak = max(abs(sample) for sample in samples) / 32768.0 + values = [1.0, rms, peak, float(args.mic_sample_rate), float(count)] + accuracy = 2 if rms > 0.0 else 1 + except Exception: + values = [0.0, 0.0, 0.0, float(args.mic_sample_rate)] + accuracy = 0 + return sensor_document( + observation_id=f"nightwing:builtin-mic:{sequence}", + device_id="nightwing", + stream_id="nightwing-builtin-mic", + kind="mimir.nightwing_builtin_mic_witness.v1", + sequence=sequence, + values=values, + accuracy=accuracy, + ) + + +def eye_tracking_observations(args: argparse.Namespace, sequence: int) -> list[bytes]: + if nw_eye_cap is None or nw_move_hint is None: + return [] + docs: list[bytes] = [] + devices = [part for part in args.eye_devices.split(",") if part] + for eye_index, device in enumerate(devices): + raw_path = None + source_id = f"ps3-eye-{eye_index}" + stats_path = f"/tmp/mimir-mvp-{source_id}.json" + cap_args = argparse.Namespace( + device=device, + width=args.eye_width, + height=args.eye_height, + format="YUYV", + fps=args.eye_fps, + seconds=args.eye_window_seconds, + buffers=8, + out=stats_path, + stdout=False, + ) + try: + stats = nw_eye_cap.capture(cap_args) + raw_path = stats["frames_path"] + width = int(stats["width"]) + height = int(stats["height"]) + frame_bytes = width * height + with open(raw_path, "rb") as handle: + raw = handle.read() + frame_count = min(len(raw) // frame_bytes, len(stats.get("timestamps_s", []))) + stride = max(1, int(args.tracking_stride)) + for frame_index in range(0, frame_count, stride): + frame = raw[frame_index * frame_bytes:(frame_index + 1) * frame_bytes] + blob = nw_move_hint.best_blob(frame, width, height) + if blob is None or float(blob["confidence"]) < args.min_confidence: + continue + values = [ + float(blob["clip_x"]), + float(blob["clip_y"]), + float(blob["area_px"]), + float(blob["radius_px"]), + float(blob["mean_luma"]), + float(blob["peak_luma"]), + float(blob["confidence"]), + float(frame_index), + float(width), + float(height), + ] + docs.append(sensor_document( + observation_id=f"nightwing:{source_id}:move-sphere:{sequence}:{frame_index}", + device_id="nightwing", + stream_id=source_id, + kind="mimir.move_controller_observation_state.v1", + sequence=sequence * 100000 + frame_index, + values=values, + accuracy=3, + )) + except Exception as ex: + docs.append(sensor_document( + observation_id=f"nightwing:{source_id}:tracking-error:{sequence}", + device_id="nightwing", + stream_id=source_id, + kind=f"tracking-error:{type(ex).__name__}", + sequence=sequence, + values=[0.0], + accuracy=0, + )) + finally: + try: + if raw_path: + os.unlink(raw_path) + except OSError: + pass + return docs + + +def run(args: argparse.Namespace) -> None: + sequence = 0 + while True: + ws = WebSocket(args.url) + try: + ws.connect() + while True: + for payload in observations(sequence): + ws.send_binary(payload) + if args.track_builtin_camera: + for payload in builtin_camera_observations(args, sequence): + ws.send_binary(payload) + if args.track_builtin_mic: + ws.send_binary(builtin_mic_observation(args, sequence)) + for payload in move_light_observations(args, sequence): + ws.send_binary(payload) + if args.track_eyes: + for payload in eye_tracking_observations(args, sequence): + ws.send_binary(payload) + sequence += 1 + if args.once: + return + time.sleep(max(0.02, args.interval)) + finally: + ws.close() + if args.once: + return + time.sleep(args.reconnect_seconds) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--url", default="ws://192.168.1.66:8796/eve/periwinkle") + parser.add_argument("--interval", type=float, default=0.25) + parser.add_argument("--reconnect-seconds", type=float, default=2.0) + parser.add_argument("--once", action="store_true") + parser.add_argument("--track-eyes", action="store_true", help="Publish compact Nightwing-local PS3 Eye Move-sphere observations.") + parser.add_argument("--track-builtin-camera", action="store_true", help="Publish compact status for Nightwing-local non-Eye camera devices.") + parser.add_argument("--track-builtin-mic", action="store_true", help="Publish compact ALSA microphone level observations.") + parser.add_argument( + "--move-light", + action="append", + type=parse_move_light, + default=[], + help="Keep a Nightwing-local PS Move sphere lit: name=/dev/hidrawN:#rrggbb.", + ) + parser.add_argument("--eye-devices", default="/dev/video2,/dev/video3") + parser.add_argument("--eye-width", type=int, default=320) + parser.add_argument("--eye-height", type=int, default=240) + parser.add_argument("--eye-fps", type=int, default=187) + parser.add_argument("--eye-window-seconds", type=float, default=0.12) + parser.add_argument("--tracking-stride", type=int, default=6) + parser.add_argument("--min-confidence", type=float, default=0.25) + parser.add_argument("--mic-device", default="default") + parser.add_argument("--mic-sample-rate", type=int, default=16000) + parser.add_argument("--mic-window-seconds", type=float, default=0.08) + args = parser.parse_args() + run(args) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/nw_eye_cap.py b/tools/nw_eye_cap.py new file mode 100644 index 0000000..dbc44d9 --- /dev/null +++ b/tools/nw_eye_cap.py @@ -0,0 +1,280 @@ +#!/usr/bin/env python3 +"""Capture raw luma frames from Nightwing V4L2 cameras for sync receipts.""" + +from __future__ import annotations + +import argparse +import ctypes +import ctypes.util +import fcntl +import json +import mmap +import os +import select +import sys +import time +from pathlib import Path + + +V4L2_BUF_TYPE_VIDEO_CAPTURE = 1 +V4L2_FIELD_NONE = 1 +V4L2_MEMORY_MMAP = 1 +VIDIOC_G_FMT = 0xC0D05604 +VIDIOC_S_FMT = 0xC0D05605 +VIDIOC_REQBUFS = 0xC0145608 +VIDIOC_QUERYBUF = 0xC0585609 +VIDIOC_QBUF = 0xC058560F +VIDIOC_DQBUF = 0xC0585611 +VIDIOC_STREAMON = 0x40045612 +VIDIOC_STREAMOFF = 0x40045613 +VIDIOC_G_PARM = 0xC0CC5615 +VIDIOC_S_PARM = 0xC0CC5616 + +_libc = ctypes.CDLL(ctypes.util.find_library("c") or "libc.so.6", use_errno=True) +_libc.ioctl.argtypes = [ctypes.c_int, ctypes.c_ulong, ctypes.c_void_p] +_libc.ioctl.restype = ctypes.c_int + + +def fourcc(text: str) -> int: + raw = text.encode("ascii") + return raw[0] | (raw[1] << 8) | (raw[2] << 16) | (raw[3] << 24) + + +class V4L2PixFormat(ctypes.Structure): + _fields_ = [ + ("width", ctypes.c_uint32), + ("height", ctypes.c_uint32), + ("pixelformat", ctypes.c_uint32), + ("field", ctypes.c_uint32), + ("bytesperline", ctypes.c_uint32), + ("sizeimage", ctypes.c_uint32), + ("colorspace", ctypes.c_uint32), + ("priv", ctypes.c_uint32), + ("flags", ctypes.c_uint32), + ("ycbcr_enc", ctypes.c_uint32), + ("quantization", ctypes.c_uint32), + ("xfer_func", ctypes.c_uint32), + ] + + +class V4L2FormatUnion(ctypes.Union): + _fields_ = [("pix", V4L2PixFormat), ("raw", ctypes.c_uint8 * 200), ("_align", ctypes.c_uint64)] + + +class V4L2Format(ctypes.Structure): + _fields_ = [("type", ctypes.c_uint32), ("fmt", V4L2FormatUnion)] + + +class V4L2RequestBuffers(ctypes.Structure): + _fields_ = [ + ("count", ctypes.c_uint32), + ("type", ctypes.c_uint32), + ("memory", ctypes.c_uint32), + ("capabilities", ctypes.c_uint32), + ("flags", ctypes.c_uint8), + ("reserved", ctypes.c_uint8 * 3), + ] + + +class V4L2Fract(ctypes.Structure): + _fields_ = [("numerator", ctypes.c_uint32), ("denominator", ctypes.c_uint32)] + + +class V4L2CaptureParm(ctypes.Structure): + _fields_ = [ + ("capability", ctypes.c_uint32), + ("capturemode", ctypes.c_uint32), + ("timeperframe", V4L2Fract), + ("extendedmode", ctypes.c_uint32), + ("readbuffers", ctypes.c_uint32), + ("reserved", ctypes.c_uint32 * 4), + ] + + +class V4L2StreamParmUnion(ctypes.Union): + _fields_ = [("capture", V4L2CaptureParm), ("raw", ctypes.c_uint8 * 200)] + + +class V4L2StreamParm(ctypes.Structure): + _fields_ = [("type", ctypes.c_uint32), ("parm", V4L2StreamParmUnion)] + + +class Timeval(ctypes.Structure): + _fields_ = [("tv_sec", ctypes.c_long), ("tv_usec", ctypes.c_long)] + + +class V4L2Timecode(ctypes.Structure): + _fields_ = [ + ("type", ctypes.c_uint32), + ("flags", ctypes.c_uint32), + ("frames", ctypes.c_uint8), + ("seconds", ctypes.c_uint8), + ("minutes", ctypes.c_uint8), + ("hours", ctypes.c_uint8), + ("userbits", ctypes.c_uint8 * 4), + ] + + +class V4L2BufferUnion(ctypes.Union): + _fields_ = [("offset", ctypes.c_uint32), ("userptr", ctypes.c_ulong), ("planes", ctypes.c_void_p), ("fd", ctypes.c_int32)] + + +class V4L2Buffer(ctypes.Structure): + _fields_ = [ + ("index", ctypes.c_uint32), + ("type", ctypes.c_uint32), + ("bytesused", ctypes.c_uint32), + ("flags", ctypes.c_uint32), + ("field", ctypes.c_uint32), + ("timestamp", Timeval), + ("timecode", V4L2Timecode), + ("sequence", ctypes.c_uint32), + ("memory", ctypes.c_uint32), + ("m", V4L2BufferUnion), + ("length", ctypes.c_uint32), + ("reserved2", ctypes.c_uint32), + ("request_fd", ctypes.c_int32), + ] + + +def xioctl(fd: int, request: int, obj: ctypes.Structure) -> None: + if _libc.ioctl(fd, request, ctypes.byref(obj)) != 0: + error = ctypes.get_errno() + raise OSError(error, os.strerror(error)) + + +def set_frame_interval(fd: int, fps: int) -> None: + parm = V4L2StreamParm() + parm.type = V4L2_BUF_TYPE_VIDEO_CAPTURE + try: + xioctl(fd, VIDIOC_G_PARM, parm) + except OSError: + pass + parm.parm.capture.timeperframe.numerator = 1 + parm.parm.capture.timeperframe.denominator = fps + xioctl(fd, VIDIOC_S_PARM, parm) + + +def yuyv_to_luma(raw: bytes, width: int, height: int, stride: int) -> bytes: + rows = [] + for row in range(height): + start = row * stride + line = raw[start:start + width * 2] + rows.append(line[0::2]) + return b"".join(rows) + + +def capture(args: argparse.Namespace) -> dict: + fd = os.open(args.device, os.O_RDWR | os.O_NONBLOCK) + maps: list[mmap.mmap] = [] + stream_stdout = bool(getattr(args, "stdout", False)) + out = Path(args.out) if args.out else None + if out is not None: + out.parent.mkdir(parents=True, exist_ok=True) + frames_path = out.with_suffix(".y8") if out is not None else None + try: + set_frame_interval(fd, args.fps) + fmt = V4L2Format() + fmt.type = V4L2_BUF_TYPE_VIDEO_CAPTURE + xioctl(fd, VIDIOC_G_FMT, fmt) + fmt.fmt.pix.width = args.width + fmt.fmt.pix.height = args.height + fmt.fmt.pix.pixelformat = fourcc(args.format) + fmt.fmt.pix.field = V4L2_FIELD_NONE + xioctl(fd, VIDIOC_S_FMT, fmt) + actual = fmt.fmt.pix + + req = V4L2RequestBuffers() + req.count = args.buffers + req.type = V4L2_BUF_TYPE_VIDEO_CAPTURE + req.memory = V4L2_MEMORY_MMAP + xioctl(fd, VIDIOC_REQBUFS, req) + for index in range(req.count): + buf = V4L2Buffer() + buf.type = V4L2_BUF_TYPE_VIDEO_CAPTURE + buf.memory = V4L2_MEMORY_MMAP + buf.index = index + xioctl(fd, VIDIOC_QUERYBUF, buf) + maps.append(mmap.mmap(fd, buf.length, mmap.MAP_SHARED, mmap.PROT_READ | mmap.PROT_WRITE, offset=buf.m.offset)) + xioctl(fd, VIDIOC_QBUF, buf) + + stream_type = ctypes.c_int(V4L2_BUF_TYPE_VIDEO_CAPTURE) + fcntl.ioctl(fd, VIDIOC_STREAMON, stream_type) + start = time.monotonic() + deadline = start + args.seconds + timestamps = [] + sequences = [] + frames = sys.stdout.buffer if stream_stdout else frames_path.open("wb") # type: ignore[union-attr] + try: + while time.monotonic() < deadline: + ready, _, _ = select.select([fd], [], [], 0.25) + if not ready: + continue + buf = V4L2Buffer() + buf.type = V4L2_BUF_TYPE_VIDEO_CAPTURE + buf.memory = V4L2_MEMORY_MMAP + try: + xioctl(fd, VIDIOC_DQBUF, buf) + except OSError: + continue + raw = maps[buf.index][:buf.bytesused] + frames.write(yuyv_to_luma(raw, int(actual.width), int(actual.height), int(actual.bytesperline))) + timestamps.append(float(buf.timestamp.tv_sec) + float(buf.timestamp.tv_usec) / 1_000_000.0) + sequences.append(int(buf.sequence)) + xioctl(fd, VIDIOC_QBUF, buf) + finally: + if not stream_stdout: + frames.close() + try: + fcntl.ioctl(fd, VIDIOC_STREAMOFF, stream_type) + except OSError: + pass + first_ts = timestamps[0] if timestamps else None + rel = [(ts - first_ts) for ts in timestamps] if first_ts is not None else [] + gap = max(0, (sequences[-1] - sequences[0] + 1) - len(sequences)) if len(sequences) >= 2 else 0 + stats = { + "device": args.device, + "frames_path": str(frames_path) if frames_path is not None else "stdout", + "width": int(actual.width), + "height": int(actual.height), + "stride": int(actual.bytesperline), + "format": args.format, + "frames": len(timestamps), + "timestamps_s": rel, + "first_device_timestamp_s": first_ts, + "sequence_gap": gap, + "duration_s": (rel[-1] - rel[0]) if len(rel) >= 2 else 0.0, + } + if out is not None: + out.write_text(json.dumps(stats, indent=2, sort_keys=True), encoding="utf-8") + else: + print(json.dumps(stats, sort_keys=True), file=sys.stderr) + return stats + finally: + for mapping in maps: + mapping.close() + os.close(fd) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--device", required=True) + parser.add_argument("--width", type=int, default=320) + parser.add_argument("--height", type=int, default=240) + parser.add_argument("--format", default="YUYV") + parser.add_argument("--fps", type=int, default=187) + parser.add_argument("--seconds", type=float, default=8.0) + parser.add_argument("--buffers", type=int, default=8) + parser.add_argument("--out", default=None) + parser.add_argument("--stdout", action="store_true", help="Stream raw luma frames to stdout and write stats to stderr.") + args = parser.parse_args() + if not args.stdout and not args.out: + parser.error("--out is required unless --stdout is set") + stats = capture(args) + if not args.stdout: + print(json.dumps(stats, sort_keys=True)) + return 0 if stats["frames"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/nw_move_hint.py b/tools/nw_move_hint.py new file mode 100644 index 0000000..2b5c9fb --- /dev/null +++ b/tools/nw_move_hint.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Nightwing-local PS Eye / PS Move localization hint extractor. + +Stdlib-only on purpose. Nightwing owns Eye frames and Move observations; the +wire format back to Starfire is compact witness state, not raw media. +""" + +from __future__ import annotations + +import argparse +import json +import math +from pathlib import Path + +import nw_eye_cap + + +def percentile(values: list[float], q: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + index = max(0, min(len(ordered) - 1, int(round((len(ordered) - 1) * q)))) + return float(ordered[index]) + + +def best_blob(frame: bytes, width: int, height: int) -> dict | None: + sampled = [frame[index] for index in range(0, len(frame), max(1, len(frame) // 4000))] + threshold = max(40, int(percentile([float(v) for v in sampled], 0.995))) + seen = bytearray(width * height) + best: dict | None = None + for y in range(height): + row = y * width + for x in range(width): + start = row + x + if seen[start] or frame[start] < threshold: + continue + stack = [start] + seen[start] = 1 + count = 0 + sx = 0.0 + sy = 0.0 + sw = 0.0 + total = 0.0 + peak = 0 + while stack: + index = stack.pop() + yy, xx = divmod(index, width) + value = int(frame[index]) + weight = max(1.0, float(value - threshold + 1)) + count += 1 + sx += xx * weight + sy += yy * weight + sw += weight + total += value + peak = max(peak, value) + for nx, ny in ((xx - 1, yy), (xx + 1, yy), (xx, yy - 1), (xx, yy + 1)): + if 0 <= nx < width and 0 <= ny < height: + ni = ny * width + nx + if not seen[ni] and frame[ni] >= threshold: + seen[ni] = 1 + stack.append(ni) + if count < 3 or sw <= 0.0: + continue + cx = sx / sw + cy = sy / sw + mean_luma = total / count + confidence = min(1.0, (mean_luma / 255.0) * min(1.0, count / 80.0) * 1.6) + blob = { + "x_px": cx, + "y_px": cy, + "clip_x": (cx / max(1, width - 1)) * 2.0 - 1.0, + "clip_y": 1.0 - (cy / max(1, height - 1)) * 2.0, + "area_px": count, + "radius_px": math.sqrt(count / math.pi), + "mean_luma": mean_luma, + "peak_luma": peak, + "confidence": confidence, + } + if best is None or (blob["confidence"], blob["area_px"], blob["peak_luma"]) > (best["confidence"], best["area_px"], best["peak_luma"]): + best = blob + return best + + +def analyze(raw_path: Path, stats: dict, events: list[dict], moves: list[str], pulse_seconds: float, source_id: str) -> dict: + width = int(stats["width"]) + height = int(stats["height"]) + frame_bytes = width * height + raw = raw_path.read_bytes() + count = min(len(raw) // frame_bytes, len(stats.get("timestamps_s", []))) + timestamps = [float(value) for value in stats.get("timestamps_s", [])[:count]] + observations = [] + for event in events: + event_index = int(event["index"]) + expected = float(event["offset"]) + move_id = moves[event_index % len(moves)] if moves else "move" + best: dict | None = None + for frame_index, timestamp in enumerate(timestamps): + if timestamp < expected - 0.18 or timestamp > expected + max(0.25, pulse_seconds + 0.18): + continue + frame = raw[frame_index * frame_bytes:(frame_index + 1) * frame_bytes] + blob = best_blob(frame, width, height) + if blob is None: + continue + blob.update({ + "event_index": event_index, + "symbol": int(event["symbol"]), + "controller_id": move_id, + "source_id": source_id, + "observed_time_s": timestamp, + "schedule_offset_s": expected, + "schedule_residual_ms": (timestamp - expected) * 1000.0, + }) + if best is None or blob["confidence"] > best["confidence"]: + best = blob + if best is not None: + observations.append(best) + return { + "kind": "mimir.move_controller_observation_state.v1", + "source_id": source_id, + "width": width, + "height": height, + "frame_count": count, + "first_device_timestamp_s": stats.get("first_device_timestamp_s"), + "duration_s": stats.get("duration_s"), + "sequence_gap": stats.get("sequence_gap"), + "observations": observations, + "observation_count": len(observations), + "stable_event_count": len([item for item in observations if float(item["confidence"]) >= 0.25]), + "mean_confidence": sum(float(item["confidence"]) for item in observations) / len(observations) if observations else 0.0, + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--device", required=True) + parser.add_argument("--source-id", required=True) + parser.add_argument("--width", type=int, default=320) + parser.add_argument("--height", type=int, default=240) + parser.add_argument("--fps", type=int, default=187) + parser.add_argument("--seconds", type=float, default=8.0) + parser.add_argument("--events-json", required=True) + parser.add_argument("--moves", default="move-a,move-b") + parser.add_argument("--pulse-ms", type=float, default=95.0) + parser.add_argument("--out", required=True) + args = parser.parse_args() + + out = Path(args.out) + raw_stats_path = out.with_suffix(".capture.json") + cap_args = argparse.Namespace( + device=args.device, + width=args.width, + height=args.height, + format="YUYV", + fps=args.fps, + seconds=args.seconds, + buffers=8, + out=str(raw_stats_path), + ) + stats = nw_eye_cap.capture(cap_args) + events = json.loads(Path(args.events_json).read_text(encoding="utf-8")) + moves = [part for part in args.moves.split(",") if part] + result = analyze(Path(stats["frames_path"]), stats, events, moves, args.pulse_ms * 0.001, args.source_id) + out.write_text(json.dumps(result, indent=2, sort_keys=True), encoding="utf-8") + Path(stats["frames_path"]).unlink(missing_ok=True) + print(json.dumps(result, sort_keys=True)) + return 0 if result["observation_count"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main())