zgaetano/rainbow-dragon
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

logger+cli: e2e test (sim->driver->logger->store->all exports), forge export subcommand (ale/csv/edl/ccc/cube); fix supervisor subscription race w/ lock-guarded continuations — 8x soak stable @ 111 tests

Browse source
This commit is contained in:
Forge Dev 2026-07-10 19:57:29 +00:00
parent 80c5ab0604
commit bb1ef127ed
4 changed files with 210 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

47
Sources/ForgeCamera/ConnectionSupervisor.swift
View file

@ -39,7 +39,34 @@ public actor ConnectionSupervisor {
public private(set) var health: ConnectionStatus = .disconnected
private var stateContinuations: [UUID: AsyncStream<CameraState>.Continuation] = [:]
/// Subscriber continuations behind a lock — registration is synchronous
/// at subscription time, so states emitted immediately after `states`
/// is accessed are never dropped (async actor-hop registration raced).
private final class SubscriberBox: @unchecked Sendable {
let lock = NSLock()
var continuations: [UUID: AsyncStream<CameraState>.Continuation] = [:]
func add(_ id: UUID, _ c: AsyncStream<CameraState>.Continuation) {
lock.lock()
continuations[id] = c
lock.unlock()
}
func remove(_ id: UUID) {
lock.lock()
continuations.removeValue(forKey: id)
lock.unlock()
}
func yieldAll(_ s: CameraState) {
lock.lock()
let conts = Array(continuations.values)
lock.unlock()
for c in conts { c.yield(s) }
}
}
private nonisolated let subscribers = SubscriberBox()
private var connectionWaiters: [UUID] = []
private var waiterMap: [UUID: CheckedContinuation<Void, Never>] = [:]
@ -52,21 +79,13 @@ public actor ConnectionSupervisor {
public nonisolated var states: AsyncStream<CameraState> {
AsyncStream { continuation in
let id = UUID()
Task { await self.register(id: id, continuation: continuation) }
continuation.onTermination = { _ in
Task { await self.unregister(id: id) }
subscribers.add(id, continuation)
continuation.onTermination = { [subscribers] _ in
subscribers.remove(id)
}
}
}
private func register(id: UUID, continuation: AsyncStream<CameraState>.Continuation) {
stateContinuations[id] = continuation
}
private func unregister(id: UUID) {
stateContinuations.removeValue(forKey: id)
}
public func start() throws {
guard connectTask == nil else { return }
health = .connecting
@ -102,9 +121,7 @@ public actor ConnectionSupervisor {
backoff.reset()
resumeAllWaiters()
}
for c in stateContinuations.values {
c.yield(state)
}
subscribers.yieldAll(state)
}
/// Await connected health. Returns false on timeout.

4
Sources/ForgeCameraARRI/ArriDriver.swift
View file

@ -32,6 +32,10 @@ public actor ArriDriver: CameraDriver {
private var pollTask: Task<Void, Never>?
private var lastState = CameraState(connection: .disconnected)
// NOTE: registration is async (actor hop). Safe here: the poll loop
// re-emits on every state *change* vs lastState, so a subscriber attached
// moments late still receives the next change. (A lock-based
// nonisolated-let variant crashed swiftc 6.0.3 codegen on Linux.)
private var stateContinuations: [UUID: AsyncStream<CameraState>.Continuation] = [:]
public init(host: String, port: Int, pollInterval: Duration = .milliseconds(150)) {

84
Sources/forge-cli/Main.swift
View file

@ -5,13 +5,95 @@ import ForgeCameraARRI
import ForgeGrade
import ForgeColor
import ForgeSim
import ForgeLibrary
import ForgeExport
@main
struct Forge: AsyncParsableCommand {
static let configuration = CommandConfiguration(
commandName: "forge",
abstract: "On-set live grading: push looks to cameras over IP.",
subcommands: [Push.self, Status.self, Sim.self])
subcommands: [Push.self, Status.self, Sim.self, Export.self])
}
struct Export: AsyncParsableCommand {
static let configuration = CommandConfiguration(abstract: "Export a day's shots for post.")
@Option(name: .long, help: "Library database path")
var db: String
@Option(name: .long, help: "Day label (e.g. 'Day 01')")
var day: String
@Option(name: .long, help: "Format: ale|csv|edl|ccc|cube")
var format: String
@Option(name: .long, help: "Output file (or directory for cube)")
var out: String
@Option(name: .long, help: "Timebase fps for ALE/EDL")
var fps: Int = 24
@Option(name: .long, help: "Cube lattice size")
var lattice: Int = 33
func run() async throws {
let store = try LibraryStore(path: db)
// Find day by label across all projects.
var target: ForgeLibrary.Day?
for p in try store.listProjects() {
if let d = try store.listDays(projectID: p.id).first(where: { $0.label == day }) {
target = d
break
}
}
guard let target else {
throw ValidationError("day '\(day)' not found in \(db)")
}
let shots = try store.listShots(dayID: target.id)
guard !shots.isEmpty else {
throw ValidationError("no shots logged for '\(day)'")
}
func firstCDL(_ s: Shot) -> ForgeColor.CDL? {
guard let data = s.gradeSnapshot,
let stack = try? GradeSnapshot.decode(data) else { return nil }
for n in stack.nodes {
if case .cdl(let c) = n.kind, n.isEnabled { return c }
}
return nil
}
switch format {
case "ale":
try PostExport.writeALE(shots: shots, fps: fps)
.write(toFile: out, atomically: true, encoding: .utf8)
case "csv":
try PostExport.writeCSV(shots: shots)
.write(toFile: out, atomically: true, encoding: .utf8)
case "edl":
try PostExport.writeEDL(shots: shots, title: day, fps: fps)
.write(toFile: out, atomically: true, encoding: .utf8)
case "ccc":
let entries = shots.compactMap { s in firstCDL(s).map { (s.clipName, $0) } }
try CDLExport.writeCCC(entries)
.write(toFile: out, atomically: true, encoding: .utf8)
case "cube":
try FileManager.default.createDirectory(atPath: out, withIntermediateDirectories: true)
var written = 0
for s in shots {
guard let snap = s.gradeSnapshot else { continue }
let cube = try CubeExport.fromSnapshot(snap, latticeSize: lattice, title: s.clipName)
try cube.write(toFile: "\(out)/\(s.clipName).cube", atomically: true, encoding: .utf8)
written += 1
}
print("wrote \(written) cubes to \(out)/")
return
default:
throw ValidationError("unknown format '\(format)' (ale|csv|edl|ccc|cube)")
}
print("wrote \(format): \(out) (\(shots.count) shots)")
}
}
struct CameraOptions: ParsableArguments {

91
Tests/ForgeLoggerTests/EndToEndExportTests.swift Normal file
View file

@ -0,0 +1,91 @@
import XCTest
import Foundation
import ForgeCamera
import ForgeCameraARRI
import ForgeColor
import ForgeGrade
import ForgeLibrary
import ForgeExport
import ForgeSim
@testable import ForgeLogger
/// Full pipeline: ARRI sim -> driver poll -> ClipLogger -> LibraryStore -> exports.
final class EndToEndExportTests: XCTestCase {
func testSimSessionToExports() async throws {
// Infra.
let sim = ArriSimulator()
try await sim.start(port: 0)
defer { Task { try? await sim.stop() } }
let port = await sim.boundPort!
let dbPath = NSTemporaryDirectory() + "forge-e2e-\(UUID().uuidString).db"
defer { try? FileManager.default.removeItem(atPath: dbPath) }
let store = try LibraryStore(path: dbPath)
let project = try store.createProject(name: "E2E")
let day = try store.createDay(projectID: project.id, label: "Day 01", date: "2026-07-10")
let driver = ArriDriver(host: "127.0.0.1", port: port, pollInterval: .milliseconds(15))
let logger = ClipLogger(store: store, dayID: day.id)
let grade = GradeStack(nodes: [GradeNode(kind: .cdl(CDL(
slope: SIMD3(1.15, 1.0, 0.92), offset: SIMD3(0.01, 0, -0.01),
power: .one, saturation: 1.1)))])
let snapshotForProvider = grade
await logger.attach(slotName: "A-Cam", driver: driver) { snapshotForProvider }
try await driver.connect()
// Three takes.
let takes: [(clip: String, tcIn: String, tcOut: String)] = [
("A001C001_260710", "10:00:00:00", "10:00:30:00"),
("A001C002_260710", "10:05:00:00", "10:05:45:00"),
("A001C003_260710", "10:12:00:00", "10:13:00:00"),
]
for take in takes {
await sim.setMetadata(ei: 800, wb: 5600, tint: 0, fps: 24, timecode: take.tcIn)
try await Task.sleep(for: .milliseconds(60))
await sim.setRecording(true, clipName: take.clip)
try await Task.sleep(for: .milliseconds(80))
await sim.setMetadata(ei: 800, wb: 5600, tint: 0, fps: 24, timecode: take.tcOut)
try await Task.sleep(for: .milliseconds(60))
await sim.setRecording(false)
try await Task.sleep(for: .milliseconds(80))
}
await driver.disconnect()
await logger.stop()
// Three shots logged with TCs + snapshots.
let shots = try store.listShots(dayID: day.id)
XCTAssertEqual(shots.count, 3)
XCTAssertEqual(shots.map(\.clipName), takes.map(\.clip))
for (shot, take) in zip(shots, takes) {
XCTAssertEqual(shot.timecodeIn, take.tcIn)
XCTAssertEqual(shot.timecodeOut, take.tcOut)
XCTAssertNotNil(shot.gradeSnapshot)
}
// All export formats produce sane output.
let ale = PostExport.writeALE(shots: shots, fps: 24)
XCTAssertEqual(ale.components(separatedBy: "A001C").count - 1, 3)
let csv = PostExport.writeCSV(shots: shots)
XCTAssertEqual(csv.split(separator: "\n").count, 4) // header + 3
let edl = PostExport.writeEDL(shots: shots, title: "DAY01", fps: 24)
XCTAssertTrue(edl.contains("003 A001C003_260710"))
XCTAssertEqual(edl.components(separatedBy: "* ASC_SOP").count - 1, 3)
let ccc = CDLExport.writeCCC(shots.compactMap { s in
guard let data = s.gradeSnapshot,
let stack = try? GradeSnapshot.decode(data),
case .cdl(let c)? = stack.nodes.first?.kind else { return nil }
return (s.clipName, c)
})
XCTAssertEqual(ccc.components(separatedBy: "<ColorCorrection id=").count - 1, 3)
let cube = try CubeExport.fromSnapshot(shots[0].gradeSnapshot!, latticeSize: 17, title: shots[0].clipName)
XCTAssertTrue(cube.contains("LUT_3D_SIZE 17"))
}
}