import * as grpc from "@grpc/grpc-js"; import { MediaClient } from "@norskvideo/norsk-api/lib/media_grpc_pb"; import { Version, CurrentLoad, Log_Level, } from "@norskvideo/norsk-api/lib/shared/common_pb"; import { NorskStatusEvent, AmdMA35DLoad, SubscriptionChannelResponse, LicenseEvent, LicenseHandshakeRequestSchema, LicenseHandshakeResponse, Notification_Level } from "@norskvideo/norsk-api/lib/media_pb"; // Licensing V2: upper bound on the license handshake (responder + RPC), so a // stalled engine or responder can't hang Norsk.connect() after the connect // watchdog has been cleared by the hello. const LICENSE_HANDSHAKE_TIMEOUT_MS = 30000; async function withTimeout(promise: Promise, ms: number, message: string): Promise { let timer: ReturnType; const timeout = new Promise((_resolve, reject) => { timer = setTimeout(() => reject(new Error(message)), ms); }); try { return await Promise.race([promise, timeout]); } finally { clearTimeout(timer!); } } import { debuglog, errorlog, exhaustiveCheck, norskHost, norskPort, provideFull } from "./shared/utils"; import { MediaNodeState, PinToKey, MediaClient as SharedMediaClient } from "./media_nodes/common"; import { AudioSignalGeneratorNode, AudioSignalGeneratorSettings, BrowserInputNode, BrowserInputSettings, DeckLinkInputNode, DeckLinkInputSettings, DeltacastInputNode, DeltacastInputSettings, FileImageInputNode, FileImageInputSettings, LocalFileInputSettings, M3u8InputNode, M3u8MediaInputSettings, M3u8MultiVariantInputNode, M3u8MultiVariantInputSettings, M3u8WebVttInputNode, M3u8WebVttInputSettings, FileMp4InputNode, FileMp4InputSettings, NorskInput, RtmpServerInputNode, RtmpServerInputSettings, RtpInputNode, RtpInputSettings, SrtInputNode, SrtInputSettings, FileTsInputNode, UdpTsInputNode, UdpTsInputSettings, TcpTsInputNode, TcpTsInputSettings, FileWebVttInputNode, WhipInputNode, WhipInputSettings, FileTsInputSettings, VideoTestcardGeneratorSettings, VideoTestcardGeneratorNode, StreamSubtitlesInputSettings, StreamSubtitlesInputNode, FileWavInputSettings, FileWavInputNode, MxlInputSettings, MxlInputNode, NdiInputSettings, NdiInputNode, SrtRawInputSettings, SrtRawInputNode, TamsFlowInputSettings, TamsFlowInputNode, CmafIngestInputSettings, CmafIngestInputNode, MoqInputSettings, MoqInputNode, } from "./media_nodes/input"; import { Gain, NorskProcessor, WebRTCBrowserNode, WebRTCBrowserSettings, NorskDuplex, SipSettings, SipNode } from "./media_nodes/processor"; import { CmafAudioOutputNode, CmafMultiVariantOutputNode, CmafMultiVariantOutputSettings, CmafOutputSettings, CmafVideoOutputNode, CmafWebVttOutputNode, CmafWebVttOutputSettings, CmafTtmlOutputNode, CmafTtmlOutputSettings, FileMp4OutputNode, FileMp4OutputSettings, FileTsOutputNode, FileTsOutputSettings, FileWavOutputNode, FileWavOutputSettings, LocalAudioMonitorNode, LocalAudioMonitorSettings, FileWebVttOutputNode, FileWebVttOutputSettings, HlsTsAudioOutputNode, HlsTsAudioOutputSettings, HlsTsCombinedPushOutputNode, HlsTsCombinedPushOutputSettings, HlsTsMultiVariantOutputNode, HlsTsMultiVariantOutputSettings, HlsTsVideoOutputNode, HlsTsVideoOutputSettings, ImagePreviewOutputNode, ImagePreviewOutputSettings, MoqOutputNode, MoqOutputSettings, MxlOutputNode, MxlOutputSettings, NdiOutputNode, NdiOutputSettings, DeckLinkOutputNode, DeckLinkOutputSettings, NorskOutput, RtmpOutputNode, RtmpOutputSettings, SrtOutputNode, SrtOutputSettings, UdpTsOutputNode, UdpTsOutputSettings, TcpTsOutputNode, TcpTsOutputSettings, WebSocketOutputNode, WebSocketOutputSettings, WhepOutputNode, WhepOutputSettings, WhipOutputNode, WhipOutputSettings, TamsVideoFlowOutputSettings, TamsVideoFlowOutputNode, TamsAudioFlowOutputSettings, TamsAudioFlowOutputNode, } from "./media_nodes/output"; import { EncryptionSettings, StreamKey, StreamMetadata, Context, Wave } from "./media_nodes/types"; import { InspectSubtitlesNode, InspectSubtitlesSettings, MetricsNode, MetricsSettings, NorskInspect, QualityMonitoringNode, QualityMonitoringSettings, QualityReportingNode, QualityReportingSettings, StreamTimestampReportNode, StreamTimestampReportSettings } from "./media_nodes/inspect"; import { GlobalOptions, hardwareInfo, recommendST2110CoresFor, rotateMoqListenerCerts, setGlobalOptions, setupIrohEndpoint, getIrohEndpointInfo, setIrohAllowedPeers, getIrohListeners, IrohEndpointSettings, IrohEndpointEvent, suppressNoSubscriberWarning, unsuppressNoSubscriberWarning, NdiDiscovery, NdiDiscoverySettings, NorskSystem, ST2110Nic, ST2110NicSettings, ST2110NmosNode, ST2110NmosNodeSettings } from "./system"; import { Nic } from "./types"; import { MediaStorePlayerNode, MediaStorePlayerSettings, NorskMediaStore, MediaStoreRecorderNode, MediaStoreRecorderSettings, MediaStoreCutRequest, MediaStoreActiveCut, MediaStoreAssetSettings, MediaStoreAsset, MediaStoreSnapshotSettings, MediaStoreSnapshot, } from "./media_nodes/mediaStore"; // eslint-disable-next-line @typescript-eslint/no-require-imports import pj = require("../package.json"); import { ConnectivityState } from "@grpc/grpc-js/build/src/connectivity-state"; import { EmptySchema, Timestamp } from "@bufbuild/protobuf/wkt"; export * from "./types"; export * from "./system"; export * from "./media_nodes/types"; export * from "./media_nodes/input"; export * from "./media_nodes/output"; export * from "./media_nodes/inspect"; export * from "./media_nodes/processor"; export * from "./media_nodes/spectrum"; export * from "./media_nodes/common"; export * from "./media_nodes/mediaStore"; export * from "./media_nodes/embeddedAI"; export * from "./media_nodes/reasoningPlan"; export * from "./media_nodes/reasoningEvaluate"; export * from "./license-identity"; export { LiveSpec, LivePlanSettings, LivePlanSession } from "./media_nodes/livePlan"; export { LiveEvaluateNode, LiveEvaluateSettings, LiveEvaluateStatus, LiveEvaluateUsage, LiveEvaluateVideoSettings, LiveReasoningProvider } from "./media_nodes/liveEvaluate"; export * from "./media_nodes/st2110"; export { MqaAudioWeights, MqaConfig, MqaTsErrorWeights, MqaVideoWeights } from "./media_nodes/mqa"; export { CurrentLoad, } from "@norskvideo/norsk-api/lib/shared/common_pb"; export { AmdMA35DLoad } from "@norskvideo/norsk-api/lib/media_pb"; export { AudioCodec } from "@norskvideo/norsk-api/lib/media_pb"; export * from "./system"; export { Version } from "@norskvideo/norsk-api/lib/shared/common_pb"; /** @public */ export type Log = { level: | "emergency" | "alert" | "critical" | "error" | "warning" | "notice" | "info" | "debug"; timestamp: Date; message: string; metadata: object; }; /** @public */ export type Notification = { level: "error" |"warning" | "info"; /** Descriptive notification message text. */ text: string; /** Code/tag identifying the class of notification from this source */ code: string; /** When was the notification raised (if a recurring issue, this particular latest instance) */ timestamp: Date; /** The media node raising the notification */ mediaNodeId?: string; /** The type of the raising media node (description) */ mediaNodeClass?: string; /** * The type of processor within the media node giving rise to the notification (this can be used to understand the * cause of the notification with reference to the visualiser for that media node). */ nodeClass?: string; /** * For media nodes that contain multiple instances of the indicated notification origin, an indicator as to the particular source, * e.g. for an SRT listener with multiple TS decoders, this will indicate which source decoder was the origin */ instance?: string /** * Labels specific to the type of notification which may give additional information */ labels: Record; } /** * @public * Retraction of a previously-raised notification (matched by `code` for the * given `mediaNodeId`). Sent when the underlying condition clears. */ export type NotificationCleared = { /** Code/tag of the notification being retracted (matches Notification.code) */ code: string; /** The media node the notification was raised against */ mediaNodeId: string; } /** * @public * Top level Norsk configuration */ export interface NorskSettings { /** * Callback URL to listen on for gRPC session with Norsk Media * Defaults to $NORSK_HOST:$NORSK_PORT if the environment variables are set * where NORSK_HOST defaults to "127.0.0.1" and NORSK_PORT to "6790" * (so "127.0.0.1:6790" if neither variable is set) */ url?: string; onAttemptingToConnect?: () => void; onConnecting?: () => void; onReady?: () => void; onFailedToConnect?: () => void; /** Code to execute if the Norsk node is shutdown - by default it logs and nothing else */ onShutdown?: () => void; onCurrentLoad?: (load: CurrentLoad) => void; onAmdMA35DLoad?: (load: AmdMA35DLoad) => void; onHello?: (version: Version) => void; onLogEvent?: (log: Log) => void; /** * Manually handle license events, such as missing/invalid licenses and * sandbox timeout. (Logs messages to console by default.) */ onLicenseEvent?: (message: string, running: boolean, fullEvent: LicenseEvent) => void; onNotification?: (notification: Notification) => void; /** A previously-raised notification has been retracted (condition cleared) */ onNotificationCleared?: (cleared: NotificationCleared) => void; /** * A connection event on the shared iroh endpoint (peer connected / * rejected by the allow-list / disconnected). remoteNodeId is the * peer's TLS-verified public key (64-char hex). */ onIrohEndpointEvent?: (event: IrohEndpointEvent) => void; /** * Licensing V2: called if the license handshake fails (or no responder was * supplied). connect() still resolves; `norsk.licenseHandshake` carries the * same outcome. Media node creation will be refused by the engine. */ onLicenseChallengeFailed?: (message: string) => void; /** * Licensing V2: respond to the engine's license challenge. Engines holding * a per-product (V2) license send a challenge nonce in their hello and * refuse to create media nodes until a product (e.g. Studio) has answered * it with its certs and signatures. V1-licensed engines never issue a * challenge, so plain SDK applications can ignore this setting entirely. */ onLicenseChallenge?: (challenge: Uint8Array) => Promise | LicenseChallengeResponse; /* Allow for the tuning of the underlying GRPC setup */ grpc?: Partial } /** * @public * Licensing V2: a product's answer to the engine's license challenge. * Certs are Norsk-root-signed JSON envelopes; signatures are Ed25519 by the * respective cert holder's private key. */ export type LicenseChallengeResponse = { /** Root-signed studio-runtime cert (JSON envelope bytes) */ studioCert: Uint8Array; /** Signature over the challenge nonce by the studio-runtime key */ studioSig: Uint8Array; /** The declared product, e.g. "norsk-studio" — must appear in the license */ productName: string; /** Image ref of the product container, checked against the license entry */ imageRef: string; /** * Root-signed product-signer cert (JSON envelope bytes). Present only for * "signed" products (e.g. Probe) where the runtime relays a signed workflow; * omit for "designer" products (Studio), which carry no workflow signature. */ productCert?: Uint8Array; /** sha256 of the workflow document — present with productCert. */ yamlHash?: Uint8Array; /** Signature over yamlHash by the product-signer key — present with productCert. */ yamlSig?: Uint8Array; }; /** * @public * The entrypoint for all Norsk Media applications * * @example * ```ts * const norsk = new Norsk(); * ``` */ export class Norsk { /** @internal */ client: MediaClient; /** @internal */ innerClient: Partial; /** @internal */ nodes: MediaNodeState[]; /** @internal */ connectivityState: number; /** @internal */ statusStream?: grpc.ClientReadableStream; /** @internal */ publicWebPort?: number; /** * All of the nodes created for this Norsk instance */ public get Nodes(): MediaNodeState[] { return [...this.nodes]; } /** * Nodes that are in the process of being created */ public get PendingNodes(): MediaNodeState[] { return [...this.pendingNodeCreations]; } /** * Implements the {@link NorskInput} interface */ public input: NorskInput; /** * Implements the {@link NorskOutput} interface */ public output: NorskOutput; /** * Implements the {@link NorskDuplex} interface */ public duplex: NorskDuplex; /** * Implements the {@link NorskProcessor} interface */ public processor: NorskProcessor; /** * Implements the {@link NorskMediaStore} interface */ public mediaStore: NorskMediaStore; /** * Implements the {@link NorskInspect} interface */ public inspect: NorskInspect; /** * Implements the {@link NorskSystem} interface */ public system: NorskSystem; /** /* The settings used to create this Norsk instance */ public settings: NorskSettings; /** * Norsk Runtime version information */ public version: Version; /** * Norsk SDK version information */ public sdkVersion: string; /** * Norsk license information */ public license: { email: string, timeout?: number, expiry?: Timestamp, activeFeaturePacks: string[], expiredFeaturePacks: string[], }; /** * Licensing V2: the outcome of the license handshake, once connect() has * resolved. `undefined` on a V1-licensed engine (no handshake); `{accepted}` * on a V2 engine. When `accepted` is false, media node creation will be * refused — `message` carries the engine's reason. connect() resolves either * way so callers can read this and surface it rather than discovering the * failure only at the first node creation. */ public licenseHandshake?: { accepted: boolean, message: string }; /* @internal */ public resolveVersion: () => void; /* @internal */ public initialised: Promise; /* @internal */ closing: boolean = false; /* @internal */ timeout?: NodeJS.Timeout; /* @internal */ pendingNodeCreations: MediaNodeState[] = []; /** @internal */ registerNode(node: N): N { this.pendingNodeCreations = this.pendingNodeCreations.filter((n) => n != node); this.nodes.push(node); if (this.closing) { void node.close(); } return node; } /** @internal */ unregisterNode(node: N): void { this.nodes = this.nodes.filter(n => n != node); // This resolves the promise if somebody called \.close on the node node.finalise(); } public async close() { this.closing = true; let timeout = 100; debuglog("Closing Norsk (%d nodes active), (%d pending creation)", this.nodes.length, this.pendingNodeCreations.length); // Close all nodes in parallel, waiting for each to finish await Promise.all([ ...this.nodes.map(async (n) => { debuglog("Closing node", n.id); await n.close(); }) ]) await new Promise((resolve) => { const t = setInterval(() => { const stillOpen = this.nodes.length const pendingOpens = this.pendingNodeCreations.length; if (stillOpen > 0 || pendingOpens > 0) { if (timeout-- <= 0) { debuglog("Closing Norsk cleanly failed (%d nodes active) (%d pending), not waiting", stillOpen, pendingOpens); clearInterval(t); resolve(); } { debuglog("Closing Norsk (%d nodes active, %d nodes pending creation)", stillOpen, pendingOpens); } return; } else { clearInterval(t); resolve(); } }, 10.0); }) this.statusStream?.cancel(); this.innerClient.subscriptions?.cancel(); try { this.client.close(); } catch (e) { errorlog("Error closing norsk client", e) } try { this.client.getChannel().close(); } catch (e) { errorlog("Error closing norsk client channel", e) } return new Promise((r) => { const t = setInterval(() => { if (this.connectivityState != 2 || timeout-- <= 0) { clearInterval(t); r(); } }, 10.0) }); } /** @internal */ handleSubscriptionChannelEvent(_data: SubscriptionChannelResponse) { // There are none } /** @internal * Licensing V2: answer the engine's license challenge via the * onLicenseChallenge responder. Without a responder we log and return — * the engine will refuse media node creation with a clear error. */ private async completeLicenseHandshake(challenge: Uint8Array): Promise { if (!this.settings.onLicenseChallenge) { throw new Error( "Norsk holds a per-product (V2) license and requires a license handshake, " + "but no onLicenseChallenge responder was provided"); } // Bound both the (arbitrary) responder and the RPC so a stalled engine or // a responder that never settles can't hang connect() forever — the // hello handler has already disarmed the connect watchdog by this point. const response = await withTimeout( Promise.resolve(this.settings.onLicenseChallenge(challenge)), LICENSE_HANDSHAKE_TIMEOUT_MS, "onLicenseChallenge responder timed out"); const empty = new Uint8Array(0); const request = provideFull(LicenseHandshakeRequestSchema, { nonce: challenge, studioCert: response.studioCert, studioSig: response.studioSig, productCert: response.productCert ?? empty, yamlHash: response.yamlHash ?? empty, yamlSig: response.yamlSig ?? empty, productName: response.productName, imageRef: response.imageRef, }); const result = await new Promise((resolve, reject) => { this.client.completeLicenseHandshake( request, new grpc.Metadata(), { deadline: Date.now() + LICENSE_HANDSHAKE_TIMEOUT_MS }, (err, resp) => (err ? reject(err) : resolve(resp))); }); if (!result.accepted) { throw new Error(`Norsk rejected the license handshake: ${result.message}`); } debuglog("License handshake accepted for product %s", response.productName); } /** @internal */ handleStatusEvent(data: NorskStatusEvent) { const messageCase = data.message.case; switch (messageCase) { case undefined: break; case "hello": { debuglog( "Norsk status channel connected: %s", data.message.value.version ); this.publicWebPort = data.message.value.publicWebPort; if (data.message.value.version === undefined) { throw new Error("Norsk version is undefined"); } else { this.version = data.message.value.version; } this.license = { email: data.message.value.email, timeout: unwrapOptional(data.message.value.timeout), expiry: data.message.value.expiry, activeFeaturePacks: data.message.value.activeFeaturePacks, expiredFeaturePacks: data.message.value.expiredFeaturePacks, }; this.settings.onHello && data.message.value.version && this.settings.onHello(data.message.value.version); const runtimeVersion = this.version.label; const sdkVersion = this.sdkVersion.replace(/\+nightly$/, ""); if (runtimeVersion != sdkVersion) { const warning = `Norsk version mismatch: SDK=${sdkVersion}, runtime=${runtimeVersion}\nBEWARE POTENTIAL GRPC API CHANGES\nFunctionality may be partially or completely broken`; console.error(warning); errorlog(warning) } if (this.timeout) clearInterval(this.timeout); this.timeout = undefined; const challenge = data.message.value.licenseChallenge; if (challenge && challenge.length > 0) { // Licensing V2: the engine requires the license handshake before it // will create media nodes. Complete it before reporting ready. The // handshake is internally bounded (see completeLicenseHandshake), so // this always settles; on failure we record the reason on // licenseHandshake (node creation will be refused) and still resolve // connect() so the caller can read it rather than hang. this.completeLicenseHandshake(challenge) .then(() => { this.licenseHandshake = { accepted: true, message: "" }; }) .catch((e) => { const msg = e instanceof Error ? e.message : String(e); this.licenseHandshake = { accepted: false, message: msg }; console.error(`License handshake failed: ${msg}`); errorlog(`License handshake failed: ${msg}`); this.settings.onLicenseChallengeFailed?.(msg); }) .finally(() => this.resolveVersion()); } else { this.resolveVersion(); } break; } case "currentLoad": { this.settings.onCurrentLoad && this.settings.onCurrentLoad(data.message.value); break; } case "ma35dLoad": { this.settings.onAmdMA35DLoad && this.settings.onAmdMA35DLoad(data.message.value); break; } case "licenseEvent": { if (this.settings.onLicenseEvent) { this.settings.onLicenseEvent(data.message.value.message, data.message.value.running, data.message.value); } else { if (data.message.value.running) { console.warn(data.message.value.message); debuglog(data.message.value.message); } else { console.error(data.message.value.message); errorlog(data.message.value.message); } } break; } case "logEvent": { let level: | "emergency" | "alert" | "critical" | "error" | "warning" | "notice" | "info" | "debug"; switch (data.message.value.level) { case Log_Level.EMERGENCY: level = "emergency"; break; case Log_Level.ALERT: level = "alert"; break; case Log_Level.CRITICAL: level = "critical"; break; case Log_Level.ERROR: level = "error"; break; case Log_Level.WARNING: level = "warning"; break; case Log_Level.NOTICE: level = "notice"; break; case Log_Level.INFO: level = "info"; break; case Log_Level.DEBUG: level = "debug"; break; default: exhaustiveCheck(data.message.value.level); } const log: Log = { level: level, message: data.message.value.msg, timestamp: new Date(Number(data.message.value.timestamp) / 1000), metadata: JSON.parse(data.message.value.metadata) }; if (this.settings.onLogEvent) { this.settings.onLogEvent(log); } else { debuglog("Norsk log event: %o", log); } break; } case "notifications": { for (const note of data.message.value.notifications) { let level: Notification["level"] = "error"; switch (note.level) { case Notification_Level.NOTIFICATION_LEVEL_ERROR: level = "error"; break; case Notification_Level.NOTIFICATION_LEVEL_WARNING: level = "warning"; break; case Notification_Level.NOTIFICATION_LEVEL_INFO: level = "info"; break; } const context: Record = {}; for (const t of note.context) { context[t.key] = t.value; } const labels: Record = {}; for (const t of note.labels) { context[t.key] = t.value; } const n ={ level, text: note.text, code: note.code, timestamp: new Date(Number(note.utcTime)), mediaNodeId: context["mediaNodeId"], mediaNodeClass: context["mediaNodeClass"], nodeClass: context["nodeClass"], instance: context["instance"], labels }; this.settings.onNotification?.(n) debuglog(`[${n.level}] ${n.mediaNodeId}: ${note.text} (${note.code})`); } break; } case "notificationCleared": { const cleared = data.message.value; this.settings.onNotificationCleared?.({ code: cleared.code, mediaNodeId: cleared.mediaNodeId }); debuglog(`[cleared] ${cleared.mediaNodeId}: (${cleared.code})`); break; } case "irohEndpointEvent": { this.settings.onIrohEndpointEvent?.({ event: data.message.value.event as IrohEndpointEvent["event"], remoteNodeId: data.message.value.remoteNodeId, }); debuglog(`[iroh] ${data.message.value.event}: ${data.message.value.remoteNodeId}`); break; } default: exhaustiveCheck(messageCase); } } /** @internal */ connectivityStateWatcher() { const channel = this.client.getChannel(); const connectivityState = channel.getConnectivityState(!this.closing); switch (connectivityState) { case 0: { if (this.connectivityState == 1 || this.connectivityState == 2) { this.settings.onShutdown && this.settings.onShutdown(); } // Idle this.settings.onAttemptingToConnect && this.settings.onAttemptingToConnect(); break; } case 1: { // Connecting this.settings.onConnecting && this.settings.onConnecting(); break; } case 2: { // Ready this.settings.onReady && this.settings.onReady(); this.statusStream = this.client.createStatusChannel(provideFull(EmptySchema, {})); this.statusStream.on("data", this.handleStatusEvent.bind(this)); this.statusStream.on("error", (err) => { errorlog("Status stream error", err); return; }); this.innerClient.subscriptions = this.client.createSubscriptionChannel(); this.innerClient.subscriptions.on("data", this.handleSubscriptionChannelEvent.bind(this)); this.innerClient.subscriptions.on("error", (err) => { errorlog("Subscription channel error", err); return; }); break; } case 3: { // Transient failure this.settings.onFailedToConnect && this.settings.onFailedToConnect(); break; } case 4: { // Shutdown this.settings.onShutdown && this.settings.onShutdown(); break; } } debuglog("Channel connectivity state change: %d (%s)", connectivityState, ConnectivityState[connectivityState]); this.connectivityState = connectivityState; channel.watchConnectivityState(connectivityState, Infinity, () => { this.connectivityStateWatcher(); }); } /** @public */ public static async connect(settings?: NorskSettings) { debuglog("Norsk SDK: %s", pj.version); settings = settings ?? {}; if (!settings.onShutdown) { settings.onShutdown = () => { debuglog("Norsk has shutdown"); }; } const norsk = new Norsk(settings); await norsk.initialised; return norsk; } /** @internal */ private _client(): SharedMediaClient { return this.innerClient as SharedMediaClient; // with apologies } /** @internal */ preRegisterNodeCreation(node: MediaNodeState) { node.on('close', () => { const i = this.pendingNodeCreations.indexOf(node); if (i >= 0) { this.pendingNodeCreations.splice(i, 1); } }) this.pendingNodeCreations.push(node); } private onConnectionTimeout() { debuglog("Timeout on connect"); this.settings.onFailedToConnect && this.settings.onFailedToConnect(); } /** @internal */ constructor(norskSettings: NorskSettings) { this.connectivityState = 0; this.sdkVersion = pj.version; this.client = new MediaClient( norskSettings.url ? norskSettings.url : norskHost() + ":" + norskPort(), grpc.credentials.createInsecure(), norskSettings.grpc ?? {} ); this.innerClient = { media: this.client, // sod it norsk: this, }; this.settings = norskSettings; this.connectivityStateWatcher(); this.timeout = setTimeout(this.onConnectionTimeout.bind(this), 60000); this.initialised = new Promise((resolve, _reject) => { this.resolveVersion = resolve; }); this.nodes = []; const registerNode = this.registerNode.bind(this); const unregisterNode = this.unregisterNode.bind(this); this.input = { rtmpServer: async (settings: RtmpServerInputSettings) => RtmpServerInputNode.create(settings, this._client(), unregisterNode).then(registerNode), fileTs: async (settings: FileTsInputSettings) => FileTsInputNode.create(settings, this._client(), unregisterNode).then(registerNode), srt: async (settings: SrtInputSettings) => SrtInputNode.create(settings, this._client(), unregisterNode).then(registerNode), srtRaw: async (settings: SrtRawInputSettings) => SrtRawInputNode.create(settings, this._client(), unregisterNode).then(registerNode), whip: async (settings: WhipInputSettings) => WhipInputNode.create(settings, this._client(), unregisterNode).then(registerNode), m3u8Media: async (settings: M3u8MediaInputSettings) => M3u8InputNode.create(settings, this._client(), unregisterNode).then(registerNode), m3u8MultiVariant: async (settings: M3u8MultiVariantInputSettings) => M3u8MultiVariantInputNode.create(settings, this._client(), unregisterNode).then(registerNode), m3u8WebVtt: async (settings: M3u8WebVttInputSettings) => M3u8WebVttInputNode.create(settings, this._client(), unregisterNode).then(registerNode), tamsFlow: async (settings: TamsFlowInputSettings) => TamsFlowInputNode.create(settings, this._client(), unregisterNode).then(registerNode), udpTs: async (settings: UdpTsInputSettings) => UdpTsInputNode.create(settings, this._client(), unregisterNode).then(registerNode), tcpTs: async (settings: TcpTsInputSettings) => TcpTsInputNode.create(settings, this._client(), unregisterNode).then(registerNode), fileWebVtt: async (settings: LocalFileInputSettings) => FileWebVttInputNode.create(settings, this._client(), unregisterNode).then(registerNode), streamSubtitles: async (settings: StreamSubtitlesInputSettings) => StreamSubtitlesInputNode.create(settings, this._client(), unregisterNode).then(registerNode), fileImage: async (settings: FileImageInputSettings) => FileImageInputNode.create(settings, this._client(), unregisterNode).then(registerNode), fileMp4: async (settings: FileMp4InputSettings) => FileMp4InputNode.create(settings, this._client(), unregisterNode).then(registerNode), fileWav: async (settings: FileWavInputSettings) => FileWavInputNode.create(settings, this._client(), unregisterNode).then(registerNode), rtp: async (settings: RtpInputSettings) => RtpInputNode.create(settings, this._client(), unregisterNode).then(registerNode), audioSignal: async (settings: AudioSignalGeneratorSettings) => AudioSignalGeneratorNode.create(settings, this._client(), unregisterNode).then(registerNode), videoTestCard: async (settings: VideoTestcardGeneratorSettings) => VideoTestcardGeneratorNode.create(settings, this._client(), unregisterNode).then(registerNode), browser: async (settings: BrowserInputSettings) => BrowserInputNode.create(settings, this._client(), unregisterNode).then(registerNode), deckLink: async (settings: DeckLinkInputSettings) => DeckLinkInputNode.create(settings, this._client(), unregisterNode).then(registerNode), deltaCast: async (settings: DeltacastInputSettings) => DeltacastInputNode.create(settings, this._client(), unregisterNode).then(registerNode), ndi: async (settings: NdiInputSettings) => NdiInputNode.create(settings, this._client(), unregisterNode).then(registerNode), mxl: async (settings: MxlInputSettings) => MxlInputNode.create(settings, this._client(), unregisterNode).then(registerNode), cmafIngest: async (settings: CmafIngestInputSettings) => CmafIngestInputNode.create(settings, this._client(), unregisterNode).then(registerNode), moq: async (settings: MoqInputSettings) => MoqInputNode.create(settings, this._client(), unregisterNode).then(registerNode), }; this.output = { cmafVideo: async (settings: CmafOutputSettings) => CmafVideoOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), cmafAudio: async (settings: CmafOutputSettings) => CmafAudioOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), cmafWebVtt: async (settings: CmafWebVttOutputSettings) => CmafWebVttOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), cmafTtml: async (settings: CmafTtmlOutputSettings) => CmafTtmlOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), hlsTsVideo: async (settings: HlsTsVideoOutputSettings) => HlsTsVideoOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), udpTs: async (settings: UdpTsOutputSettings) => UdpTsOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), tcpTs: async (settings: TcpTsOutputSettings) => TcpTsOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), srt: async (settings: SrtOutputSettings) => SrtOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), hlsTsAudio: async (settings: HlsTsAudioOutputSettings) => HlsTsAudioOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), hlsTsCombinedPush: async (settings: HlsTsCombinedPushOutputSettings) => HlsTsCombinedPushOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), hlsTsMultiVariant: async (settings: HlsTsMultiVariantOutputSettings) => HlsTsMultiVariantOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), cmafMultiVariant: async (settings: CmafMultiVariantOutputSettings) => CmafMultiVariantOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), tamsVideoFlowPush: async (settings: TamsVideoFlowOutputSettings) => TamsVideoFlowOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), tamsAudioFlowPush: async (settings: TamsAudioFlowOutputSettings) => TamsAudioFlowOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), whip: async (settings: WhipOutputSettings) => WhipOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), whep: async (settings: WhepOutputSettings) => WhepOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), imagePreview: async (settings: ImagePreviewOutputSettings) => ImagePreviewOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), rtmp: async (settings: RtmpOutputSettings) => RtmpOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), fileTs: async (settings: FileTsOutputSettings) => FileTsOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), fileMp4: async (settings: FileMp4OutputSettings) => FileMp4OutputNode.create(settings, this._client(), unregisterNode).then(registerNode), fileWav: async (settings: FileWavOutputSettings) => FileWavOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), localAudioMonitor: async (settings: LocalAudioMonitorSettings) => LocalAudioMonitorNode.create(settings, this._client(), unregisterNode).then(registerNode), fileWebVtt: async (settings: FileWebVttOutputSettings) => FileWebVttOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), moq: async (settings: MoqOutputSettings) => MoqOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), ndi: async (settings: NdiOutputSettings) => NdiOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), mxl: async (settings: MxlOutputSettings) => MxlOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), deckLink: async (settings: DeckLinkOutputSettings) => DeckLinkOutputNode.create(settings, this._client(), unregisterNode).then(registerNode), webSocket: async (settings: WebSocketOutputSettings) => WebSocketOutputNode.create(settings, this._client(), unregisterNode).then(registerNode) }; this.mediaStore = { player: async (settings: MediaStorePlayerSettings) => MediaStorePlayerNode.create(settings, this._client(), unregisterNode).then(registerNode), recorder: async (settings: MediaStoreRecorderSettings) => MediaStoreRecorderNode.create(settings, this._client(), unregisterNode).then(registerNode), makeCut: (cutRequest: MediaStoreCutRequest) => MediaStoreActiveCut.create(cutRequest, this._client()), asyncLoadAsset: async (settings: MediaStoreAssetSettings) => MediaStoreAsset.create(settings, this._client()), snapshot: async (settings: MediaStoreSnapshotSettings) => MediaStoreSnapshot.create(settings, this._client()) } this.inspect = { streamTimestampReport: async (settings: StreamTimestampReportSettings) => StreamTimestampReportNode.create(settings, this._client(), unregisterNode).then(registerNode), subtitles: async (settings: InspectSubtitlesSettings) => InspectSubtitlesNode.create(settings, this._client(), unregisterNode).then(registerNode), streamMetrics: async (settings: MetricsSettings) => MetricsNode.create(settings, this._client(), unregisterNode).then(registerNode), qualityMonitoring: async (settings: QualityMonitoringSettings) => QualityMonitoringNode.create(settings, this._client(), unregisterNode).then(registerNode), qualityReporting: async (settings: QualityReportingSettings) => QualityReportingNode.create(settings, this._client(), unregisterNode).then(registerNode), }; this.processor = new NorskProcessor(this, this._client()); this.system = { hardwareInfo: async () => hardwareInfo(this.client), setGlobalOptions: async (options: GlobalOptions) => setGlobalOptions(this.client, options), rotateMoqListenerCerts: async (type: 'wt' | 'quic', port: number, certFile: string, keyFile: string) => rotateMoqListenerCerts(this.client, type, port, certFile, keyFile), iroh: { setup: async (settings: IrohEndpointSettings) => setupIrohEndpoint(this.client, settings), info: async () => getIrohEndpointInfo(this.client), setAllowedPeers: async (peers: string[] | 'any') => setIrohAllowedPeers(this.client, peers), listeners: async () => getIrohListeners(this.client), }, suppressNoSubscriberWarning: async (mediaNodeId: string) => suppressNoSubscriberWarning(this.client, mediaNodeId), unsuppressNoSubscriberWarning: async (mediaNodeId: string) => unsuppressNoSubscriberWarning(this.client, mediaNodeId), ndiDiscovery: async (settings: NdiDiscoverySettings) => NdiDiscovery.create(settings, this.client), st2110: { node: async (settings: ST2110NmosNodeSettings) => ST2110NmosNode.create(settings, this._client(), unregisterNode), nic: async (settings: ST2110NicSettings) => ST2110Nic.create(settings, this.client), // Same as `nic()`, but resolves once the server-side gen_server // has passed preflight — without waiting for MTL_Init / // MTL_Start. Useful when orchestrating a workflow that wants // downstream component setup to overlap with NIC bring-up. nicAfterPreflight: async (settings: ST2110NicSettings) => ST2110Nic.createAfterPreflight(settings, this.client), recommendCoresFor: async (settings: { nic: Nic; count: number }) => recommendST2110CoresFor(this.client, settings), } }; this.duplex = { webRtcBrowser: async (settings: WebRTCBrowserSettings) => WebRTCBrowserNode.create(settings, this._client(), unregisterNode).then(registerNode), sip: async (settings: SipSettings) => SipNode.create(settings, this._client(), unregisterNode).then(registerNode) }; } } /** * @public * Filters a context to only the audio streams within it * @param streams - The media context from which to return the streams * @returns The audio streams in the media context */ export function audioStreams( streams: readonly StreamMetadata[] ): StreamMetadata[] { return streams.filter((stream) => stream.message.case === "audio"); } /** * @public * Filters a context to only the video streams within it * @param streams - The media context from which to return the streams * @returns The video streams in the media context */ export function videoStreams( streams: readonly StreamMetadata[] ): StreamMetadata[] { return streams.filter((stream) => stream.message.case === "video"); } /** * @public * Filters a context to only the subtitle streams within it * @param streams - The media context from which to return the streams * @returns The subtitle streams in the media context */ export function subtitleStreams( streams: readonly StreamMetadata[] ): StreamMetadata[] { return streams.filter((stream) => stream.message.case === "subtitle"); } /** * @public * Filters a context to only the playlist streams within it * @param streams - The media context from which to return the streams * @returns The playlist streams in the media context */ export function playlistStreams(streams: readonly StreamMetadata[]): StreamMetadata[] { return streams.filter((stream) => stream.message.case === "playlist"); } /** * @public * Filters a context to only the ancillary streams within it * @param streams - The media context from which to return the streams * @returns The ancillary streams in the media context */ export function ancillaryStreams( streams: readonly StreamMetadata[] ): StreamMetadata[] { return streams.filter((stream) => stream.message.case === "ancillary"); } /** * @public * Returns the stream keys for audio streams in a media context * @param streams - The media context from which to return the stream keys * @returns The audio stream keys in the media context */ export function audioStreamKeys( streams: readonly StreamMetadata[] ): StreamKey[] { return removeUndefined( audioStreams(streams).map((stream) => stream.streamKey) ); } /** * @public * Returns the stream keys for video streams in a media context * @param streams - The media context from which to return the stream keys * @returns The video stream keys in the media context */ export function videoStreamKeys( streams: readonly StreamMetadata[] ): StreamKey[] { return removeUndefined( videoStreams(streams).map((stream) => stream.streamKey) ); } /** * @public * Returns the stream keys for subtitle streams in a media context * @param streams - The media context from which to return the stream keys * @returns The subtitle stream keys in the media context */ export function subtitleStreamKeys( streams: readonly StreamMetadata[] ): StreamKey[] { return removeUndefined( subtitleStreams(streams).map((stream) => stream.streamKey) ); } /** * @public * Returns the stream keys for playlist streams in a media context * @param streams - The media context from which to return the stream keys * @returns The playlist stream keys in the media context */ export function playlistStreamKeys(streams: readonly StreamMetadata[]): StreamKey[] { return removeUndefined( playlistStreams(streams).map((stream) => stream.streamKey) ); } /** * @public * Returns the stream keys for ancillary streams in a media context * @param streams - The media context from which to return the stream keys * @returns The ancillary stream keys in the media context */ export function ancillaryStreamKeys( streams: readonly StreamMetadata[] ): StreamKey[] { return removeUndefined( ancillaryStreams(streams).map((stream) => stream.streamKey) ); } function removeUndefined(values: readonly (T | undefined)[]): T[] { // This isn't really typechecked as you would like :( return values.filter((v): v is T => v !== undefined); } /** @public */ export function newSilentMatrix(rows: number, cols: number): Gain[][] { return new Array(rows).fill(0).map(() => new Array(cols).fill(null)); } /** * @public * * Provided for compatibilty, this is just e.g. `{ type: "sine", freq: 444 }` */ export function mkSine(freq: number): Wave { return { type: "sine", freq } } /** * @public * Select all the streams from the input * @param streams - The streams from the inbound Context * @returns Array of selected StreamKeys */ export function selectAll(streams: readonly StreamMetadata[]): StreamKey[] { return streams.map(s => s.streamKey); } /** * @public * Select all the audio and video streams from the input * @param streams - The streams from the inbound Context * @returns Array of selected StreamKeys */ export function selectAV(streams: readonly StreamMetadata[]): StreamKey[] { const audio = audioStreamKeys(streams); const video = videoStreamKeys(streams); return audio.concat(video); } /** * @public * Select all the subtitle streams from the input * @param streams - The streams from the inbound Context * @returns Array of selected StreamKeys */ export function selectSubtitles(streams: readonly StreamMetadata[]): StreamKey[] { return subtitleStreamKeys(streams); } /** * @public * Select all the audio streams from the input * @param streams - The streams from the inbound Context * @returns Array of selected StreamKeys */ export function selectAudio(streams: readonly StreamMetadata[]): StreamKey[] { return audioStreamKeys(streams); } /** * @public * Select all the video streams from the input * @param streams - The streams from the inbound Context * @returns Array of selected StreamKeys */ export function selectVideo(streams: readonly StreamMetadata[]): StreamKey[] { return videoStreamKeys(streams); } /** * @public * Select all the ancillary data streams from the input * @param streams - The streams from the inbound Context * @returns Array of selected StreamKeys */ export function selectAncillary(streams: readonly StreamMetadata[]): StreamKey[] { return ancillaryStreamKeys(streams); } /** * @public * Create a selector selecting all the video streams from the input with the specified rendition name * @param renditionName - The streams from the inbound Context * @returns Array of selected StreamKeys */ export function selectVideoRendition(renditionName: string) { return (streams: readonly StreamMetadata[]): StreamKey[] => videoStreamKeys(streams).filter((s) => s.renditionName == renditionName); } /** * @public * Select the audio stream with the given renditionName — e.g. to subscribe to * one named output of an {@link NorskTransform.audioGraph} node. */ export function selectAudioRendition(renditionName: string) { return (streams: readonly StreamMetadata[]): StreamKey[] => audioStreamKeys(streams).filter((s) => s.renditionName == renditionName); } /** @public */ export function selectExactKey(key: StreamKey) { return (streams: readonly StreamMetadata[]): StreamKey[] => streams.map((s) => s.streamKey).filter((s) => s && streamKeysAreEqual(s, key)) as StreamKey[]; } /** @public */ export function selectPlaylist(streams: readonly StreamMetadata[]): StreamKey[] { return playlistStreamKeys(streams); } /** * @public * Generate encryption parameters from from an encryption KeyID and Key, * in the form KEYID:KEY, both 16byte hexadecimal */ export function mkEncryption( encryption: string | undefined, pssh?: string | undefined, ): EncryptionSettings | undefined { let encryption_params = encryption ? encryption.split(":").map((s) => s.trim()) : undefined; if (encryption_params && encryption_params.length !== 2) { errorlog( "Warning: bad encryption format, must have two fields (hexadecimal key id and hexadecimal key)" ); encryption_params = undefined; } return encryption_params ? { encryptionKeyId: encryption_params[0], encryptionKey: encryption_params[1], encryptionPssh: pssh || "", } : undefined; } /** @public */ export function videoToPin(pin: Pins): (streams: StreamMetadata[]) => PinToKey { return function (streams: StreamMetadata[]): PinToKey { const video = videoStreamKeys(streams); return toPin(pin, video); }; } /** @public */ export function audioToPin(pin: Pins): (streams: StreamMetadata[]) => PinToKey { return function (streams: StreamMetadata[]): PinToKey { const audio = audioStreamKeys(streams); return toPin(pin, audio); }; } /** @public */ export function avToPin(pin: Pins): (streams: StreamMetadata[]) => PinToKey { return function (streams: StreamMetadata[]): PinToKey { const audio = audioStreamKeys(streams); const video = videoStreamKeys(streams); const keys = audio.concat(video); return toPin(pin, keys); }; } /** @public */ export function sourceToPin(source: string, pin: Pins): (streams: StreamMetadata[]) => PinToKey { return function (streams: StreamMetadata[]): PinToKey { const matching = removeUndefined(streams).filter((s) => s.streamKey?.sourceName == source).map((s) => s.streamKey) as StreamKey[]; return toPin(pin, matching); }; } /** @public */ export function subtitlesToPin(pin: Pins): (streams: StreamMetadata[]) => PinToKey { return function (streams: StreamMetadata[]): PinToKey { const subs = subtitleStreamKeys(streams); return toPin(pin, subs); }; } /** @public */ export function ancillaryToPin(pin: Pins): (streams: StreamMetadata[]) => PinToKey { return function (streams: StreamMetadata[]): PinToKey { const anc = ancillaryStreamKeys(streams); return toPin(pin, anc); }; } /** @internal */ function toPin(pin: Pins, keys: StreamKey[]): PinToKey { // We want to simply use // return { [pin]: keys }; // but doing so loses the types // let o: PinToKey = {}; // o[pin] = keys; // return o; return { [pin]: keys } as PinToKey; } /** * @public * Validation function to require at least one audio and at least one video stream. Often the default validation * will happen to ensure this, as audio and video are subscribed from separate media nodes, but when one media node * will produce both audio and video, default validation cannot know that both are required. */ export function requireAV(ctx: Context): boolean { const streams = ctx.streams; return ( audioStreamKeys(streams).length >= 1 && videoStreamKeys(streams).length >= 1 ) } /** * @public * Validation function to require exactly N audio and exactly M video streams. Often the default validation * will happen to ensure this, as audio and video are subscribed from separate media nodes, but when one media node * will produce both audio and video, default validation cannot know that both are required. */ export function requireExactAV({ audio, video }: { audio: number, video: number }): (ctx: Context) => boolean { return (ctx: Context) => { const streams = ctx.streams; return ( audioStreamKeys(streams).length == audio && videoStreamKeys(streams).length == video ) } } /** * @public * Compares two stream keys by value, returning true if the stream keys refer to the same stream */ export function streamKeysAreEqual(l: StreamKey, r: StreamKey): unknown { return l.streamId == r.streamId && l.sourceName == r.sourceName && l.programNumber == r.programNumber && l.renditionName == r.renditionName; } function unwrapOptional(wrapper?: { value: T }): T | undefined { return wrapper?.value }