From 154ab42d15d1f5fb8897f437a80cbf1d88c1c819 Mon Sep 17 00:00:00 2001
From: Koushik Dutta <koushd@gmail.com>
Date: Sat, 22 Apr 2023 08:16:34 -0700
Subject: [PATCH] videonalaysis: refactor to avoid holding onto generators

---
 plugins/objectdetector/src/main.ts         | 63 ++++++++++++----------
 plugins/python-codecs/package-lock.json    |  4 +-
 plugins/python-codecs/package.json         |  2 +-
 plugins/python-codecs/src/gst_generator.py |  2 +
 server/package-lock.json                   |  4 +-
 5 files changed, 41 insertions(+), 34 deletions(-)

diff --git a/plugins/objectdetector/src/main.ts b/plugins/objectdetector/src/main.ts
index bacb2beb7..8cb2e9560 100644
--- a/plugins/objectdetector/src/main.ts
+++ b/plugins/objectdetector/src/main.ts
@@ -17,7 +17,7 @@ const { systemManager } = sdk;
 const defaultDetectionDuration = 20;
 const defaultDetectionInterval = 60;
 const defaultDetectionTimeout = 60;
-const defaultMotionDuration = 10;
+const defaultMotionDuration = 30;
 
 const BUILTIN_MOTION_SENSOR_ASSIST = 'Assist';
 const BUILTIN_MOTION_SENSOR_REPLACE = 'Replace';
@@ -277,28 +277,9 @@ class ObjectDetectionMixin extends SettingsMixinDeviceBase<VideoCamera & Camera
       });
   }
 
-  async runPipelineAnalysis(signal: Deferred<void>, options: {
+  async createFrameGenerator(signal: Deferred<void>, options: {
     snapshotPipeline: boolean,
-  }) {
-    const start = Date.now();
-    this.analyzeStop = start + this.getDetectionDuration();
-
-    let lastStatusTime = Date.now();
-    let lastStatus = 'starting';
-    const updatePipelineStatus = (status: string) => {
-      lastStatus = status;
-      lastStatusTime = Date.now();
-    }
-
-    let frameGenerator: AsyncGenerator<VideoFrame & MediaObject, void>;
-    let detectionGenerator: AsyncGenerator<ObjectDetectionGeneratorResult, void>;
-    const interval = setInterval(() => {
-      if (Date.now() - lastStatusTime > 30000) {
-        signal.resolve();
-        this.console.error('VideoAnalysis is hung and will terminate:', lastStatus);
-      }
-    }, 30000);
-    signal.promise.finally(() => clearInterval(interval));
+  }, updatePipelineStatus: (status: string) => void) {
 
     let newPipeline: string = this.newPipeline;
     if (!this.hasMotionType && (!newPipeline || newPipeline === 'Default')) {
@@ -312,7 +293,7 @@ class ObjectDetectionMixin extends SettingsMixinDeviceBase<VideoCamera & Camera
       options.snapshotPipeline = true;
       this.console.log('decoder:', 'Snapshot +', this.objectDetection.name);
       const self = this;
-      frameGenerator = (async function* gen() {
+      return (async function* gen() {
         try {
           while (!signal.finished) {
             const now = Date.now();
@@ -362,7 +343,7 @@ class ObjectDetectionMixin extends SettingsMixinDeviceBase<VideoCamera & Camera
         audio: null,
       });
 
-      frameGenerator = await videoFrameGenerator.generateVideoFrames(stream, {
+      return await videoFrameGenerator.generateVideoFrames(stream, {
         queue: 0,
         resize: this.model?.inputSize ? {
           width: this.model.inputSize[0],
@@ -371,17 +352,41 @@ class ObjectDetectionMixin extends SettingsMixinDeviceBase<VideoCamera & Camera
         format: this.model?.inputFormat,
       });
     }
+  }
+
+  async runPipelineAnalysis(signal: Deferred<void>, options: {
+    snapshotPipeline: boolean,
+  }) {
+    const start = Date.now();
+    this.analyzeStop = start + this.getDetectionDuration();
+
+    let lastStatusTime = Date.now();
+    let lastStatus = 'starting';
+    const updatePipelineStatus = (status: string) => {
+      lastStatus = status;
+      lastStatusTime = Date.now();
+    }
+
+    const interval = setInterval(() => {
+      if (Date.now() - lastStatusTime > 30000) {
+        signal.resolve();
+        this.console.error('VideoAnalysis is hung and will terminate:', lastStatus);
+      }
+    }, 30000);
+    signal.promise.finally(() => clearInterval(interval));
 
     const currentDetections = new Set<string>();
     let lastReport = 0;
-    detectionGenerator = await sdk.connectRPCObject(await this.objectDetection.generateObjectDetections(frameGenerator, {
-      settings: this.getCurrentSettings(),
-      sourceId: this.id,
-    }));
 
     updatePipelineStatus('waiting result');
 
-    for await (const detected of detectionGenerator) {
+    for await (const detected of
+      await sdk.connectRPCObject(
+        await this.objectDetection.generateObjectDetections(
+          await this.createFrameGenerator(signal, options, updatePipelineStatus), {
+          settings: this.getCurrentSettings(),
+          sourceId: this.id,
+        }))) {
       if (signal.finished) {
         break;
       }
diff --git a/plugins/python-codecs/package-lock.json b/plugins/python-codecs/package-lock.json
index 7cb8dcd1f..cf3255a7a 100644
--- a/plugins/python-codecs/package-lock.json
+++ b/plugins/python-codecs/package-lock.json
@@ -1,12 +1,12 @@
 {
    "name": "@scrypted/python-codecs",
-   "version": "0.1.40",
+   "version": "0.1.41",
    "lockfileVersion": 3,
    "requires": true,
    "packages": {
       "": {
          "name": "@scrypted/python-codecs",
-         "version": "0.1.40",
+         "version": "0.1.41",
          "devDependencies": {
             "@scrypted/sdk": "file:../../sdk"
          }
diff --git a/plugins/python-codecs/package.json b/plugins/python-codecs/package.json
index 5b1ff801a..2fe18ae79 100644
--- a/plugins/python-codecs/package.json
+++ b/plugins/python-codecs/package.json
@@ -1,6 +1,6 @@
 {
    "name": "@scrypted/python-codecs",
-   "version": "0.1.40",
+   "version": "0.1.41",
    "description": "Python Codecs for Scrypted",
    "keywords": [
       "scrypted",
diff --git a/plugins/python-codecs/src/gst_generator.py b/plugins/python-codecs/src/gst_generator.py
index 8dbede268..8360f7e6a 100644
--- a/plugins/python-codecs/src/gst_generator.py
+++ b/plugins/python-codecs/src/gst_generator.py
@@ -30,6 +30,8 @@ async def createPipelineIterator(pipeline: str):
         elif t == str(Gst.MessageType.WARNING):
             err, debug = message.parse_warning()
             print('Warning: %s: %s\n' % (err, debug))
+            print('Ending stream due to warning. If this camera is causing errors, switch to the libav decoder.');
+            finish();
         elif t == str(Gst.MessageType.ERROR):
             err, debug = message.parse_error()
             print('Error: %s: %s\n' % (err, debug))
diff --git a/server/package-lock.json b/server/package-lock.json
index f38029345..aa54eea18 100644
--- a/server/package-lock.json
+++ b/server/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "@scrypted/server",
-  "version": "0.7.85",
+  "version": "0.7.86",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@scrypted/server",
-      "version": "0.7.85",
+      "version": "0.7.86",
       "license": "ISC",
       "dependencies": {
         "@mapbox/node-pre-gyp": "^1.0.10",