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 } }