import Foundation import NIO import NIOHTTP1 import NIOWebSocket /// 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 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"), ] /// 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: ParamValue) { params[key] = value } /// 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) } } // 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) } 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: 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 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 boundPort = channel.localAddress?.port } 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 group = nil boundPort = nil } } /// Inbound WebSocket frames -> sim actor. final class RedWSHandler: ChannelInboundHandler, @unchecked Sendable { typealias InboundIn = WebSocketFrame typealias OutboundOut = WebSocketFrame private let sim: RedSimulator init(sim: RedSimulator) { self.sim = sim } 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) 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() } }