WIP: video detecting plugin fixes

Author: khsa1

external_plugins/image_detecting_plugin/Dockerfile

@@ -7,6 +7,7 @@ RUN find / -path /proc -prune -o -path /sys -prune -o -path /dev -prune -o -writ
 RUN apt update && apt install -y motion vim
 RUN pip install watchdog
 RUN pip install PILLOW
+RUN pip install PyYaml
 
 # Config files
 ADD motion.default /etc/default/motion

external_plugins/image_detecting_plugin/image_detecting_plugin.py

@@ -5,21 +5,41 @@
 import time
 import uuid 
 import zmq 
+import yaml
+import logging
 from subprocess import Popen
 
 from watchdog.observers import Observer
 from watchdog.events import FileSystemEventHandler
 
 from ctevents import ctevents
-from pyevents.events import get_plugin_socket
+from pyevents.events import get_plugin_socket, send_quit_command
+from ctevents.ctevents import send_terminate_plugin_fb_event
 
 # Path to a directory that this plugin "watches" for new image files. 
 # By default, we set this directory to `/var/lib/motion` in the container, assuming
 # that the Linux Motion package will also be configured and running in the same container.
+log_level = os.environ.get("IMAGE_GENERATING_LOG_LEVEL", "INFO")
 DATA_MONITORING_PATH = os.environ.get("DATA_MONITORING_PATH", "/var/lib/motion")
 MIN_SECONDS_BETWEEN_IMAGES = float(os.environ.get("MIN_SECONDS_BETWEEN_IMAGES", "2.0"))
 MODE = os.environ.get("MODE", "demo")
 
+logger = logging.getLogger("Image Generating Plugin")
+if log_level == "DEBUG":
+    logger.setLevel(logging.DEBUG)
+elif log_level == "INFO":
+    logger.setLevel(logging.INFO)
+elif log_level == "WARN":
+    logger.setLevel(logging.WARN)
+elif log_level == "ERROR":
+    logger.setLevel(logging.ERROR)
+if not logger.handlers:
+    formatter = logging.Formatter('%(asctime)s %(levelname)s: %(message)s '
+            '[in %(pathname)s:%(lineno)d]')
+    handler = logging.StreamHandler()
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+
 def get_socket():
     """
     This function creates the zmq socket object and generates the event-engine plugin socket
@@ -166,23 +186,22 @@ def process_file(self, file_path):
             logging.error(f"Error processing {file_path}: {e}")
 
 def get_duration():
-    if MODE == 'video_simuation':
-        video_info_file = os.environ.get('TRAPS_VIDEO_INFO_PATH', '/video_info/video_info.yaml')
-        while True:
-            if os.path.exists(video_info_file):
-                try:
-                    with open(video_info_file, 'r') as f:
-                        video_info = yaml.safe_load(f)
-                except Exception as e:
-                    logging.error(f'Error processing {video_info_file}: {e}')
-            else:
-                sleep(1)
+    if MODE == 'simulation':
+        video_info_file = os.environ.get('TRAPS_VIDEO_INFO_PATH', '/video_info.yaml')
+
+        while not os.path.exists(video_info_file):
+            time.sleep(1)
+
+        try:
+            with open(video_info_file, 'r') as f:
+                video_info = yaml.safe_load(f)
+        except Exception as e:
+            logging.error(f'Error processing {video_info_file}: {e}')
 
         if 'duration' not in video_info.keys():
             logging.error(f'duration value not set in {video_info_file}')
         else:
-            duration = video_info['duration']
-
+            return video_info['duration']
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO,
@@ -205,6 +224,8 @@ def get_duration():
     log_handler = LogFileHandler(log_observer, '/var/log/motion/motion.log')
     log_observer.schedule(log_handler, '/var/log/motion', recursive=False)
     log_observer.start()
+    log_observer.join()
+    logger.info('motion has connected to camera')
 
     # instantiate and start the event handler 
     event_handler = NewFileHandler()
@@ -215,11 +236,17 @@ def get_duration():
     # run for specified video duration, or if undefined until interrupted 
     try:
         if duration:
+            logger.info(f'Running motion for {duration} seconds')
             time.sleep(duration)
+            observer.stop()
         else:
+            logger.info('No duration specified. Running motion indefinitely')
             while True:
                 time.sleep(1)
     except KeyboardInterrupt:
         observer.stop()
     observer.join()
     motion_proc.kill()
+    logger.info('Sending quit command')
+    send_terminate_plugin_fb_event(socket, "*", "35f20cdd-a404-4436-8df9-d80a9de91147")
+    send_quit_command(socket)

external_plugins/video_generating_plugin/Dockerfile

@@ -3,10 +3,9 @@ FROM tapis/camera_traps_py_3.13:$REL
 
 # Update apt packages and install dependencies
 RUN apt-get update && apt-get install -y ffmpeg v4l-utils libv4l-dev
+RUN pip install PyYaml requests
 
 ADD video_generating_plugin.py /video_generating_plugin.py
-ADD example_video.mp4 /example_video.mp4
-ADD ground_truth.yml /ground_truth.yml
 
 RUN chmod -R 0777 /video_generating_plugin.py || true 
 

external_plugins/video_generating_plugin/video_generating_plugin.py

@@ -1,15 +1,21 @@
 import os
 import zmq
-from ctevents import ctevents
-from ctevents.ctevents import send_terminating_plugin_fb_event
+from ctevents import MonitorPowerStartEvent, MonitorPowerStopEvent, PluginTerminateEvent
+from ctevents.ctevents import socket_message_to_typed_event, send_terminate_plugin_fb_event, send_monitor_power_start_fb_event
 from pyevents.events import get_plugin_socket, get_next_msg, send_quit_command
 import logging
-from subprocess import run, PIPE, STDOUT
+from subprocess import Popen, run, PIPE, STDOUT
+from math import ceil
+import yaml
+import requests
 
 log_level = os.environ.get("VIDEO_GENERATING_LOG_LEVEL", "INFO")
 input_video_path = os.environ.get("INPUT_VIDEO_PATH", "/example_video.mp4")
+use_ground_truth_url = os.environ.get("USE_CUSTOM_GROUND_TRUTH_FILE_URL", False)
+ground_truth_url = os.environ.get("CUSTOM_GROUND_TRUTH_URL")
 ground_truth_file = os.environ.get("GROUND_TRUTH_FILE", "/ground_truth.yml")
-device = os.environ.get("DEVICE", "http://0.0.0.0/8090")
+device = os.environ.get("DEVICE", "/dev/video0")
+mode = os.environ.get("MODE", "device")
 
 logger = logging.getLogger("Image Generating Plugin")
 if log_level == "DEBUG":
@@ -33,7 +39,7 @@ def get_socket():
     for the port configured for this plugin.
     """
     # get the port assigned to the Image Generating plugin
-    PORT = os.environ.get('VIDEO_GENERATING_PLUGIN_PORT', 6000)
+    PORT = os.environ.get('VIDEO_GENERATING_PLUGIN_PORT', 6003)
     # create the zmq context object
     context = zmq.Context()
     socket = get_plugin_socket(context, PORT)
@@ -42,18 +48,33 @@ def get_socket():
 
 def get_video_duration():
     result = run(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", input_video_path], stdout=PIPE, stderr=STDOUT)
-    return ceil(float(result.stdout))
+    duration = ceil(float(result.stdout))
+    logger.info(f'{input_video_path} has a duration of {duration} seconds')
+    return duration
 
 
 def load_ground_truth():
-    return None
-    with open(ground_truth_file, 'r') as f:
-        video_info = yaml.safe_load(f)
+    video_info = {}
+    try:
+        if use_ground_truth_url:
+            logger.info(f"Retrieving custom ground truth file: {ground_truth_url}")
+            response = requests.get(ground_truth_url)
+            if response.status_code == 200:
+                yml_content = response.content.decode('utf-8').splitlines()
+                video_info = yaml.safe_load(yml_content)
+        else:
+            with open(ground_truth_file, 'r') as f:
+                video_info = yaml.safe_load(f)
+    except FileNotFoundError:
+        logger.error(f"File not found: {ground_truth_file}")
+    except Exception as e:
+        logger.error(f'An error occurred: {e}')
     video_info['duration'] = get_video_duration()
-    OUTPUT_DIR = os.environ.get('TRAPS_VIDEO_OUTPUT_PATH', '/output')
+    OUTPUT_DIR = os.environ.get('TRAPS_VIDEO_OUTPUT_PATH', '/video_info')
     video_info_file = os.path.join(OUTPUT_DIR, 'video_info.yaml')
     with open(video_info_file, 'w') as f:
         yaml.dump(video_info, f)
+    logger.info(f'Updating {OUTPUT_DIR}/video_info.yaml')
 
 def monitor_generating_power():
     """
@@ -64,7 +85,7 @@ def monitor_generating_power():
     monitor_type = [1]
     monitor_seconds = 0
     if monitor_flag:
-        ctevents.send_monitor_power_start_fb_event(socket, pid, monitor_type, monitor_seconds)
+        send_monitor_power_start_fb_event(socket, pid, monitor_type, monitor_seconds)
         logger.info(f"Monitoring image generating power")
 
 def is_v4l2loopback_available():
@@ -76,42 +97,50 @@ def is_v4l2loopback_available():
 
 def stream_file_to_device(input_video_path):
     logger.info(f'starting video device stream to {device}')
-    run(['ffmpeg', '-re', '-stream_loop', '-1', '-i', input_video_path, '-f', 'v4l2', '-pix_fmt', 'yuv420p', device])
-    logger.info('Video stream finished')
-
-def stream_file_to_netcam(input_video_path, device):
-    logger.info(f'starting  netcam stream at {netcam_url}')
-    run(['ffmpeg', '-re', '-i', input_video_path, '-f', 'mjpeg', '-pix_fmt', 'yuv420p',  '-listen', '1', netcam_url])
-    logger.info('Video stream finished')
+    return Popen(['ffmpeg', '-re', '-stream_loop', '-1', '-i', input_video_path, '-f', 'v4l2', '-pix_fmt', 'yuv420p', device])
 
-def get_device_type():
-    if '/dev/video' in device:
-        return 'video_device'
-    elif 'http://' in device:
-        return 'netcam'
 
 def process_video(input_video_path, ground_truth):
     """
-    Main function that starts a video stream, either on /dev/video or as a netcam
+    Main function that starts a video stream on a /dev/video device
     """
     logger.info(f"The input video path specified by the user:{input_video_path}")
-    device_type = get_device_type()
-    if device_type == 'video_device':
-        if is_v4l2loopback_available():
-            stream_file_to_device(input_video_path)
-        else:
-            logger.info('v4l2loopback not available for {device}. Falling back to netcam stream.')
-            stream_file_to_netcam(input_video_path, DEFAULT_NETCAM)
+    if is_v4l2loopback_available():
+        return stream_file_to_device(input_video_path)
     else:
-        stream_file_to_netcam(input_video_path)
+        logger.warning('v4l2loopback not available for {device}. Shutting down')
 
 def main():
     global socket
-    socket = get_socket()
     ground_truth = load_ground_truth()
-    monitor_generating_power()
-    #motion_ready_signal(socket)
-    process_video(input_video_path, ground_truth)
+    stream_proc = None
+    if mode == 'device':
+        monitor_generating_power()
+        stream_proc = process_video(input_video_path, ground_truth)
+    done = False
+    while not done:
+        socket = get_socket()
+        try:
+            message = get_next_msg(socket)
+        except zmq.error.Again:
+            logger.debug(f"Got a zmq.error.Again; i.e., waited {SOCKET_TIMEOUT} ms without getting a message")
+            continue
+        except Exception as e:
+            logger.debug(f"Got exception from get_next_msg; type(e): {type(e)}; e: {e}")
+            done = True
+            logger.info("Video generating plugin stopping due to timeout limit...")
+            continue
+        if not message:
+            logger.info("No message found in get_next_msg")
+
+        event = socket_message_to_typed_event(message)
+        logger.info(f"Got a message from the event socket of type: {type(event)}")
+        if isinstance(event, PluginTerminateEvent):
+            logging.info('received PluginTerminateEvent')
+            done = True
+
+    if stream_proc:
+        stream_proc.kill()
     send_quit_command(socket)
 
 if __name__ == '__main__':

installer/Dockerfile

@@ -13,6 +13,7 @@ ADD example_input.yml /host/input.yml
 RUN mkdir /defaults 
 ADD defaults.yml /defaults/defaults.yml 
 ADD example_images /defaults/example_images
+ADD example_video /defaults/example_video
 ADD ground_truth.csv /defaults/ground_truth.csv
 
 # The installer app itself
@@ -23,4 +24,4 @@ ADD compile.py /compile.py
 RUN chmod 0777 -R /installer
 RUN chmod 0777  /compile.py 
 
-CMD [ "python", "/compile.py" ]
+CMD [ "python", "/compile.py" ]

installer/compile.py

@@ -98,11 +98,10 @@ def get_vars(input_data, default_data):
                                  'deploy_video_generating': True,
                                  'deploy_image_detecting': True,
                                  'deploy_reporter': False,
-                                 'deploy_ckn': True,
+                                 'deploy_ckn': False,
                                  'deploy_ckn_mqtt': False,
-                                 'deploy_oracle': True,
-                                 'motion_video_device': 'http://video_generating:8090',
-                                 'generating_video_device': 'http://0.0.0.0:8090',
+                                 'deploy_oracle': False,
+                                 'use_bundled_example_images': False,
                                  'inference_server': False}
 
     if vars.get("mode") == 'demo':
@@ -157,11 +156,16 @@ def get_vars(input_data, default_data):
             vars['model_id'] = '41d3ed40-b836-4a62-b3fb-67cee79f33d9-model'
 
     # for video simulations, determine if motion is using device or netcam
-    if vars.get('motion_video_device'):
-        if '/dev' in vars.get('motion_video_device'):
+    if vars.get('mode') == 'video_simulation':
+        if vars.get('motion_video_device'):
             vars['motion_video_type'] = 'device'
-        elif '://' in vars.get('motion_video_device'):
-            vars['motion_video_type'] = 'netcam'
+        else:
+            vars['motion_video_type'] = 'file'
+        if vars.get('use_example_video'):
+            vars['local_video_path'] = './video.mp4'
+        else:
+            if not vars.get('local_video_path'):
+                vars['local_video_path'] = './video.mp4'
 
     # Add the installer's UID and GID
     vars["uid"] = uid
@@ -329,6 +333,30 @@ def generate_additional_directories(vars, full_install_dir):
             sys.exit(1)
         print(f"Using URL for images: {vars['source_image_url']}")
 
+    if vars['use_example_video'] == True: 
+        try:
+            shutil.copy("/defaults/example_video/example_video.mp4", os.path.join(full_install_dir, "video.mp4"))
+            shutil.copy("/defaults/example_video/ground_truth.yml", os.path.join(full_install_dir, "ground_truth.yml"))
+        except Exception as e:
+            print(f"ERROR: Could not copy bundled example video; error: {e}")
+            print("Exiting...")
+            sys.exit(1)
+    else:
+        if vars['source_video_url']:
+            try:
+                rsp = requests.get(vars['source_video_url'])
+                rsp.raise_for_status()
+            except Exception as e:
+                print(f'Error: could not download video at URL {source_video_url}; details: {e}')
+                sys.exit(1)
+            video_install_path = os.path.join(full_install_dir, 'video.mp4')
+            with open(video_install_path, 'wb') as f:
+                f.write(rsp.content)
+        elif vars['local_video_path']:
+            full_video_path = os.path.join(full_install_dir, vars.get('local_video_path'))
+            if not os.path.exists(full_video_path):
+                print(f'ERROR: local_video_path must be a relative path to the install directory and must already exist; the computed path ({full_video_dir}) does not exist.\nExiting...')
+                sys.exit(1)
 
     # create output directories if they do not exist     
     images_output_dir = os.path.join(full_install_dir, vars["images_output_dir"])
@@ -347,6 +375,11 @@ def generate_additional_directories(vars, full_install_dir):
     detection_output_dir = os.path.join(full_install_dir, vars["detection_reporter_plugin_output_dir"])
     if not os.path.exists(detection_output_dir):
         os.makedirs(detection_output_dir)
+
+    if vars['deploy_video_generating']:
+        video_output_dir = os.path.join(full_install_dir, vars["video_output_dir"])
+        if not os.path.exists(video_output_dir):
+            os.makedirs(video_output_dir)
 
 
 def main():

installer/defaults.yml

@@ -33,13 +33,16 @@ motion_event_gap: 1
 motion_threshold: 100
 motion_width: 640
 motion_height: 480
+image_detecting_log_level: DEBUG
 
 # video generating plugin
 deploy_video_generating: false
 video_generating_plugin_image: tapis/video_generating_plugin
 video_generating_log_level: DEBUG
+video_output_dir: video_output_dir
+use_example_video: true
 source_video_url:
-generating_video_device:
+local_video_path:
 
 
 # Image Scoring Plugin