diff --git a/Package.swift b/Package.swift index ea14e1e..dcd7bc7 100644 --- a/Package.swift +++ b/Package.swift @@ -20,6 +20,7 @@ let package = Package( "ForgeCamera", .product(name: "NIO", package: "swift-nio"), .product(name: "NIOHTTP1", package: "swift-nio"), + .product(name: "NIOWebSocket", package: "swift-nio"), ]), // Vendor drivers .target(name: "ForgeCameraARRI", dependencies: [ @@ -30,6 +31,8 @@ let package = Package( .target(name: "ForgeCameraRED", dependencies: [ "ForgeCamera", .product(name: "NIO", package: "swift-nio"), + .product(name: "NIOHTTP1", package: "swift-nio"), + .product(name: "NIOWebSocket", package: "swift-nio"), ]), .target(name: "ForgeCameraSony", dependencies: [ "ForgeCamera", diff --git a/Sources/ForgeCamera/GradeGang.swift b/Sources/ForgeCamera/GradeGang.swift index 25323ff..26a2109 100644 --- a/Sources/ForgeCamera/GradeGang.swift +++ b/Sources/ForgeCamera/GradeGang.swift @@ -82,19 +82,40 @@ public actor GradeGang { guard let driver = drivers[target] else { continue } let caps = driver.capabilities do { - // Prefer exact requested lattice; else largest supported size - // *below* it (downgrade only — baking a 17³-quality grade into a - // 99³ lattice wastes upload time and implies false precision). - guard let size = caps.lut3dSizes.contains(latticeSize) - ? latticeSize - : caps.lut3dSizes.filter({ $0 < latticeSize }).max() else { - throw CameraError.unsupportedLook("no compatible LUT size <= \(latticeSize) (camera accepts \(caps.lut3dSizes))") + if caps.lut3dSizes.isEmpty { + // CDL-only camera (RED over RCP2: no live LUT upload). + // Split CDL out; if the baked remainder is non-identity the + // camera cannot reproduce the grade — hard error so the + // operator knows monitor ≠ grade. + guard caps.supportsNativeCDL else { + throw CameraError.unsupportedLook("camera accepts neither LUTs nor CDL") + } + let look = Flattener.flatten( + stack: grade, + latticeSize: 17, // cheap lattice, only used for the identity check + splitLeadingCDL: true) + guard look.cdl != nil else { + throw CameraError.unsupportedLook("grade has no leading CDL; CDL-only camera cannot represent it") + } + if let lut = look.lut, !lut.isApproximatelyIdentity() { + throw CameraError.unsupportedLook("grade has nodes beyond CDL; CDL-only camera cannot represent them") + } + try await driver.push(look: FlattenedLook(cdl: look.cdl, lut: nil, latticeSize: 0)) + } else { + // Prefer exact requested lattice; else largest supported size + // *below* it (downgrade only — baking a 17³-quality grade into a + // 99³ lattice wastes upload time and implies false precision). + guard let size = caps.lut3dSizes.contains(latticeSize) + ? latticeSize + : caps.lut3dSizes.filter({ $0 < latticeSize }).max() else { + throw CameraError.unsupportedLook("no compatible LUT size <= \(latticeSize) (camera accepts \(caps.lut3dSizes))") + } + let look = Flattener.flatten( + stack: grade, + latticeSize: size, + splitLeadingCDL: caps.supportsNativeCDL) + try await driver.push(look: look) } - let look = Flattener.flatten( - stack: grade, - latticeSize: size, - splitLeadingCDL: caps.supportsNativeCDL) - try await driver.push(look: look) } catch { failures.append((slot: target, error: error)) } diff --git a/Sources/ForgeCameraRED/RedDriver.swift b/Sources/ForgeCameraRED/RedDriver.swift index e0adf76..e54a767 100644 --- a/Sources/ForgeCameraRED/RedDriver.swift +++ b/Sources/ForgeCameraRED/RedDriver.swift @@ -1,44 +1,59 @@ import Foundation +import NIO +import NIOHTTP1 +import NIOWebSocket import ForgeCamera import ForgeGrade import ForgeColor -#if canImport(Glibc) -import Glibc -#endif - -/// RED parameter names — single source of truth, correct against RED SDK -/// (RCP2) when hardware/SDK agreement lands. Sim mirrors these. +/// RCP2 param IDs (wire form: RCP_PARAM_ prefix stripped). +/// Verified against RED parameter docs via docs/research/red-rcp2-protocol.md. enum RedParams { - static let recordState = "RECORD_STATE" + static let recordState = "RECORD_STATE" // cur.val: 0 idle, 1 recording, set "2" = toggle static let clipName = "CLIP_NAME" static let iso = "ISO" - static let colorTemp = "COLOR_TEMP" + static let colorTemp = "COLOR_TEMPERATURE" static let tint = "TINT" - static let sensorFPS = "SENSOR_FPS" + static let sensorFPS = "SENSOR_FRAME_RATE" static let timecode = "TIMECODE" - static let ipp2CDL = "IPP2_CDL" + static let cdlEnable = "CDL_ENABLE" + static let cdlSlopeR = "CDL_SLOPE_RED" + static let cdlSlopeG = "CDL_SLOPE_GREEN" + static let cdlSlopeB = "CDL_SLOPE_BLUE" + static let cdlOffsetR = "CDL_OFFSET_RED" + static let cdlOffsetG = "CDL_OFFSET_GREEN" + static let cdlOffsetB = "CDL_OFFSET_BLUE" + static let cdlPowerR = "CDL_POWER_RED" + static let cdlPowerG = "CDL_POWER_GREEN" + static let cdlPowerB = "CDL_POWER_BLUE" + static let cdlSaturation = "CDL_SATURATION" } -/// DSMC3 driver: RCP2-style length-prefixed JSON over TCP. -/// Request/response serialized through the actor — one in-flight request. +/// DSMC3 driver — real RCP2: WebSocket ws://:9998/rcp, JSON text frames. +/// Handshake: rcp_config w/ client name. CDL pushed as per-channel rcp_set ops. +/// NO live LUT upload (RCP2 limitation) — capabilities.lut3dSizes empty; grade +/// pipeline keeps camera-native CDL only, remainder must stay identity. public actor RedDriver: CameraDriver { public nonisolated let capabilities = CameraCapabilities( vendor: .red, supportsNativeCDL: true, - lut3dSizes: [17, 33], + lut3dSizes: [], metadataFields: [.clipName, .exposureIndex, .whiteBalance, .tint, .fps, .timecode]) private let host: String private let port: Int private let pollInterval: Duration - private var socketFD: Int32 = -1 + private var group: MultiThreadedEventLoopGroup? + private var channel: Channel? private var pollTask: Task? + + /// Latest values per param, merged into CameraState on any change. + private var paramCache: [String: String] = [:] private var lastState = CameraState(connection: .disconnected) private var stateContinuations: [UUID: AsyncStream.Continuation] = [:] - public init(host: String, port: Int, pollInterval: Duration = .milliseconds(150)) { + public init(host: String, port: Int = 9998, pollInterval: Duration = .milliseconds(150)) { self.host = host self.port = port self.pollInterval = pollInterval @@ -69,111 +84,71 @@ public actor RedDriver: CameraDriver { } } - // MARK: Socket transport (blocking, actor-serialized, short timeouts) - - private func openSocket() throws { - let fd = socket(AF_INET, Int32(SOCK_STREAM.rawValue), 0) - guard fd >= 0 else { throw CameraError.connectionFailed("socket() failed") } - - var tv = timeval(tv_sec: 3, tv_usec: 0) - setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, socklen_t(MemoryLayout.size)) - setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, socklen_t(MemoryLayout.size)) - - var addr = sockaddr_in() - addr.sin_family = sa_family_t(AF_INET) - addr.sin_port = in_port_t(UInt16(port).bigEndian) - guard inet_pton(AF_INET, host, &addr.sin_addr) == 1 else { - close(fd) - throw CameraError.connectionFailed("bad host \(host)") - } - let rc = withUnsafePointer(to: &addr) { ptr in - ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { sa in - Glibc.connect(fd, sa, socklen_t(MemoryLayout.size)) - } - } - guard rc == 0 else { - close(fd) - throw CameraError.connectionFailed("connect to \(host):\(port) failed") - } - socketFD = fd - } - - private func closeSocket() { - if socketFD >= 0 { - close(socketFD) - socketFD = -1 - } - } - - private func sendAll(_ data: Data) throws { - var sent = 0 - try data.withUnsafeBytes { (buf: UnsafeRawBufferPointer) in - while sent < buf.count { - let n = write(socketFD, buf.baseAddress!.advanced(by: sent), buf.count - sent) - guard n > 0 else { throw CameraError.connectionFailed("write failed") } - sent += n - } - } - } - - private func recvExact(_ count: Int) throws -> Data { - var out = Data(capacity: count) - var remaining = count - var chunk = [UInt8](repeating: 0, count: 64 * 1024) - while remaining > 0 { - let n = read(socketFD, &chunk, min(remaining, chunk.count)) - guard n > 0 else { throw CameraError.connectionFailed("read failed/timeout") } - out.append(contentsOf: chunk[0.. [String: Any] { - guard socketFD >= 0 else { throw CameraError.connectionFailed("not connected") } - let payload = try JSONSerialization.data(withJSONObject: obj, options: [.sortedKeys]) - var frame = Data() - var lenBE = UInt32(payload.count).bigEndian - withUnsafeBytes(of: &lenBE) { frame.append(contentsOf: $0) } - frame.append(payload) - try sendAll(frame) - - let header = try recvExact(4) - // Byte-wise assembly — Data slices may be misaligned for a u32 load. - let length = header.reduce(UInt32(0)) { ($0 << 8) | UInt32($1) } - let body = try recvExact(Int(length)) - guard let reply = try JSONSerialization.jsonObject(with: body) as? [String: Any] else { - throw CameraError.connectionFailed("bad reply JSON") - } - return reply - } - - private func getParam(_ name: String) throws -> String { - let reply = try request(["type": "get", "param": name]) - guard reply["ok"] as? Bool == true, let value = reply["value"] as? String else { - throw CameraError.connectionFailed("get \(name) failed") - } - return value - } - - private func setParam(_ name: String, _ value: String) throws { - let reply = try request(["type": "set", "param": name, "value": value]) - guard reply["ok"] as? Bool == true else { - throw CameraError.pushFailed("set \(name) failed") - } - } - - // MARK: CameraDriver + // MARK: WebSocket transport public func connect() async throws { - closeSocket() - try openSocket() - let reply = try request(["type": "handshake"]) - guard reply["ok"] as? Bool == true else { - closeSocket() - throw CameraError.connectionFailed("handshake rejected") + await teardown() + let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) + self.group = group + + let upgradeResult: EventLoopPromise = group.next().makePromise(of: Channel.self) + let driver = self + + let upgrader = NIOWebSocketClientUpgrader( + requestKey: Data((0..<16).map { _ in UInt8.random(in: 0...255) }).base64EncodedString(), + upgradePipelineHandler: { channel, _ in + channel.pipeline.addHandler(RedClientWSHandler(driver: driver)).map { + upgradeResult.succeed(channel) + } + }) + + let bootstrap = ClientBootstrap(group: group) + .channelInitializer { channel in + let upgradeConfig = NIOHTTPClientUpgradeConfiguration( + upgraders: [upgrader], + completionHandler: { _ in + channel.pipeline.removeHandler(name: "sendInitialRequest", promise: nil) + }) + return channel.pipeline.addHTTPClientHandlers(withClientUpgrade: upgradeConfig).flatMap { + channel.pipeline.addHandler( + SendUpgradeRequestHandler(host: self.host, port: self.port, path: "/rcp"), + name: "sendInitialRequest") + } + } + + do { + let ch = try await bootstrap.connect(host: host, port: port).get() + _ = ch + // Wait for upgrade completion (or fail after timeout). + let upgraded: Channel = try await withThrowingTaskGroup(of: Channel.self) { grp in + grp.addTask { + try await upgradeResult.futureResult.get() + } + grp.addTask { + try await Task.sleep(for: .seconds(3)) + throw CameraError.connectionFailed("WS upgrade timeout") + } + let first = try await grp.next()! + grp.cancelAll() + return first + } + channel = upgraded + } catch { + upgradeResult.fail(error) + await teardown() + throw CameraError.connectionFailed("connect \(host):\(port): \(error)") } + + // RCP2 handshake. + try send([ + "type": "rcp_config", + "strings_decoded": 0, + "json_minified": 1, + "include_cacheable_flags": 0, + "encoding_type": "html", + "client": ["name": "Forge"], + ]) + emit(CameraState(connection: .connected)) startPolling() } @@ -181,65 +156,183 @@ public actor RedDriver: CameraDriver { public func disconnect() async { pollTask?.cancel() pollTask = nil - closeSocket() + await teardown() emit(CameraState(connection: .disconnected)) } - public func push(look: FlattenedLook) async throws { - if let lut = look.lut, !capabilities.lut3dSizes.contains(lut.size) { - throw CameraError.unsupportedLook("LUT size \(lut.size) not in \(capabilities.lut3dSizes)") + private func teardown() async { + if let ch = channel { + _ = try? await ch.close() } - if let cdl = look.cdl { - let obj: [String: Any] = [ - "slope": [cdl.slope.x, cdl.slope.y, cdl.slope.z], - "offset": [cdl.offset.x, cdl.offset.y, cdl.offset.z], - "power": [cdl.power.x, cdl.power.y, cdl.power.z], - "saturation": cdl.saturation, - ] - let json = String(data: try JSONSerialization.data(withJSONObject: obj, options: [.sortedKeys]), encoding: .utf8)! - try setParam(RedParams.ipp2CDL, json) + channel = nil + if let g = group { + try? await g.shutdownGracefully() } - if let lut = look.lut { - let reply = try request(["type": "lut", "size": lut.size, "table": lut.table]) - guard reply["ok"] as? Bool == true else { - throw CameraError.pushFailed("LUT upload rejected") - } + group = nil + } + + private func send(_ obj: [String: Any]) throws { + guard let ch = channel else { + throw CameraError.connectionFailed("not connected") + } + let data = try JSONSerialization.data(withJSONObject: obj, options: [.sortedKeys]) + var buf = ch.allocator.buffer(capacity: data.count) + buf.writeBytes(data) + let frame = WebSocketFrame(fin: true, opcode: .text, maskKey: randomMask(), data: buf) + ch.writeAndFlush(NIOAny(frame), promise: nil) + } + + private func randomMask() -> WebSocketMaskingKey { + WebSocketMaskingKey([UInt8.random(in: 0...255), .random(in: 0...255), .random(in: 0...255), .random(in: 0...255)])! + } + + // MARK: Inbound (called from WS handler) + + func handleInbound(_ data: Data) { + guard let obj = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + let id = obj["id"] as? String else { return } + // Value under cur.val (int/double/string). + guard let cur = obj["cur"] as? [String: Any], let val = cur["val"] else { return } + let stringValue: String + if let s = val as? String { + stringValue = s + } else if let i = val as? Int { + stringValue = String(i) + } else if let d = val as? Double { + stringValue = String(d) + } else { + return + } + paramCache[id] = stringValue + rebuildState() + } + + private func rebuildState() { + let newState = CameraState( + connection: .connected, + isRecording: paramCache[RedParams.recordState] == "1", + clipName: (paramCache[RedParams.clipName]?.isEmpty ?? true) ? nil : paramCache[RedParams.clipName], + metadata: CameraMetadata( + exposureIndex: paramCache[RedParams.iso].flatMap(Int.init), + whiteBalance: paramCache[RedParams.colorTemp].flatMap(Int.init), + tint: paramCache[RedParams.tint].flatMap(Int.init), + fps: paramCache[RedParams.sensorFPS].flatMap(Double.init), + timecode: paramCache[RedParams.timecode])) + if newState != lastState { + emit(newState) } } - // MARK: Polling + func handleChannelClosed() { + if lastState.connection == .connected { + emit(CameraState(connection: .disconnected)) + } + } + + // MARK: Polling (rcp_get; responses arrive via handleInbound) private func startPolling() { pollTask?.cancel() pollTask = Task { + let polled = [ + RedParams.recordState, RedParams.clipName, RedParams.iso, + RedParams.colorTemp, RedParams.tint, RedParams.sensorFPS, RedParams.timecode, + ] while !Task.isCancelled { - pollOnce() + for p in polled { + try? self.send(["type": "rcp_get", "id": p]) + } try? await Task.sleep(for: pollInterval) } } } - private func pollOnce() { - do { - let rec = try getParam(RedParams.recordState) == "1" - let clip = try getParam(RedParams.clipName) - let newState = CameraState( - connection: .connected, - isRecording: rec, - clipName: clip.isEmpty ? nil : clip, - metadata: CameraMetadata( - exposureIndex: Int(try getParam(RedParams.iso)), - whiteBalance: Int(try getParam(RedParams.colorTemp)), - tint: Int(try getParam(RedParams.tint)), - fps: Double(try getParam(RedParams.sensorFPS)), - timecode: try getParam(RedParams.timecode))) - if newState != lastState { - emit(newState) - } - } catch { - if lastState.connection == .connected { - emit(CameraState(connection: .disconnected)) - } + // MARK: Look push + + public func push(look: FlattenedLook) async throws { + // RCP2 cannot upload LUTs. Non-identity LUT = grade won't match on camera. + if let lut = look.lut, !lut.isApproximatelyIdentity() { + throw CameraError.unsupportedLook( + "RED RCP2 cannot receive live 3D LUTs — use CDL-only grade or pre-load LUT on camera") + } + guard let cdl = look.cdl else { + throw CameraError.unsupportedLook("nothing to push (no CDL)") + } + let sets: [(String, Any)] = [ + (RedParams.cdlSlopeR, Double(cdl.slope.x)), + (RedParams.cdlSlopeG, Double(cdl.slope.y)), + (RedParams.cdlSlopeB, Double(cdl.slope.z)), + (RedParams.cdlOffsetR, Double(cdl.offset.x)), + (RedParams.cdlOffsetG, Double(cdl.offset.y)), + (RedParams.cdlOffsetB, Double(cdl.offset.z)), + (RedParams.cdlPowerR, Double(cdl.power.x)), + (RedParams.cdlPowerG, Double(cdl.power.y)), + (RedParams.cdlPowerB, Double(cdl.power.z)), + (RedParams.cdlSaturation, Double(cdl.saturation)), + (RedParams.cdlEnable, 1), + ] + for (param, value) in sets { + try send(["type": "rcp_set", "id": param, "value": value]) } } + +} + +// MARK: - NIO handlers + +/// Sends the initial HTTP upgrade request when the channel becomes active. +final class SendUpgradeRequestHandler: ChannelInboundHandler, RemovableChannelHandler, @unchecked Sendable { + typealias InboundIn = HTTPClientResponsePart + typealias OutboundOut = HTTPClientRequestPart + + private let host: String + private let port: Int + private let path: String + + init(host: String, port: Int, path: String) { + self.host = host + self.port = port + self.path = path + } + + func channelActive(context: ChannelHandlerContext) { + var headers = HTTPHeaders() + headers.add(name: "Host", value: "\(host):\(port)") + headers.add(name: "Content-Length", value: "0") + let head = HTTPRequestHead(version: .http1_1, method: .GET, uri: path, headers: headers) + context.write(wrapOutboundOut(.head(head)), promise: nil) + context.writeAndFlush(wrapOutboundOut(.end(nil)), promise: nil) + context.fireChannelActive() + } +} + +/// Inbound WS frames -> driver actor. +final class RedClientWSHandler: ChannelInboundHandler, @unchecked Sendable { + typealias InboundIn = WebSocketFrame + + private let driver: RedDriver + + init(driver: RedDriver) { + self.driver = driver + } + + func channelRead(context: ChannelHandlerContext, data: NIOAny) { + let frame = unwrapInboundIn(data) + switch frame.opcode { + case .text: + var buf = frame.unmaskedData + guard let bytes = buf.readBytes(length: buf.readableBytes) else { return } + let payload = Data(bytes) + Task { await driver.handleInbound(payload) } + case .connectionClose: + context.close(promise: nil) + default: + break + } + } + + func channelInactive(context: ChannelHandlerContext) { + Task { await driver.handleChannelClosed() } + context.fireChannelInactive() + } } diff --git a/Sources/ForgeColor/Lut3D.swift b/Sources/ForgeColor/Lut3D.swift index 515b97f..77d7d08 100644 --- a/Sources/ForgeColor/Lut3D.swift +++ b/Sources/ForgeColor/Lut3D.swift @@ -42,6 +42,26 @@ public struct Lut3D: Sendable { return SIMD3(table[i], table[i + 1], table[i + 2]) } + /// True if lattice ≈ identity within tolerance (sampled grid corners). + /// Used to detect "nothing left after CDL split" flatten results. + public func isApproximatelyIdentity(tolerance: Float = 1.0 / 512) -> Bool { + let denom = Float(size - 1) + let step = max(1, size / 4) + for b in stride(from: 0, to: size, by: step) { + for g in stride(from: 0, to: size, by: step) { + for r in stride(from: 0, to: size, by: step) { + let idx = ((b * size + g) * size + r) * 3 + if abs(table[idx] - Float(r) / denom) > tolerance || + abs(table[idx + 1] - Float(g) / denom) > tolerance || + abs(table[idx + 2] - Float(b) / denom) > tolerance { + return false + } + } + } + } + return true + } + /// Trilinear interpolation; input clamped to [0,1]. public func sample(_ rgb: SIMD3) -> SIMD3 { let maxIdx = Float(size - 1) diff --git a/Sources/ForgeSim/RedSimulator.swift b/Sources/ForgeSim/RedSimulator.swift index 609df16..55305eb 100644 --- a/Sources/ForgeSim/RedSimulator.swift +++ b/Sources/ForgeSim/RedSimulator.swift @@ -1,85 +1,150 @@ import Foundation import NIO +import NIOHTTP1 +import NIOWebSocket -/// RCP2-style TCP simulator: 4-byte big-endian length prefix + JSON message. -/// Message types: handshake, get {param}, set {param,value}, lut {size,table}. -/// Replies mirror type with {ok:true} or value payloads. -/// NOTE: real RCP2 framing/params verified against RED SDK in hardware phase; -/// this encodes the documented shape (length-prefixed JSON over TCP). +/// RCP2-faithful RED camera simulator. +/// Wire format verified against real Komodo clients (docs/research/red-rcp2-protocol.md): +/// - WebSocket server at /rcp (real camera: port 9998) +/// - Text frames, one JSON object each +/// - Handshake: client sends {"type":"rcp_config",...}, sim replies ack +/// - {"type":"rcp_get","id":P} -> {"id":P,"cur":{"val":...}} +/// - {"type":"rcp_set","id":P,"value":V} -> stores, replies current value +/// - pushNotification() sends unsolicited {"id":P,"cur":{"val":...}} to all clients public actor RedSimulator { + + /// Param values are ints, doubles, or strings on the wire. + public enum ParamValue: Sendable, Equatable { + case int(Int) + case double(Double) + case string(String) + + var jsonValue: Any { + switch self { + case .int(let i): return i + case .double(let d): return d + case .string(let s): return s + } + } + + static func from(_ any: Any) -> ParamValue? { + if let i = any as? Int { return .int(i) } + if let d = any as? Double { return .double(d) } + if let s = any as? String { return .string(s) } + return nil + } + } + private var group: MultiThreadedEventLoopGroup? private var channel: Channel? public private(set) var boundPort: Int? - public private(set) var handshakeCount = 0 - public private(set) var params: [String: String] = [ - "RECORD_STATE": "0", - "CLIP_NAME": "", - "ISO": "800", - "COLOR_TEMP": "5600", - "TINT": "0", - "SENSOR_FPS": "24", - "TIMECODE": "00:00:00:00", - "MODEL": "V-RAPTOR-SIM", + public private(set) var params: [String: ParamValue] = [ + "RECORD_STATE": .int(0), + "CLIP_NAME": .string(""), + "ISO": .int(800), + "COLOR_TEMPERATURE": .int(5600), + "TINT": .int(0), + "SENSOR_FRAME_RATE": .string("24"), + "TIMECODE": .string("00:00:00:00"), + "CDL_ENABLE": .int(0), + "CAMERA_TYPE": .string("V-RAPTOR-SIM"), ] - public struct UploadedLut: Sendable { - public let size: Int - public let entryCount: Int - } - public private(set) var uploadedLuts: [UploadedLut] = [] + /// Raw rcp_config messages received (handshake verification; decode in consumers). + public private(set) var receivedConfigsData: [Data] = [] + + private var clients: [ObjectIdentifier: Channel] = [:] public init() {} - public func setParam(_ key: String, value: String) { + public func setParam(_ key: String, value: ParamValue) { params[key] = value } - /// Data-in/Data-out across the actor boundary (JSON dicts aren't Sendable). - func handleFrame(_ payload: Data) -> Data { - guard let obj = try? JSONSerialization.jsonObject(with: payload) as? [String: Any] else { - return (try? JSONSerialization.data(withJSONObject: ["ok": false, "error": "bad JSON"])) ?? Data() + /// Send unsolicited notification to all connected clients (camera-initiated update). + public func pushNotification(param: String, value: ParamValue) { + params[param] = value + let payload: [String: Any] = ["id": param, "type": "rcp_cur_int", "cur": ["val": value.jsonValue]] + guard let data = try? JSONSerialization.data(withJSONObject: payload, options: [.sortedKeys]) else { return } + for ch in clients.values { + Self.sendText(data, over: ch) } - let reply = handle(message: obj) - return (try? JSONSerialization.data(withJSONObject: reply, options: [.sortedKeys])) ?? Data() } - func handle(message: [String: Any]) -> [String: Any] { - switch message["type"] as? String { - case "handshake": - handshakeCount += 1 - return ["type": "handshake", "ok": true, "model": params["MODEL"] ?? ""] - case "get": - guard let param = message["param"] as? String, let value = params[param] else { - return ["type": "get", "ok": false, "error": "unknown param"] + // MARK: Frame handling + + func clientConnected(_ ch: Channel) { + clients[ObjectIdentifier(ch)] = ch + } + + func clientDisconnected(_ ch: Channel) { + clients.removeValue(forKey: ObjectIdentifier(ch)) + } + + func handleText(_ data: Data, from ch: Channel) { + guard let obj = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + let type = obj["type"] as? String else { return } + + switch type { + case "rcp_config": + receivedConfigsData.append(data) + reply(["type": "rcp_config_ack", "ok": true], to: ch) + case "rcp_get": + guard let id = obj["id"] as? String else { return } + if let value = params[id] { + reply(["id": id, "type": "rcp_cur_int", "cur": ["val": value.jsonValue]], to: ch) + } else { + reply(["id": id, "type": "rcp_error", "error": "unknown param"], to: ch) } - return ["type": "get", "ok": true, "param": param, "value": value] - case "set": - guard let param = message["param"] as? String, let value = message["value"] as? String else { - return ["type": "set", "ok": false, "error": "bad set"] - } - params[param] = value - return ["type": "set", "ok": true, "param": param] - case "lut": - guard let size = message["size"] as? Int, - let table = message["table"] as? [Any], - table.count == size * size * size * 3 else { - return ["type": "lut", "ok": false, "error": "bad lut"] - } - uploadedLuts.append(UploadedLut(size: size, entryCount: table.count)) - return ["type": "lut", "ok": true] + case "rcp_set": + guard let id = obj["id"] as? String, + let raw = obj["value"], let value = ParamValue.from(raw) else { return } + params[id] = value + reply(["id": id, "type": "rcp_cur_int", "cur": ["val": value.jsonValue]], to: ch) default: - return ["ok": false, "error": "unknown type"] + reply(["type": "rcp_error", "error": "unsupported type \(type)"], to: ch) } } + private func reply(_ obj: [String: Any], to ch: Channel) { + guard let data = try? JSONSerialization.data(withJSONObject: obj, options: [.sortedKeys]) else { return } + Self.sendText(data, over: ch) + } + + private static func sendText(_ data: Data, over ch: Channel) { + var buf = ch.allocator.buffer(capacity: data.count) + buf.writeBytes(data) + let frame = WebSocketFrame(fin: true, opcode: .text, data: buf) + ch.writeAndFlush(NIOAny(frame), promise: nil) + } + + // MARK: Lifecycle + public func start(port: Int) async throws { let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) self.group = group let sim = self + + let upgrader = NIOWebSocketServerUpgrader( + shouldUpgrade: { channel, head in + // Real camera serves /rcp. + guard head.uri == "/rcp" else { + return channel.eventLoop.makeSucceededFuture(nil) + } + return channel.eventLoop.makeSucceededFuture(HTTPHeaders()) + }, + upgradePipelineHandler: { channel, _ in + Task { await sim.clientConnected(channel) } + return channel.pipeline.addHandler(RedWSHandler(sim: sim)) + }) + let bootstrap = ServerBootstrap(group: group) .serverChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1) .childChannelInitializer { channel in - channel.pipeline.addHandler(RCPFrameHandler(sim: sim)) + let config = NIOHTTPServerUpgradeConfiguration( + upgraders: [upgrader], + completionHandler: { _ in }) + return channel.pipeline.configureHTTPServerPipeline(withServerUpgrade: config) } let channel = try await bootstrap.bind(host: "127.0.0.1", port: port).get() self.channel = channel @@ -87,6 +152,10 @@ public actor RedSimulator { } public func stop() async throws { + for ch in clients.values { + _ = try? await ch.close() + } + clients.removeAll() try await channel?.close() try await group?.shutdownGracefully() channel = nil @@ -95,41 +164,43 @@ public actor RedSimulator { } } -/// Length-prefixed JSON framing: u32 BE length + payload. -final class RCPFrameHandler: ChannelInboundHandler, @unchecked Sendable { - typealias InboundIn = ByteBuffer - typealias OutboundOut = ByteBuffer +/// Inbound WebSocket frames -> sim actor. +final class RedWSHandler: ChannelInboundHandler, @unchecked Sendable { + typealias InboundIn = WebSocketFrame + typealias OutboundOut = WebSocketFrame private let sim: RedSimulator - private var buffer = ByteBuffer() init(sim: RedSimulator) { self.sim = sim } func channelRead(context: ChannelHandlerContext, data: NIOAny) { - var incoming = unwrapInboundIn(data) - buffer.writeBuffer(&incoming) - - while true { - guard buffer.readableBytes >= 4, - let length = buffer.getInteger(at: buffer.readerIndex, as: UInt32.self), - buffer.readableBytes >= 4 + Int(length) else { break } - buffer.moveReaderIndex(forwardBy: 4) - guard let bytes = buffer.readBytes(length: Int(length)) else { break } + let frame = unwrapInboundIn(data) + switch frame.opcode { + case .text: + var buf = frame.unmaskedData + guard let bytes = buf.readBytes(length: buf.readableBytes) else { return } let payload = Data(bytes) - - let channel = context.channel - let loop = context.eventLoop - Task { - let replyData = await self.sim.handleFrame(payload) - loop.execute { - var out = channel.allocator.buffer(capacity: replyData.count + 4) - out.writeInteger(UInt32(replyData.count)) - out.writeBytes(replyData) - channel.writeAndFlush(out, promise: nil) - } - } + let ch = context.channel + Task { await sim.handleText(payload, from: ch) } + case .connectionClose: + let ch = context.channel + Task { await sim.clientDisconnected(ch) } + context.close(promise: nil) + case .ping: + var buf = frame.unmaskedData + let pong = WebSocketFrame(fin: true, opcode: .pong, data: buf) + context.writeAndFlush(wrapOutboundOut(pong), promise: nil) + _ = buf + default: + break } } + + func channelInactive(context: ChannelHandlerContext) { + let ch = context.channel + Task { await sim.clientDisconnected(ch) } + context.fireChannelInactive() + } } diff --git a/Tests/ForgeCameraREDTests/RedDriverTests.swift b/Tests/ForgeCameraREDTests/RedDriverTests.swift index b9687a4..fcd24da 100644 --- a/Tests/ForgeCameraREDTests/RedDriverTests.swift +++ b/Tests/ForgeCameraREDTests/RedDriverTests.swift @@ -6,6 +6,10 @@ import ForgeColor import ForgeSim @testable import ForgeCameraRED +/// Tests against verified RCP2 wire protocol: +/// WebSocket ws://:9998/rcp, rcp_config handshake, rcp_get/rcp_set, +/// responses {"id":..., "cur":{"val":...}}, per-channel CDL params, +/// no live LUT upload (docs/research/red-rcp2-protocol.md). final class RedDriverTests: XCTestCase { var sim: RedSimulator! @@ -25,16 +29,16 @@ final class RedDriverTests: XCTestCase { return RedDriver(host: "127.0.0.1", port: port, pollInterval: pollInterval) } - // Capabilities: RED = native CDL (IPP2), 17/33 LUTs. + // RED: native CDL yes, live LUT upload NO (RCP2 has no LUT-push op). func testCapabilities() async { let driver = await makeDriver() XCTAssertEqual(driver.capabilities.vendor, .red) XCTAssertTrue(driver.capabilities.supportsNativeCDL) - XCTAssertTrue(driver.capabilities.lut3dSizes.contains(33)) + XCTAssertTrue(driver.capabilities.lut3dSizes.isEmpty) } - // Connect performs RCP-style handshake, emits connected. - func testConnectHandshake() async throws { + // Connect: WS upgrade on /rcp + rcp_config handshake with client name. + func testConnectSendsRcpConfig() async throws { let driver = await makeDriver() var iterator = driver.state.makeAsyncIterator() try await driver.connect() @@ -46,8 +50,22 @@ final class RedDriverTests: XCTestCase { } } XCTAssertTrue(connected) - let handshakes = await sim.handshakeCount - XCTAssertEqual(handshakes, 1) + // Config delivery is async post-upgrade; allow brief settle. + var configsData: [Data] = [] + for _ in 0..<20 { + configsData = await sim.receivedConfigsData + if !configsData.isEmpty { break } + try await Task.sleep(for: .milliseconds(25)) + } + XCTAssertEqual(configsData.count, 1) + guard let first = configsData.first else { + await driver.disconnect() + return + } + let config = try JSONSerialization.jsonObject(with: first) as? [String: Any] + XCTAssertEqual(config?["type"] as? String, "rcp_config") + let client = config?["client"] as? [String: Any] + XCTAssertEqual(client?["name"] as? String, "Forge") await driver.disconnect() } @@ -59,25 +77,25 @@ final class RedDriverTests: XCTestCase { } catch {} } - // Rec state + clip name via param polling. + // Rec state: RECORD_STATE cur.val 1 -> recording, clip name via CLIP_NAME. func testRecStateEvents() async throws { let driver = await makeDriver() var iterator = driver.state.makeAsyncIterator() try await driver.connect() - await sim.setParam("RECORD_STATE", value: "1") - await sim.setParam("CLIP_NAME", value: "B001_C001_0710AB") + await sim.setParam("RECORD_STATE", value: .int(1)) + await sim.setParam("CLIP_NAME", value: .string("B001_C001_0710AB")) + // Params arrive one frame at a time — wait for a state with BOTH rec + clip. var sawRec = false - for _ in 0..<30 { - if let s = await iterator.next(), s.isRecording { - XCTAssertEqual(s.clipName, "B001_C001_0710AB") + for _ in 0..<60 { + if let s = await iterator.next(), s.isRecording, s.clipName == "B001_C001_0710AB" { sawRec = true break } } XCTAssertTrue(sawRec) - await sim.setParam("RECORD_STATE", value: "0") + await sim.setParam("RECORD_STATE", value: .int(0)) var sawStop = false for _ in 0..<30 { if let s = await iterator.next(), !s.isRecording { @@ -89,24 +107,25 @@ final class RedDriverTests: XCTestCase { await driver.disconnect() } - // Metadata params surface. + // Metadata via real param names: ISO, COLOR_TEMPERATURE, TINT, SENSOR_FRAME_RATE, TIMECODE. func testMetadata() async throws { - await sim.setParam("ISO", value: "1600") - await sim.setParam("COLOR_TEMP", value: "4500") - await sim.setParam("TINT", value: "3") - await sim.setParam("SENSOR_FPS", value: "23.98") - await sim.setParam("TIMECODE", value: "15:30:00:00") + await sim.setParam("ISO", value: .int(1600)) + await sim.setParam("COLOR_TEMPERATURE", value: .int(4500)) + await sim.setParam("TINT", value: .int(3)) + await sim.setParam("SENSOR_FRAME_RATE", value: .string("23.98")) + await sim.setParam("TIMECODE", value: .string("15:30:00:00")) let driver = await makeDriver() var iterator = driver.state.makeAsyncIterator() try await driver.connect() + // Wait for a state where ALL polled params have landed. var got = false - for _ in 0..<30 { - if let s = await iterator.next(), let md = s.metadata, md.exposureIndex == 1600 { - XCTAssertEqual(md.whiteBalance, 4500) - XCTAssertEqual(md.tint, 3) - XCTAssertEqual(md.timecode, "15:30:00:00") + for _ in 0..<60 { + if let s = await iterator.next(), let md = s.metadata, + md.exposureIndex == 1600, md.whiteBalance == 4500, + md.tint == 3, md.timecode == "15:30:00:00", + let fps = md.fps, abs(fps - 23.98) < 0.001 { got = true break } @@ -115,37 +134,99 @@ final class RedDriverTests: XCTestCase { await driver.disconnect() } - // Push: CDL to IPP2 slots + LUT upload. - func testPushLookWithNativeCDL() async throws { + // CDL push: 10 rcp_set ops (9 SOP channels + saturation) + CDL_ENABLE. + func testPushCDLSetsPerChannelParams() async throws { let driver = await makeDriver() try await driver.connect() - let cdl = CDL(slope: SIMD3(1.1, 1.0, 0.9), offset: SIMD3(0.02, 0, 0), power: .one, saturation: 1.3) - let look = FlattenedLook(cdl: cdl, lut: Lut3D.identity(size: 33), latticeSize: 33) - try await driver.push(look: look) + let cdl = CDL( + slope: SIMD3(1.1, 1.0, 0.9), + offset: SIMD3(0.02, 0.0, -0.01), + power: SIMD3(0.95, 1.0, 1.05), + saturation: 1.3) + try await driver.push(look: FlattenedLook(cdl: cdl, lut: nil, latticeSize: 33)) - let cdlValue = await sim.params["IPP2_CDL"] - XCTAssertNotNil(cdlValue) - let obj = try JSONSerialization.jsonObject(with: Data(cdlValue!.utf8)) as? [String: Any] - let slope = obj?["slope"] as? [Double] - XCTAssertEqual(slope?[0] ?? 0, 1.1, accuracy: 1e-5) - - let luts = await sim.uploadedLuts - XCTAssertEqual(luts.count, 1) - XCTAssertEqual(luts[0].size, 33) + // rcp_set frames are fire-and-forget; wait for sim to process all 11. + var params: [String: RedSimulator.ParamValue] = [:] + for _ in 0..<40 { + params = await sim.params + if params["CDL_SATURATION"] != nil, case .int(1)? = params["CDL_ENABLE"] { break } + try await Task.sleep(for: .milliseconds(25)) + } + func f(_ key: String) -> Float? { + if case .double(let d)? = params[key] { return Float(d) } + if case .int(let i)? = params[key] { return Float(i) } + return nil + } + XCTAssertEqual(f("CDL_SLOPE_RED") ?? 0, 1.1, accuracy: 1e-4) + XCTAssertEqual(f("CDL_SLOPE_GREEN") ?? 0, 1.0, accuracy: 1e-4) + XCTAssertEqual(f("CDL_SLOPE_BLUE") ?? 0, 0.9, accuracy: 1e-4) + XCTAssertEqual(f("CDL_OFFSET_RED") ?? 0, 0.02, accuracy: 1e-4) + XCTAssertEqual(f("CDL_OFFSET_BLUE") ?? 0, -0.01, accuracy: 1e-4) + XCTAssertEqual(f("CDL_POWER_RED") ?? 0, 0.95, accuracy: 1e-4) + XCTAssertEqual(f("CDL_POWER_BLUE") ?? 0, 1.05, accuracy: 1e-4) + XCTAssertEqual(f("CDL_SATURATION") ?? 0, 1.3, accuracy: 1e-4) + if case .int(let enabled)? = params["CDL_ENABLE"] { + XCTAssertEqual(enabled, 1) + } else { + XCTFail("CDL_ENABLE not set") + } await driver.disconnect() } - // Unsupported LUT size rejected. - func testUnsupportedLutRejected() async throws { + // LUT in look -> rejected: RCP2 cannot upload LUTs live. + func testPushWithLutRejected() async throws { let driver = await makeDriver() try await driver.connect() + let look = FlattenedLook(cdl: .identity, lut: Lut3D.build(size: 17) { $0 * 0.9 }, latticeSize: 17) do { - try await driver.push(look: FlattenedLook(cdl: nil, lut: Lut3D.identity(size: 65), latticeSize: 65)) + try await driver.push(look: look) XCTFail("expected unsupportedLook") } catch let e as CameraError { guard case .unsupportedLook = e else { return XCTFail("wrong error") } } await driver.disconnect() } + + // Identity LUT in look tolerated (flattener artifact), CDL still pushed. + func testPushWithIdentityLutAccepted() async throws { + let driver = await makeDriver() + try await driver.connect() + let cdl = CDL(slope: SIMD3(1.2, 1, 1), offset: .zero, power: .one, saturation: 1) + let look = FlattenedLook(cdl: cdl, lut: Lut3D.identity(size: 17), latticeSize: 17) + try await driver.push(look: look) + var pushed = false + for _ in 0..<40 { + let params = await sim.params + if case .double(let d)? = params["CDL_SLOPE_RED"], abs(Float(d) - 1.2) < 1e-4 { + pushed = true + break + } + try await Task.sleep(for: .milliseconds(25)) + } + XCTAssertTrue(pushed, "CDL not pushed") + await driver.disconnect() + } + + // Unsolicited camera notification (rec start w/o poll) surfaces in state stream. + func testUnsolicitedNotification() async throws { + let driver = await makeDriver(pollInterval: .seconds(10)) // poll effectively off + var iterator = driver.state.makeAsyncIterator() + try await driver.connect() + // Drain the connected event. + _ = await iterator.next() + + await sim.pushNotification(param: "RECORD_STATE", value: .int(1)) + await sim.pushNotification(param: "CLIP_NAME", value: .string("B001_C009")) + + var sawRec = false + for _ in 0..<10 { + if let s = await iterator.next(), s.isRecording { + sawRec = true + break + } + } + XCTAssertTrue(sawRec) + await driver.disconnect() + } } diff --git a/Tests/ForgeCameraTests/GangTests.swift b/Tests/ForgeCameraTests/GangTests.swift index 964507c..aeda416 100644 --- a/Tests/ForgeCameraTests/GangTests.swift +++ b/Tests/ForgeCameraTests/GangTests.swift @@ -109,6 +109,42 @@ final class GangTests: XCTestCase { } catch {} } + // CDL-only camera (RED/RCP2, lut3dSizes empty): CDL-only grade pushes as bare CDL. + func testCDLOnlyCameraPushesCDLOnlyGrade() async throws { + let red = RecordingDriver(nativeCDL: true, lutSizes: []) + let gang = GradeGang(latticeSize: 33) + // Grade = single CDL node — representable on a CDL-only camera. + let grade = GradeStack(nodes: [ + GradeNode(kind: .cdl(CDL(slope: SIMD3(1.3, 1, 1), offset: .zero, power: .one, saturation: 1.1))), + ]) + await gang.setGrade(grade, for: "R") + await gang.register(slot: "R", driver: red) + try await gang.pushGrade(from: "R") + + let pushes = await red.pushedLooks + XCTAssertEqual(pushes.count, 1) + XCTAssertNotNil(pushes[0].cdl) + XCTAssertNil(pushes[0].lut) + XCTAssertEqual(pushes[0].cdl?.slope.x ?? 0, 1.3, accuracy: 1e-5) + } + + // CDL-only camera + grade with non-CDL nodes: hard error (camera can't show it). + func testCDLOnlyCameraRejectsComplexGrade() async throws { + let red = RecordingDriver(nativeCDL: true, lutSizes: []) + let gang = GradeGang(latticeSize: 33) + await gang.setGrade(makeGrade(), for: "R") // CDL + saturation node + await gang.register(slot: "R", driver: red) + do { + try await gang.pushGrade(from: "R") + XCTFail("expected partial failure") + } catch let e as GangError { + guard case .partialFailure(let failures) = e else { return XCTFail("wrong error") } + XCTAssertEqual(failures.count, 1) + } + let pushes = await red.pushedLooks + XCTAssertTrue(pushes.isEmpty) + } + // Push failure on one camera doesn't block others (best-effort multi-push). func testPushPartialFailure() async throws { let good = RecordingDriver(nativeCDL: true)