Merge pull request #15 from Eyevinn/fix/start-prod-w-websocket

Solves the problem with starting productions with websocket-sources

.env.sample

@@ -4,6 +4,7 @@ MONGODB_URI=${MONGODB_URI:-mongodb://api:password@localhost:27017/live-gui}
 # Ateliere Live System Controlleer
 LIVE_URL=${LIVE_URL:-https://localhost:8080}
 LIVE_CREDENTIALS=${LIVE_CREDENTIALS:-admin:admin}
+CONTROL_PANEL_WS==${}
 # This ENV variable disables SSL Verification, use if the above LIVE_URL doesn't have a proper certificate
 NODE_TLS_REJECT_UNAUTHORIZED=${NODE_TLS_REJECT_UNAUTHORIZED:-1}
 
@@ -14,3 +15,6 @@ BCRYPT_SALT_ROUNDS=${BCRYPT_SALT_ROUNDS:-10}
 
 # i18n
 UI_LANG=${UI_LANG:-en}
+
+# Mediaplayer - path on the system controller
+MEDIAPLAYER_PLACEHOLDER=/media/media_placeholder.mp4

package.json

@@ -10,7 +10,7 @@
     "pretty:format": "prettier --write .",
     "typecheck": "tsc --noEmit -p tsconfig.json",
     "lint": "next lint",
-    "dev": "./update_gui_version.sh && next dev",
+    "dev": "next dev",
     "build": "next build",
     "start": "next start",
     "version:rc": "npm version prerelease --preid=rc",
@@ -32,6 +32,7 @@
     "@sinclair/typebox": "^0.25.24",
     "@tabler/icons": "^2.22.0",
     "@tabler/icons-react": "^2.20.0",
+    "@types/ws": "^8.5.12",
     "bcrypt": "^5.1.0",
     "cron": "^2.3.1",
     "date-fns": "^2.30.0",
@@ -48,7 +49,8 @@
     "react-dom": "^18.2.0",
     "react-hot-toast": "^2.4.1",
     "tailwind-merge": "^1.13.2",
-    "uuid": "^9.0.0"
+    "uuid": "^9.0.0",
+    "ws": "^8.18.0"
   },
   "devDependencies": {
     "@commitlint/cli": "^17.4.2",

src/api/ateliereLive/pipelines/multiviews/multiviews.ts

@@ -64,13 +64,12 @@ export async function createMultiviewForPipeline(
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     productionSettings.pipelines[multiviewIndex].pipeline_id!;
   const sources = await getSourcesByIds(
-    sourceRefs.map((ref) => ref._id.toString())
+    sourceRefs.map((ref) => (ref._id ? ref._id.toString() : ''))
   );
   const sourceRefsWithLabels = sourceRefs.map((ref) => {
+    const refId = ref._id ? ref._id.toString() : '';
     if (!ref.label) {
-      const source = sources.find(
-        (source) => source._id.toString() === ref._id.toString()
-      );
+      const source = sources.find((source) => source._id.toString() === refId);
       ref.label = source?.name || '';
     }
     return ref;

src/api/ateliereLive/pipelines/streams/streams.ts

@@ -66,6 +66,7 @@ export async function createStream(
         return pipeline.uuid;
       })
     );
+
     const ingestUuid = await getUuidFromIngestName(
       source.ingest_name,
       false
@@ -79,13 +80,15 @@ export async function createStream(
       source.ingest_source_name,
       false
     );
+
     const audioMapping =
       source.audio_stream.audio_mapping &&
       source.audio_stream.audio_mapping.length > 0
         ? source.audio_stream.audio_mapping
         : [[0, 1]];
 
     await initDedicatedPorts();
+
     for (const pipeline of production_settings.pipelines) {
       const availablePorts = getAvailablePortsForIngest(
         source.ingest_name,
@@ -101,28 +104,29 @@ export async function createStream(
       Log().info(
         `Allocated port ${availablePort} on '${source.ingest_name}' for ${source.ingest_source_name}`
       );
+
       const stream: PipelineStreamSettings = {
+        ingest_id: ingestUuid,
+        source_id: sourceId,
         pipeline_id: pipeline.pipeline_id!,
+        input_slot: input_slot,
         alignment_ms: pipeline.alignment_ms,
-        audio_format: pipeline.audio_format,
-        audio_sampling_frequency: pipeline.audio_sampling_frequency,
-        bit_depth: pipeline.bit_depth,
-        convert_color_range: pipeline.convert_color_range,
-        encoder: pipeline.encoder,
-        encoder_device: pipeline.encoder_device,
-        format: pipeline.format,
+        max_network_latency_ms: pipeline.max_network_latency_ms,
+        width: pipeline.width,
+        height: pipeline.height,
         frame_rate_d: pipeline.frame_rate_d,
         frame_rate_n: pipeline.frame_rate_n,
+        format: pipeline.format,
+        encoder: pipeline.encoder,
+        encoder_device: pipeline.encoder_device,
         gop_length: pipeline.gop_length,
-        height: pipeline.height,
-        max_network_latency_ms: pipeline.max_network_latency_ms,
         pic_mode: pipeline.pic_mode,
-        speed_quality_balance: pipeline.speed_quality_balance,
         video_kilobit_rate: pipeline.video_kilobit_rate,
-        width: pipeline.width,
-        ingest_id: ingestUuid,
-        source_id: sourceId,
-        input_slot,
+        bit_depth: pipeline.bit_depth,
+        speed_quality_balance: pipeline.speed_quality_balance,
+        convert_color_range: pipeline.convert_color_range,
+        audio_sampling_frequency: pipeline.audio_sampling_frequency,
+        audio_format: pipeline.audio_format,
         audio_mapping: JSON.stringify(audioMapping),
         interfaces: [
           {
@@ -131,6 +135,7 @@ export async function createStream(
           }
         ]
       };
+
       try {
         Log().info(
           `Connecting '${source.ingest_name}/${ingestUuid}}:${source.ingest_source_name}' to '${pipeline.pipeline_name}/${pipeline.pipeline_id}'`
@@ -147,6 +152,7 @@ export async function createStream(
         Log().info(
           `Stream '${result.stream_uuid}' from '${source.ingest_name}/${ingestUuid}' to '${pipeline.pipeline_name}/${pipeline.pipeline_id}' connected`
         );
+
         sourceToPipelineStreams.push({
           source_id: source._id.toString(),
           stream_uuid: result.stream_uuid,

src/api/ateliereLive/websocket.ts

@@ -0,0 +1,40 @@
+import WebSocket from 'ws';
+
+function createWebSocket(): Promise<WebSocket> {
+  return new Promise((resolve, reject) => {
+    const ws = new WebSocket(`ws://${process.env.CONTROL_PANEL_WS}`);
+    ws.on('error', reject);
+    ws.on('open', () => {
+      // const send = ws.send.bind(ws);
+      // ws.send = (message) => {
+      //   console.debug(`[websocket] sending message: ${message}`);
+      //   send(message);
+      // };
+      resolve(ws);
+    });
+  });
+}
+
+export async function createControlPanelWebSocket() {
+  const ws = await createWebSocket();
+  return {
+    createHtml: (input: number) => {
+      ws.send(`html create ${input} 1920 1080`);
+    },
+    createMediaplayer: (input: number) => {
+      ws.send(`media create ${input} ${process.env.MEDIAPLAYER_PLACEHOLDER}`);
+    },
+    closeHtml: (input: number) => {
+      ws.send(`html close ${input}`);
+      ws.send('html reset');
+    },
+    closeMediaplayer: (input: number) => {
+      ws.send(`media close ${input}`);
+      ws.send('media reset');
+    },
+    close: () =>
+      setTimeout(() => {
+        ws.close();
+      }, 1000)
+  };
+}

src/api/manager/productions.ts

@@ -28,21 +28,44 @@ export async function setProductionsIsActiveFalse(): Promise<
 export async function putProduction(
   id: string,
   production: Production
-): Promise<void> {
+): Promise<Production> {
   const db = await getDatabase();
+  const newSourceId = new ObjectId().toString();
+
+  const sources = production.sources
+    ? production.sources.flatMap((singleSource) => {
+        return singleSource._id
+          ? singleSource
+          : {
+              _id: newSourceId,
+              type: singleSource.type,
+              label: singleSource.label,
+              input_slot: singleSource.input_slot
+            };
+      })
+    : [];
+
   await db.collection('productions').findOneAndReplace(
     { _id: new ObjectId(id) },
     {
       name: production.name,
       isActive: production.isActive,
-      sources: production.sources,
+      sources: sources,
       production_settings: production.production_settings
     }
   );
 
   if (!production.isActive) {
     deleteMonitoring(db, id);
   }
+
+  return {
+    _id: new ObjectId(id).toString(),
+    name: production.name,
+    isActive: production.isActive,
+    sources: sources,
+    production_settings: production.production_settings
+  };
 }
 
 export async function postProduction(data: Production): Promise<ObjectId> {

src/api/manager/sources.ts

@@ -1,6 +1,6 @@
 import inventory from './mocks/inventory.json';
 import { Source } from '../../interfaces/Source';
-import { ObjectId } from 'mongodb';
+import { ObjectId, OptionalId, WithId } from 'mongodb';
 import { getDatabase } from '../mongoClient/dbClient';
 
 export function getMockedSources() {
@@ -9,37 +9,45 @@ export function getMockedSources() {
 
 export async function postSource(data: Source): Promise<ObjectId> {
   const db = await getDatabase();
-  return (await db.collection('inventory').insertOne(data))
-    .insertedId as ObjectId;
+  const insertData: OptionalId<Omit<Source, '_id'>> & { _id?: ObjectId } = {
+    ...data,
+    _id: typeof data._id === 'string' ? new ObjectId(data._id) : data._id
+  };
+  const result = await db.collection('inventory').insertOne(insertData);
+  return result.insertedId as ObjectId;
 }
 
 export async function getSources() {
   const db = await getDatabase();
   return await db.collection<Source>('inventory').find().toArray();
 }
-
-export async function getSourcesByIds(_ids: string[]) {
+export async function getSourcesByIds(
+  _ids: string[]
+): Promise<WithId<Source>[]> {
   const db = await getDatabase().catch(() => {
-    throw "Can't connect to Database";
-  });
-  const objectIds = _ids.map((id: string) => {
-    return new ObjectId(id);
+    throw new Error("Can't connect to Database");
   });
+  const objectIds = _ids.map((id: string) => new ObjectId(id));
 
-  return (
-    await db
-      .collection<Source>('inventory')
-      .find({
-        _id: {
-          $in: objectIds
-        }
-      })
-      .toArray()
-  ).sort(
-    (a, b) =>
-      _ids.findIndex((id) => a._id.equals(id)) -
-      _ids.findIndex((id) => b._id.equals(id))
-  );
+  const sources = await db
+    .collection<Source>('inventory')
+    .find({
+      _id: {
+        $in: objectIds
+      }
+    })
+    .toArray();
+
+  return sources.sort((a, b) => {
+    const findIndex = (id: ObjectId | string) =>
+      _ids.findIndex((originalId) =>
+        id instanceof ObjectId
+          ? id.equals(new ObjectId(originalId))
+          : id === originalId
+      );
+
+    return findIndex(a._id) - findIndex(b._id);
+  });
 }
 
 export async function updateSource(source: any) {