diff --git a/Sources/ForgeCamera/ConnectionSupervisor.swift b/Sources/ForgeCamera/ConnectionSupervisor.swift index acda02a..c0f1096 100644 --- a/Sources/ForgeCamera/ConnectionSupervisor.swift +++ b/Sources/ForgeCamera/ConnectionSupervisor.swift @@ -39,7 +39,34 @@ public actor ConnectionSupervisor { public private(set) var health: ConnectionStatus = .disconnected - private var stateContinuations: [UUID: AsyncStream.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.Continuation] = [:] + + func add(_ id: UUID, _ c: AsyncStream.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] = [:] @@ -52,21 +79,13 @@ public actor ConnectionSupervisor { public nonisolated var states: AsyncStream { 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.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. diff --git a/Sources/ForgeCameraARRI/ArriDriver.swift b/Sources/ForgeCameraARRI/ArriDriver.swift index 12eb284..9e30403 100644 --- a/Sources/ForgeCameraARRI/ArriDriver.swift +++ b/Sources/ForgeCameraARRI/ArriDriver.swift @@ -32,6 +32,10 @@ public actor ArriDriver: CameraDriver { private var pollTask: Task? 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.Continuation] = [:] public init(host: String, port: Int, pollInterval: Duration = .milliseconds(150)) { diff --git a/Sources/forge-cli/Main.swift b/Sources/forge-cli/Main.swift index 4a67e04..b05ac16 100644 --- a/Sources/forge-cli/Main.swift +++ b/Sources/forge-cli/Main.swift @@ -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 { diff --git a/Tests/ForgeLoggerTests/EndToEndExportTests.swift b/Tests/ForgeLoggerTests/EndToEndExportTests.swift new file mode 100644 index 0000000..b6690f2 --- /dev/null +++ b/Tests/ForgeLoggerTests/EndToEndExportTests.swift @@ -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: "