diff --git a/Packages/MusubiKit/Sources/Intercom/HTTPServer.swift b/Packages/MusubiKit/Sources/Intercom/HTTPServer.swift new file mode 100644 index 0000000..a9740e5 --- /dev/null +++ b/Packages/MusubiKit/Sources/Intercom/HTTPServer.swift @@ -0,0 +1,155 @@ +import Foundation +import Network +import OSLog + +struct HTTPRequest: Equatable { + var method: String + var path: String + var headers: [String: String] + var body: [UInt8] +} + +struct HTTPResponse { + var status: Int + var contentType: String + var body: [UInt8] + + static func text(_ status: Int, _ text: String) -> HTTPResponse { + HTTPResponse(status: status, contentType: "text/plain; charset=utf-8", body: Array(text.utf8)) + } + + static func json(_ value: some Encodable) -> HTTPResponse { + HTTPResponse(status: 200, contentType: "application/json", body: Array((try? JSONEncoder().encode(value)) ?? Data())) + } + + var serialized: [UInt8] { + let reason = [200: "OK", 204: "No Content", 400: "Bad Request", 404: "Not Found", 405: "Method Not Allowed", + 413: "Payload Too Large", 503: "Service Unavailable"][status] ?? "Error" + let head = "HTTP/1.1 \(status) \(reason)\r\nContent-Type: \(contentType)\r\nContent-Length: \(body.count)\r\n" + + "Cache-Control: no-store\r\nConnection: close\r\n\r\n" + return Array(head.utf8) + body + } +} + +/// Incremental HTTP/1.1 request parser (no chunked bodies; the phone page only sends small JSON). +struct HTTPRequestParser { + enum ParseError: Error { case malformed, tooLarge } + + static let maxHeader = 8 * 1024 + static let maxBody = 64 * 1024 + + private var buffer: [UInt8] = [] + + /// Appends bytes; returns the request once it is complete. + mutating func append(_ bytes: [UInt8]) throws -> HTTPRequest? { + buffer += bytes + guard let headerEnd = Self.find([13, 10, 13, 10], in: buffer) else { + if buffer.count > Self.maxHeader { throw ParseError.tooLarge } + return nil + } + guard let head = String(bytes: buffer[0..= 0 else { throw ParseError.malformed } + guard length <= Self.maxBody else { throw ParseError.tooLarge } + let bodyStart = headerEnd + 4 + guard buffer.count >= bodyStart + length else { return nil } + let path = String(requestLine[1].split(separator: "?", maxSplits: 1).first ?? "/") + return HTTPRequest(method: String(requestLine[0]), path: path, headers: headers, + body: Array(buffer[bodyStart.. Int? { + guard haystack.count >= needle.count else { return nil } + for i in 0...(haystack.count - needle.count) where haystack[i] == needle[0] { + if Array(haystack[i.. HTTPResponse + + private let queue = DispatchQueue(label: "studio.zielke.musubi.intercom.http") + private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom") + private var listener: NWListener? + private let handler: Handler + + init(handler: @escaping Handler) { + self.handler = handler + } + + func start(interface: NWInterface, port: UInt16) async throws { + let parameters = NWParameters.tcp + parameters.requiredInterface = interface + parameters.allowLocalEndpointReuse = true + parameters.includePeerToPeer = false + let listener = try NWListener(using: parameters, on: NWEndpoint.Port(rawValue: port) ?? .any) + let once = OnceFlag() + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + listener.stateUpdateHandler = { [log] state in + switch state { + case .ready: + if once.claim() { continuation.resume() } + case .failed(let error): + log.error("HTTP failed: \(error.localizedDescription, privacy: .public)") + if once.claim() { continuation.resume(throwing: error) } + default: break + } + } + listener.newConnectionHandler = { [weak self] connection in self?.serve(connection) } + listener.start(queue: queue) + } + self.listener = listener + } + + func stop() { + listener?.cancel() + listener = nil + } + + private func serve(_ connection: NWConnection) { + connection.start(queue: queue) + read(connection, parser: ParserBox()) + } + + private func read(_ connection: NWConnection, parser: ParserBox) { + connection.receive(minimumIncompleteLength: 1, maximumLength: 16 * 1024) { [weak self] data, _, isComplete, error in + guard let self else { return } + do { + if let data, let request = try parser.value.append(Array(data)) { + // Signalling may block briefly (session setup); keep the HTTP queue free. + DispatchQueue.global(qos: .userInitiated).async { + self.respond(self.handler(request), on: connection) + } + return + } + } catch HTTPRequestParser.ParseError.tooLarge { + respond(.text(413, "Zu groß."), on: connection) + return + } catch { + respond(.text(400, "Ungültige Anfrage."), on: connection) + return + } + if isComplete || error != nil { connection.cancel() } else { read(connection, parser: parser) } + } + } + + private func respond(_ response: HTTPResponse, on connection: NWConnection) { + connection.send(content: Data(response.serialized), completion: .contentProcessed { _ in connection.cancel() }) + } +} + +/// Per-connection parser state; only touched on the HTTP queue. +private final class ParserBox: @unchecked Sendable { + var value = HTTPRequestParser() +} diff --git a/Packages/MusubiKit/Sources/Intercom/IntercomEngine.swift b/Packages/MusubiKit/Sources/Intercom/IntercomEngine.swift new file mode 100644 index 0000000..9fa4367 --- /dev/null +++ b/Packages/MusubiKit/Sources/Intercom/IntercomEngine.swift @@ -0,0 +1,218 @@ +import Foundation +import Network +import OSLog + +/// Media transport for all phones: one UDP port on the clinic interface, demultiplexed into sessions. +/// Everything runs on one serial queue; public methods may be called from any thread. +final class IntercomEngine: @unchecked Sendable { + struct SessionInfo: Sendable, Equatable { + let id: UUID + let name: String + let role: String + let connected: Bool + } + + var onSessionsChanged: (@Sendable ([SessionInfo]) -> Void)? + var onKeyFrameNeeded: (@Sendable () -> Void)? + + let queue = DispatchQueue(label: "studio.zielke.musubi.intercom.media", qos: .userInteractive) + private let certificate: DTLSCertificate + private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom") + + // queue only + private var listener: NWListener? + private var sessions: [UUID: PeerSession] = [:] + private var connections: [UUID: NWConnection] = [:] + private var timer: DispatchSourceTimer? + + init(certificate: DTLSCertificate) { + self.certificate = certificate + } + + /// Binds the media port; returns once listening (or throws). + func start(interface: NWInterface, port: UInt16) async throws { + let parameters = NWParameters.udp + parameters.requiredInterface = interface + parameters.allowLocalEndpointReuse = true + let listener = try NWListener(using: parameters, on: NWEndpoint.Port(rawValue: port) ?? .any) + let once = OnceFlag() + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + listener.stateUpdateHandler = { [log] state in + switch state { + case .ready: + if once.claim() { continuation.resume() } + case .failed(let error): + log.error("Media port failed: \(error.localizedDescription, privacy: .public)") + if once.claim() { continuation.resume(throwing: error) } + default: break + } + } + listener.newConnectionHandler = { [weak self] connection in self?.accept(connection) } + listener.start(queue: queue) + queue.async { self.listener = listener } + } + queue.async { self.startTimer() } + } + + func stop() { + queue.sync { + timer?.cancel() + timer = nil + listener?.cancel() + listener = nil + connections.values.forEach { $0.cancel() } + connections = [:] + sessions = [:] + } + onSessionsChanged?([]) + } + + /// Creates a session for a browser offer and returns its id and the SDP answer. + func createSession(offer: String, name: String, role: String, address: String, port: UInt16) throws -> (UUID, String)? { + let accepted = try PeerSession.accept(offerSDP: offer, name: name, role: role, certificate: certificate, + address: address, port: port) + guard let accepted else { return nil } + queue.sync { + sessions[accepted.session.id] = accepted.session + publish() + } + log.info("Session for \(name, privacy: .public) (\(role, privacy: .public)) created") + return (accepted.session.id, accepted.answer) + } + + func removeSession(_ id: UUID) { + queue.async { + self.sessions[id] = nil + self.connections.removeValue(forKey: id)?.cancel() + self.publish() + } + } + + /// Sends one Opus packet (20 ms) to every connected phone. + func sendAudio(opus: [UInt8], timestamp: UInt32) { + queue.async { + for (id, session) in self.sessions { + guard let packet = session.audioPacket(opus: opus, timestamp: timestamp) else { continue } + self.connections[id]?.send(content: Data(packet), completion: .idempotent) + } + } + } + + /// Sends one H.264 access unit to every connected phone. + func sendVideo(nalUnits: [[UInt8]], timestamp: UInt32) { + let payloads = H264Packetizer.payloads(for: nalUnits) + queue.async { + for (id, session) in self.sessions { + guard let connection = self.connections[id] else { continue } + for packet in session.videoPackets(payloads: payloads, timestamp: timestamp) { + connection.send(content: Data(packet), completion: .idempotent) + } + } + } + } + + // MARK: - queue + + private func accept(_ connection: NWConnection) { + connection.start(queue: queue) + receive(on: connection) + } + + private func receive(on connection: NWConnection) { + connection.receiveMessage { [weak self] data, _, _, error in + guard let self else { return } + if let data, !data.isEmpty { self.handle(Array(data), from: connection) } + if error == nil { self.receive(on: connection) } else { connection.cancel() } + } + } + + private func handle(_ packet: [UInt8], from connection: NWConnection) { + switch PeerSession.classify(packet) { + case .stun: + guard let message = STUN.parse(packet), + case .hostPort(let host, let port) = connection.endpoint else { return } + let address: [UInt8] + switch host { + case .ipv4(let v4): address = Array(v4.rawValue) + case .ipv6(let v6): + let raw = Array(v6.rawValue) + // IPv4-mapped (::ffff:a.b.c.d) on a dual-stack socket. + address = raw.prefix(12).elementsEqual([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xFF]) ? Array(raw.suffix(4)) : raw + default: return + } + let localUfrag = message.username?.split(separator: ":").first.map(String.init) + guard let session = sessions.values.first(where: { $0.localUfrag == localUfrag }), + let reply = session.answerSTUN(packet, message: message, address: address, port: port.rawValue) else { return } + if connections[session.id] !== connection { + connections[session.id] = connection + log.info("\(session.name, privacy: .public) reachable via \(String(describing: connection.endpoint), privacy: .public)") + } + connection.send(content: Data(reply), completion: .idempotent) + + case .dtls: + guard let (id, session) = session(for: connection) else { return } + let result = session.receiveDTLS(packet) + for reply in result.replies { connection.send(content: Data(reply), completion: .idempotent) } + if result.becameReady { + log.info("\(session.name, privacy: .public) connected (DTLS-SRTP)") + publish() + onKeyFrameNeeded?() + } else if case .failed(let reason) = session.dtlsState { + log.error("DTLS with \(session.name, privacy: .public) failed: \(reason, privacy: .public)") + sessions[id] = nil + connections[id] = nil + publish() + } + + case .rtcp: + if let (_, session) = session(for: connection), session.receiveRTCP(packet) { onKeyFrameNeeded?() } + + case .other: + break + } + } + + private func session(for connection: NWConnection) -> (UUID, PeerSession)? { + guard let id = connections.first(where: { $0.value === connection })?.key, let session = sessions[id] else { return nil } + return (id, session) + } + + private func startTimer() { + let timer = DispatchSource.makeTimerSource(queue: queue) + timer.schedule(deadline: .now() + 1, repeating: 1) + timer.setEventHandler { [weak self] in self?.tick() } + timer.resume() + self.timer = timer + } + + private func tick() { + let now = Date() + var changed = false + for (id, session) in sessions { + if session.isExpired(now: now) { + log.info("\(session.name, privacy: .public) gone (no consent)") + sessions[id] = nil + connections.removeValue(forKey: id)?.cancel() + changed = true + continue + } + if let connection = connections[id] { + for report in session.senderReports(now: now) { connection.send(content: Data(report), completion: .idempotent) } + } + } + if changed { publish() } + } + + private func publish() { + let infos = sessions.values + .sorted { $0.created < $1.created } + .map { SessionInfo(id: $0.id, name: $0.name, role: $0.role, connected: $0.isMediaReady) } + onSessionsChanged?(infos) + } +} + +final class OnceFlag: @unchecked Sendable { + private let lock = NSLock() + private var done = false + func claim() -> Bool { lock.withLock { defer { done = true }; return !done } } +} diff --git a/Packages/MusubiKit/Sources/Intercom/IntercomServer.swift b/Packages/MusubiKit/Sources/Intercom/IntercomServer.swift new file mode 100644 index 0000000..677e0f6 --- /dev/null +++ b/Packages/MusubiKit/Sources/Intercom/IntercomServer.swift @@ -0,0 +1,238 @@ +import CoreVideo +import Darwin +import Foundation +import Network +import Observation +import OSLog + +/// Intercom stage 1 (M4): phones on the clinic interface open a page, hear talkback and see PGM. +/// Web server and media port bind only to the chosen interface. +@MainActor @Observable +public final class IntercomServer { + public struct Participant: Identifiable, Equatable, Sendable { + public let id: UUID + public let name: String + public let role: String + public let connected: Bool + } + + public static let httpPort: UInt16 = 8080 + public static let mediaPort: UInt16 = 50_000 + + public private(set) var participants: [Participant] = [] + /// Address phones open (also encoded in the QR code). + public private(set) var url: URL? + public private(set) var error: String? + public private(set) var isRunning = false + + /// Thread-safe media input (PGM frames, talkback audio). + public nonisolated let input: IntercomInput + + private var engine: IntercomEngine? + private var http: HTTPServer? + private var retryTask: Task? + private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom") + + public init() { + input = IntercomInput() + } + + public func start(interfaceName: String) async { + stop() + guard let interface = await Self.interface(named: interfaceName), let address = Self.ipv4Address(of: interfaceName) else { + error = "Klinik-Schnittstelle \(interfaceName) hat keine Adresse. Neuer Versuch in 5 s." + log.error("Intercom: interface \(interfaceName, privacy: .public) unavailable") + scheduleRetry(interfaceName) + return + } + do { + let engine = IntercomEngine(certificate: try DTLSCertificate()) + engine.onSessionsChanged = { [weak self, input] infos in + input.hasListeners = infos.contains { $0.connected } + Task { @MainActor in + self?.participants = infos.map { Participant(id: $0.id, name: $0.name, role: $0.role, connected: $0.connected) } + } + } + engine.onKeyFrameNeeded = { [input] in input.requestKeyFrame() } + try await engine.start(interface: interface, port: Self.mediaPort) + + let mediaPort = Self.mediaPort + let http = HTTPServer { [weak engine] request in + guard let engine else { return .text(503, "Musubi startet neu.") } + return Self.route(request, engine: engine, address: address, mediaPort: mediaPort) + } + try await http.start(interface: interface, port: Self.httpPort) + + input.attach(engine) + self.engine = engine + self.http = http + url = URL(string: "http://\(address):\(Self.httpPort)/") + isRunning = true + error = nil + log.info("Intercom on \(interfaceName, privacy: .public) at \(address, privacy: .public)") + } catch { + stop() + self.error = "Intercom lässt sich auf \(interfaceName) nicht starten. Neuer Versuch in 5 s." + log.error("Intercom start failed: \(error.localizedDescription, privacy: .public)") + scheduleRetry(interfaceName) + } + } + + public func stop() { + retryTask?.cancel() + retryTask = nil + input.attach(nil) + http?.stop() + http = nil + engine?.stop() + engine = nil + participants = [] + url = nil + isRunning = false + } + + private func scheduleRetry(_ interfaceName: String) { + retryTask = Task { [weak self] in + try? await Task.sleep(for: .seconds(5)) + guard !Task.isCancelled else { return } + await self?.start(interfaceName: interfaceName) + } + } + + // MARK: - Signalling + + private struct OfferRequest: Decodable { let name: String; let role: String; let sdp: String } + private struct OfferResponse: Encodable { let id: UUID; let sdp: String } + private struct LeaveRequest: Decodable { let id: UUID } + + nonisolated private static func route(_ request: HTTPRequest, engine: IntercomEngine, address: String, mediaPort: UInt16) -> HTTPResponse { + switch (request.method, request.path) { + case ("GET", "/"), ("GET", "/index.html"): + return HTTPResponse(status: 200, contentType: "text/html; charset=utf-8", body: Array(PhonePage.html.utf8)) + case ("POST", "/api/offer"): + guard let offer = try? JSONDecoder().decode(OfferRequest.self, from: Data(request.body)) else { + return .text(400, "Angebot unlesbar.") + } + let name = String(offer.name.trimmingCharacters(in: .whitespacesAndNewlines).prefix(40)) + let role = String(offer.role.trimmingCharacters(in: .whitespacesAndNewlines).prefix(40)) + guard !name.isEmpty, !role.isEmpty else { return .text(400, "Name und Rolle fehlen.") } + do { + guard let (id, sdp) = try engine.createSession(offer: offer.sdp, name: name, role: role, + address: address, port: mediaPort) else { + return .text(400, "Dein Browser kann weder Opus noch H.264 empfangen.") + } + return .json(OfferResponse(id: id, sdp: sdp)) + } catch { + return .text(400, "Angebot unvollständig.") + } + case ("POST", "/api/leave"): + if let leave = try? JSONDecoder().decode(LeaveRequest.self, from: Data(request.body)) { engine.removeSession(leave.id) } + return HTTPResponse(status: 204, contentType: "text/plain", body: []) + case (_, "/"), (_, "/api/offer"), (_, "/api/leave"): + return .text(405, "Methode nicht erlaubt.") + default: + return .text(404, "Nicht gefunden.") + } + } + + // MARK: - Interfaces + + private static func interface(named name: String) async -> NWInterface? { + await withCheckedContinuation { continuation in + let monitor = NWPathMonitor() + let once = OnceFlag() + monitor.pathUpdateHandler = { path in + guard once.claim() else { return } + monitor.cancel() + continuation.resume(returning: path.availableInterfaces.first { $0.name == name }) + } + monitor.start(queue: .global(qos: .utility)) + } + } + + /// First IPv4 address of a BSD interface (e.g. "en0"). + nonisolated static func ipv4Address(of interfaceName: String) -> String? { + var list: UnsafeMutablePointer? + guard getifaddrs(&list) == 0, let first = list else { return nil } + defer { freeifaddrs(list) } + for entry in sequence(first: first, next: { $0.pointee.ifa_next }) { + let ifa = entry.pointee + guard String(cString: ifa.ifa_name) == interfaceName, let addr = ifa.ifa_addr, + addr.pointee.sa_family == UInt8(AF_INET) else { continue } + var host = [CChar](repeating: 0, count: Int(NI_MAXHOST)) + guard getnameinfo(addr, socklen_t(addr.pointee.sa_len), &host, socklen_t(host.count), nil, 0, NI_NUMERICHOST) == 0 else { continue } + return String(decoding: host.prefix { $0 != 0 }.map { UInt8(bitPattern: $0) }, as: UTF8.self) + } + return nil + } +} + +/// Media input for the intercom; safe to call from any thread. +public final class IntercomInput: @unchecked Sendable { + private let lock = NSLock() + private var engine: IntercomEngine? + private var listeners = false + private var talkback = TalkbackEncoder() + private let preview = PreviewEncoder() + private var resamplePosition: Double = 0 + private var resampleLast: Float = 0 + private var scratch: [Float] = [] + + init() { + preview.onAccessUnit = { [weak self] nalUnits, timestamp in + self?.currentEngine?.sendVideo(nalUnits: nalUnits, timestamp: timestamp) + } + } + + var hasListeners: Bool { + get { lock.withLock { listeners } } + set { lock.withLock { listeners = newValue } } + } + + private var currentEngine: IntercomEngine? { lock.withLock { engine } } + + func attach(_ engine: IntercomEngine?) { + lock.withLock { + self.engine = engine + listeners = false + } + } + + func requestKeyFrame() { preview.requestKeyFrame() } + + /// One PGM frame (any size, BGRA or UYVY). Dropped while no phone is connected. + public func pushVideo(_ pixelBuffer: CVPixelBuffer) { + guard hasListeners else { return } + preview.encode(pixelBuffer) + } + + /// Talkback audio, mono, any sample rate. Pass silence while the talk key is up so the stream keeps its timing. + /// Call from one thread only (not the audio render thread). + public func pushTalkback(_ samples: UnsafeBufferPointer, sampleRate: Double) { + guard let engine = currentEngine, hasListeners, let talkback else { return } + let packets: [([UInt8], UInt32)] + if sampleRate == TalkbackEncoder.sampleRate { + packets = talkback.append(samples) + } else { + resample(samples, from: sampleRate) + packets = scratch.withUnsafeBufferPointer { talkback.append($0) } + } + for (packet, timestamp) in packets { engine.sendAudio(opus: packet, timestamp: timestamp) } + } + + /// Linear interpolation to 48 kHz; good enough for speech. + private func resample(_ input: UnsafeBufferPointer, from rate: Double) { + scratch.removeAll(keepingCapacity: true) + let step = rate / TalkbackEncoder.sampleRate + var position = resamplePosition + while position < Double(input.count) { + let index = Int(position) + let fraction = Float(position - Double(index)) + let previous = index == 0 ? resampleLast : input[index - 1] + scratch.append(previous + (input[index] - previous) * fraction) + position += step + } + resamplePosition = position - Double(input.count) + resampleLast = input.last ?? resampleLast + } +} diff --git a/Packages/MusubiKit/Sources/Intercom/Media/PreviewEncoder.swift b/Packages/MusubiKit/Sources/Intercom/Media/PreviewEncoder.swift new file mode 100644 index 0000000..014a94b --- /dev/null +++ b/Packages/MusubiKit/Sources/Intercom/Media/PreviewEncoder.swift @@ -0,0 +1,160 @@ +import CoreMedia +import QuartzCore +import CoreVideo +import Foundation +import OSLog +import VideoToolbox + +/// Encodes the PGM preview for phones: scaled down, 25 fps, H.264 constrained baseline, no B-frames. +/// `encode` may be called from the NDI receive thread; work happens on an own queue. +final class PreviewEncoder: @unchecked Sendable { + /// Called with the NAL units of one access unit and its 90 kHz timestamp. + var onAccessUnit: (@Sendable ([[UInt8]], UInt32) -> Void)? + + let width: Int32 + let height: Int32 + let frameRate: Double + private let bitRate: Int + private let queue = DispatchQueue(label: "studio.zielke.musubi.intercom.video", qos: .userInitiated) + private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom") + + // queue only + private var session: VTCompressionSession? + private var transfer: VTPixelTransferSession? + private var pool: CVPixelBufferPool? + private var forceKeyFrame = true + private var lastFrameTime: Double = -1 + private var busy = false + private let start = CACurrentMediaTime() + + init(width: Int32 = 960, height: Int32 = 540, frameRate: Double = 25, bitRate: Int = 1_200_000) { + self.width = width + self.height = height + self.frameRate = frameRate + self.bitRate = bitRate + } + + deinit { + if let session { VTCompressionSessionInvalidate(session) } + if let transfer { VTPixelTransferSessionInvalidate(transfer) } + } + + func requestKeyFrame() { + queue.async { self.forceKeyFrame = true } + } + + /// Offers a frame; frames above the target rate, or while the encoder is still busy, are dropped. + func encode(_ pixelBuffer: CVPixelBuffer) { + let now = CACurrentMediaTime() - start + // CVPixelBuffer is immutable once handed over by the receiver. + nonisolated(unsafe) let pixelBuffer = pixelBuffer + queue.async { + guard !self.busy, now - self.lastFrameTime >= 1 / self.frameRate - 0.004 else { return } + self.lastFrameTime = now + self.encodeOnQueue(pixelBuffer, time: now) + } + } + + private func encodeOnQueue(_ source: CVPixelBuffer, time: Double) { + guard let session = session ?? makeSession(), let scaled = scale(source) else { return } + var options: CFDictionary? + if forceKeyFrame { + options = [kVTEncodeFrameOptionKey_ForceKeyFrame: true] as CFDictionary + forceKeyFrame = false + } + busy = true + let pts = CMTime(seconds: time, preferredTimescale: 90_000) + let status = VTCompressionSessionEncodeFrame(session, imageBuffer: scaled, presentationTimeStamp: pts, duration: .invalid, + frameProperties: options, infoFlagsOut: nil) { [weak self] status, _, sample in + guard let self else { return } + self.queue.async { self.busy = false } + guard status == noErr, let sample else { return } + self.emit(sample, timestamp: UInt32(truncatingIfNeeded: Int64(time * 90_000))) + } + if status != noErr { + busy = false + log.error("Preview encode failed: \(status)") + } + } + + private func makeSession() -> VTCompressionSession? { + var session: VTCompressionSession? + let status = VTCompressionSessionCreate(allocator: nil, width: width, height: height, codecType: kCMVideoCodecType_H264, + encoderSpecification: nil, imageBufferAttributes: nil, compressedDataAllocator: nil, + outputCallback: nil, refcon: nil, compressionSessionOut: &session) + guard status == noErr, let session else { + log.error("Preview encoder unavailable: \(status)") + return nil + } + let properties: [CFString: Any] = [ + kVTCompressionPropertyKey_RealTime: true, + kVTCompressionPropertyKey_ProfileLevel: kVTProfileLevel_H264_ConstrainedBaseline_AutoLevel, + kVTCompressionPropertyKey_AllowFrameReordering: false, + kVTCompressionPropertyKey_AverageBitRate: bitRate, + kVTCompressionPropertyKey_ExpectedFrameRate: frameRate, + kVTCompressionPropertyKey_MaxKeyFrameInterval: Int(frameRate * 2), + kVTCompressionPropertyKey_MaxKeyFrameIntervalDuration: 2, + ] + for (key, value) in properties { VTSessionSetProperty(session, key: key, value: value as CFTypeRef) } + VTCompressionSessionPrepareToEncodeFrames(session) + self.session = session + return session + } + + /// Scales any NDI frame (BGRA or UYVY, any size) into a pooled NV12 buffer of the preview size. + private func scale(_ source: CVPixelBuffer) -> CVPixelBuffer? { + if transfer == nil { + var newTransfer: VTPixelTransferSession? + VTPixelTransferSessionCreate(allocator: nil, pixelTransferSessionOut: &newTransfer) + if let newTransfer { + VTSessionSetProperty(newTransfer, key: kVTPixelTransferPropertyKey_ScalingMode, value: kVTScalingMode_Letterbox) + } + transfer = newTransfer + } + if pool == nil { + let attributes: [CFString: Any] = [ + kCVPixelBufferPixelFormatTypeKey: kCVPixelFormatType_420YpCbCr8BiPlanarVideoRange, + kCVPixelBufferWidthKey: width, + kCVPixelBufferHeightKey: height, + kCVPixelBufferIOSurfacePropertiesKey: [:] as CFDictionary, + ] + CVPixelBufferPoolCreate(nil, nil, attributes as CFDictionary, &pool) + } + guard let transfer, let pool else { return nil } + var target: CVPixelBuffer? + CVPixelBufferPoolCreatePixelBuffer(nil, pool, &target) + guard let target, VTPixelTransferSessionTransferImage(transfer, from: source, to: target) == noErr else { return nil } + return target + } + + private func emit(_ sample: CMSampleBuffer, timestamp: UInt32) { + guard let block = CMSampleBufferGetDataBuffer(sample) else { return } + var length = 0 + var pointer: UnsafeMutablePointer? + guard CMBlockBufferGetDataPointer(block, atOffset: 0, lengthAtOffsetOut: nil, totalLengthOut: &length, + dataPointerOut: &pointer) == noErr, let pointer else { return } + let avcc = [UInt8](UnsafeRawBufferPointer(start: pointer, count: length)) + var nalUnits = H264Packetizer.nalUnits(avcc: avcc) + + let attachments = CMSampleBufferGetSampleAttachmentsArray(sample, createIfNecessary: false) as? [[CFString: Any]] + let isKeyFrame = !(attachments?.first?[kCMSampleAttachmentKey_NotSync] as? Bool ?? false) + if isKeyFrame, let format = CMSampleBufferGetFormatDescription(sample) { + nalUnits = Self.parameterSets(format) + nalUnits + } + onAccessUnit?(nalUnits, timestamp) + } + + private static func parameterSets(_ format: CMFormatDescription) -> [[UInt8]] { + var count = 0 + CMVideoFormatDescriptionGetH264ParameterSetAtIndex(format, parameterSetIndex: 0, parameterSetPointerOut: nil, + parameterSetSizeOut: nil, parameterSetCountOut: &count, nalUnitHeaderLengthOut: nil) + return (0..? + var size = 0 + guard CMVideoFormatDescriptionGetH264ParameterSetAtIndex(format, parameterSetIndex: index, parameterSetPointerOut: &pointer, + parameterSetSizeOut: &size, parameterSetCountOut: nil, + nalUnitHeaderLengthOut: nil) == noErr, let pointer else { return nil } + return Array(UnsafeBufferPointer(start: pointer, count: size)) + } + } +} diff --git a/Packages/MusubiKit/Sources/Intercom/Media/TalkbackEncoder.swift b/Packages/MusubiKit/Sources/Intercom/Media/TalkbackEncoder.swift new file mode 100644 index 0000000..a1bc935 --- /dev/null +++ b/Packages/MusubiKit/Sources/Intercom/Media/TalkbackEncoder.swift @@ -0,0 +1,90 @@ +import AudioToolbox +import Foundation +import OSLog + +/// Encodes mono 48 kHz Float32 into 20 ms Opus packets with AudioToolbox. +/// Not thread-safe; the caller feeds it from one thread. +final class TalkbackEncoder { + static let sampleRate = 48_000.0 + static let frameSamples = 960 // 20 ms + + private var converter: AudioConverterRef? + private var pending: [Float] = [] + private var timestamp: UInt32 = .random(in: 0...UInt32.max) + private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom") + + init?(bitRate: UInt32 = 32_000) { + var input = AudioStreamBasicDescription( + mSampleRate: Self.sampleRate, mFormatID: kAudioFormatLinearPCM, + mFormatFlags: kAudioFormatFlagIsFloat | kAudioFormatFlagIsPacked, + mBytesPerPacket: 4, mFramesPerPacket: 1, mBytesPerFrame: 4, mChannelsPerFrame: 1, mBitsPerChannel: 32, mReserved: 0) + var output = AudioStreamBasicDescription( + mSampleRate: Self.sampleRate, mFormatID: kAudioFormatOpus, mFormatFlags: 0, + mBytesPerPacket: 0, mFramesPerPacket: UInt32(Self.frameSamples), mBytesPerFrame: 0, mChannelsPerFrame: 1, + mBitsPerChannel: 0, mReserved: 0) + let status = AudioConverterNew(&input, &output, &converter) + guard status == noErr, let converter else { + log.error("Opus encoder unavailable: \(status)") + return nil + } + var rate = bitRate + AudioConverterSetProperty(converter, kAudioConverterEncodeBitRate, UInt32(MemoryLayout.size), &rate) + } + + deinit { if let converter { AudioConverterDispose(converter) } } + + /// Appends mono samples at 48 kHz; returns (packet, RTP timestamp) for every full 20 ms. + func append(_ samples: UnsafeBufferPointer) -> [([UInt8], UInt32)] { + pending += samples + var out: [([UInt8], UInt32)] = [] + while pending.count >= Self.frameSamples { + let frame = Array(pending.prefix(Self.frameSamples)) + pending.removeFirst(Self.frameSamples) + if let packet = encode(frame) { out.append((packet, timestamp)) } + timestamp &+= UInt32(Self.frameSamples) + } + return out + } + + /// Input for one converter call; owns stable memory the converter may read after the callback returns. + private final class Feed { + let samples: UnsafeMutablePointer + let count: Int + var consumed = false + init(_ frame: [Float]) { + count = frame.count + samples = .allocate(capacity: count) + samples.initialize(from: frame, count: count) + } + deinit { samples.deallocate() } + } + + private func encode(_ frame: [Float]) -> [UInt8]? { + guard let converter else { return nil } + let feed = Feed(frame) + var packetSize: UInt32 = 1500 + var outBytes = [UInt8](repeating: 0, count: Int(packetSize)) + var packets: UInt32 = 1 + var description = AudioStreamPacketDescription() + let status = withExtendedLifetime(feed) { outBytes.withUnsafeMutableBytes { raw -> OSStatus in + var list = AudioBufferList(mNumberBuffers: 1, mBuffers: AudioBuffer(mNumberChannels: 1, mDataByteSize: packetSize, mData: raw.baseAddress)) + let status = AudioConverterFillComplexBuffer(converter, { _, ioPackets, ioData, _, userData in + let feed = Unmanaged.fromOpaque(userData!).takeUnretainedValue() + guard !feed.consumed else { + ioPackets.pointee = 0 + return 1 // no more data for this call + } + feed.consumed = true + ioPackets.pointee = UInt32(feed.count) + ioData.pointee.mBuffers.mData = UnsafeMutableRawPointer(feed.samples) + ioData.pointee.mBuffers.mDataByteSize = UInt32(feed.count * 4) + ioData.pointee.mBuffers.mNumberChannels = 1 + return noErr + }, Unmanaged.passUnretained(feed).toOpaque(), &packets, &list, &description) + packetSize = list.mBuffers.mDataByteSize + return status + } } + guard (status == noErr || status == 1), packets == 1, packetSize > 0 else { return nil } + return Array(outBytes.prefix(Int(packetSize))) + } +} diff --git a/Packages/MusubiKit/Sources/Intercom/PhonePage.swift b/Packages/MusubiKit/Sources/Intercom/PhonePage.swift new file mode 100644 index 0000000..90bf3a3 --- /dev/null +++ b/Packages/MusubiKit/Sources/Intercom/PhonePage.swift @@ -0,0 +1,151 @@ +/// The page phones open from the QR code. Colors follow design/tokens.json (dark). +enum PhonePage { + static let html = #""" + + + + + + + Musubi + + + +
+

Musubi

+

Talkback hören, Sendung sehen.

+ +
+ + + + + + +
+ + +
+ + + + """# +} diff --git a/Packages/MusubiKit/Sources/Intercom/RTC/PeerSession.swift b/Packages/MusubiKit/Sources/Intercom/RTC/PeerSession.swift new file mode 100644 index 0000000..2497d5b --- /dev/null +++ b/Packages/MusubiKit/Sources/Intercom/RTC/PeerSession.swift @@ -0,0 +1,150 @@ +import Foundation + +/// One phone: ICE-lite credentials, DTLS, SRTP and the two outgoing RTP streams. +/// Owned and called by `IntercomEngine` on its queue only. +final class PeerSession { + struct Offer { + let answer: String + let session: PeerSession + } + + enum Kind { case stun, dtls, rtcp, other } + + let id = UUID() + let name: String + let role: String + let localUfrag: String + let localPassword: String + let remoteUfrag: String + let created = Date() + + private(set) var lastConsent: Date? + private(set) var connectedSince: Date? + private var dtls: DTLSServer + private var srtp: SRTPContext? + private var audio: RTPStream? + private var video: RTPStream? + private var lastAudioTimestamp: UInt32 = 0 + private var lastVideoTimestamp: UInt32 = 0 + private var reportsDue = Date() + + var isMediaReady: Bool { srtp != nil } + var dtlsState: DTLSServer.State { dtls.state } + + private init(name: String, role: String, remoteUfrag: String, certificate: DTLSCertificate, remoteFingerprint: String) { + self.name = name + self.role = role + self.remoteUfrag = remoteUfrag + localUfrag = Self.token(8) + localPassword = Self.token(24) + dtls = DTLSServer(certificate: certificate, expectedFingerprint: remoteFingerprint) + } + + /// Accepts a browser offer and returns the answer; nil if the offer has nothing Musubi can send. + static func accept(offerSDP: String, name: String, role: String, certificate: DTLSCertificate, + address: String, port: UInt16) throws -> Offer? { + let offer = try SDPOffer(offerSDP) + let session = PeerSession(name: name, role: role, remoteUfrag: offer.iceUfrag, certificate: certificate, + remoteFingerprint: offer.fingerprint) + var tracks: [SDPAnswer.Track] = [] + for media in offer.media { + let ssrc = UInt32.random(in: 1...UInt32.max) + var payloadType: Int? + if media.port != 0 { + if media.kind == "audio", session.audio == nil, let pt = offer.opusPayloadType(in: media) { + payloadType = pt + session.audio = RTPStream(ssrc: ssrc, payloadType: UInt8(pt)) + } else if media.kind == "video", session.video == nil, let pt = offer.h264PayloadType(in: media) { + payloadType = pt + session.video = RTPStream(ssrc: ssrc, payloadType: UInt8(pt)) + } + } + tracks.append(.init(mid: media.mid, kind: media.kind, payloadType: payloadType, ssrc: ssrc)) + } + guard session.audio != nil || session.video != nil else { return nil } + let answer = SDPAnswer(iceUfrag: session.localUfrag, icePwd: session.localPassword, + fingerprint: certificate.fingerprint, address: address, port: port, tracks: tracks) + return Offer(answer: answer.text, session: session) + } + + static func classify(_ packet: [UInt8]) -> Kind { + if STUN.looksLikeSTUN(packet) { return .stun } + if DTLSServer.looksLikeDTLS(packet) { return .dtls } + if RTCPFeedback.isRTCP(packet) { return .rtcp } + return .other + } + + /// Answers a connectivity check if it carries this session's credentials. + func answerSTUN(_ packet: [UInt8], message: STUN.Message, address: [UInt8], port: UInt16) -> [UInt8]? { + guard message.type == STUN.bindingRequest, message.username == "\(localUfrag):\(remoteUfrag)", + STUN.verify(packet, message: message, key: Array(localPassword.utf8)) else { return nil } + lastConsent = Date() + return STUN.bindingResponse(transactionID: message.transactionID, address: address, port: port, + key: Array(localPassword.utf8)) + } + + /// Feeds DTLS; returns datagrams to send and whether media just became ready. + func receiveDTLS(_ packet: [UInt8]) -> (replies: [[UInt8]], becameReady: Bool) { + let replies = dtls.handle(packet) + guard srtp == nil, let keys = dtls.keyMaterial else { return (replies, false) } + // DTLS server = answerer: we send with the server keys. + srtp = SRTPContext(masterKey: keys.serverKey, masterSalt: keys.serverSalt) + connectedSince = Date() + return (replies, true) + } + + /// True if the browser asks for a key frame. + func receiveRTCP(_ packet: [UInt8]) -> Bool { + guard let srtpIn = inboundRTCP, let plain = srtpIn.unprotectRTCP(packet) else { return false } + return RTCPFeedback.requestsKeyFrame(plain) + } + + private lazy var inboundRTCP: SRTPContext? = { + guard let keys = dtls.keyMaterial else { return nil } + return SRTPContext(masterKey: keys.clientKey, masterSalt: keys.clientSalt) + }() + + func audioPacket(opus: [UInt8], timestamp: UInt32) -> [UInt8]? { + guard let srtp, audio != nil else { return nil } + lastAudioTimestamp = timestamp + return srtp.protectRTP(audio!.packet(timestamp: timestamp, marker: false, payload: opus)) + } + + func videoPackets(payloads: [[UInt8]], timestamp: UInt32) -> [[UInt8]] { + guard let srtp, video != nil else { return [] } + lastVideoTimestamp = timestamp + return payloads.enumerated().map { index, payload in + srtp.protectRTP(video!.packet(timestamp: timestamp, marker: index == payloads.count - 1, payload: payload)) + } + } + + /// Sender reports roughly once per second. + func senderReports(now: Date = Date()) -> [[UInt8]] { + guard let srtp, now >= reportsDue else { return [] } + reportsDue = now.addingTimeInterval(1) + var out: [[UInt8]] = [] + if let audio { + out.append(srtp.protectRTCP(RTCPFeedback.senderReport(ssrc: audio.ssrc, rtpTimestamp: lastAudioTimestamp, + packets: audio.packetCount, octets: audio.octetCount, now: now))) + } + if let video { + out.append(srtp.protectRTCP(RTCPFeedback.senderReport(ssrc: video.ssrc, rtpTimestamp: lastVideoTimestamp, + packets: video.packetCount, octets: video.octetCount, now: now))) + } + return out + } + + /// No consent for 30 s (RFC 7675), or never connected within 30 s, or DTLS ended. + func isExpired(now: Date = Date()) -> Bool { + switch dtls.state { + case .closed, .failed: return true + default: break + } + return now.timeIntervalSince(lastConsent ?? created) > 30 + } + + private static func token(_ length: Int) -> String { + let alphabet = Array("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") + return String((0..