Pynest

.gitignore

@@ -16,7 +16,7 @@ leftover*
 # tmp file workaround
 lost*
 
-# ignore bin paths
+# Ignore bin paths
 bin/ustreamer
 bin/camera-streamer
 
@@ -30,3 +30,6 @@ tools/.config
 
 # Ignore pkglist
 tools/pkglist.sh
+
+# Ignore pycache
+**/__pycache__/

crowsnest

@@ -19,19 +19,19 @@ set -Ee
 # Base Path
 BASE_CN_PATH="$(dirname "$(readlink -f "${0}")")"
 
-## Import Librarys
-# shellcheck source-path=SCRIPTDIR/../libs/
-. "${BASE_CN_PATH}/libs/camera-streamer.sh"
-. "${BASE_CN_PATH}/libs/configparser.sh"
-. "${BASE_CN_PATH}/libs/core.sh"
-. "${BASE_CN_PATH}/libs/hwhandler.sh"
-. "${BASE_CN_PATH}/libs/init_stream.sh"
-. "${BASE_CN_PATH}/libs/logging.sh"
-. "${BASE_CN_PATH}/libs/messages.sh"
-. "${BASE_CN_PATH}/libs/ustreamer.sh"
-. "${BASE_CN_PATH}/libs/v4l2_control.sh"
-. "${BASE_CN_PATH}/libs/versioncontrol.sh"
-. "${BASE_CN_PATH}/libs/watchdog.sh"
+function missing_args_msg {
+    echo -e "crowsnest: Missing Arguments!"
+    echo -e "\n\tTry: crowsnest -h\n"
+}
+
+function check_cfg {
+    if [ ! -r "${1}" ]; then
+        log_msg "ERROR: No Configuration File found. Exiting!"
+        exit 1
+    else
+        return 0
+    fi
+}
 
 #### MAIN
 ## Args given?
@@ -74,16 +74,25 @@ while getopts ":vhc:s:d" arg; do
     esac
 done
 
-init_logging
-initial_check
-construct_streamer
+function set_log_path {
+    #Workaround sed ~ to BASH VAR $HOME
+    CROWSNEST_LOG_PATH=$(get_param "crowsnest" log_path | sed "s#^~#${HOME}#gi")
+    declare -g CROWSNEST_LOG_PATH
+    #Workaround: Make Dir if not exist
+    if [ ! -d "$(dirname "${CROWSNEST_LOG_PATH}")" ]; then
+        mkdir -p "$(dirname "${CROWSNEST_LOG_PATH}")"
+    fi
+}
 
-## Loop and Watchdog
-## In this case watchdog acts more like a "cable defect detector"
-## The User gets a message if Device is lost.
-clean_watchdog
-while true ; do
-    crowsnest_watchdog
-    sleep 120 & sleep_pid="$!"
-    wait "${sleep_pid}"
-done
+function get_param {
+    local cfg section param
+    cfg="${CROWSNEST_CFG}"
+    section="${1}"
+    param="${2}"
+    crudini --get "${cfg}" "${section}" "${param}" 2> /dev/null | \
+    sed 's/\#.*//;s/[[:space:]]*$//'
+    return
+}
+
+set_log_path
+python3 "${BASE_CN_PATH}/crowsnest.py" -c "${CROWSNEST_CFG}" -l "${CROWSNEST_LOG_PATH}"

crowsnest.py

@@ -0,0 +1,134 @@
+#!/usr/bin/python3
+
+import argparse
+import configparser
+import asyncio
+import signal
+import traceback
+
+from pylibs.components.crowsnest import Crowsnest
+from pylibs import utils, watchdog, logger, logging_helper
+
+parser = argparse.ArgumentParser(
+    prog='Crowsnest',
+    description='Crowsnest - A webcam daemon for Raspberry Pi OS distributions like MainsailOS'
+)
+config = configparser.ConfigParser(inline_comment_prefixes='#')
+
+parser.add_argument('-c', '--config', help='Path to config file', required=True)
+parser.add_argument('-l', '--log_path', help='Path to log file', required=True)
+
+args = parser.parse_args()
+
+watchdog_running = True
+
+def initial_parse_config():
+    global crowsnest, config, args
+    config_path = args.config
+    try:
+        config.read(config_path)
+    except configparser.ParsingError as e:
+        logger.log_multiline(e.message, logger.log_error)
+        logger.log_error("Failed to parse config! Exiting...")
+        exit(1)
+    crowsnest = Crowsnest('crowsnest')
+    if not config.has_section('crowsnest') or not crowsnest.parse_config_section(config['crowsnest']):
+        logger.log_error("Failed to parse config for '[crowsnest]' section! Exiting...")
+        exit(1)
+
+async def start_sections():
+    global config, sect_exec_tasks
+    sect_objs = []
+    sect_exec_tasks = set()
+
+    # Catches SIGINT and SIGTERM to exit gracefully and cancel all tasks
+    signal.signal(signal.SIGINT, exit_gracefully)
+    signal.signal(signal.SIGTERM, exit_gracefully)
+
+    try:
+        if not len(config.sections()) > 1:
+            logger.log_quiet("No Cams / Services to start! Exiting ...")
+            return
+        logger.log_quiet("Try to parse configured Cams / Services...")
+        for section in config.sections():
+            section_header = section.split(' ')
+            section_object = None
+            section_keyword = section_header[0]
+
+            # Skip crowsnest section
+            if section_keyword == 'crowsnest':
+                continue
+
+            section_name = ' '.join(section_header[1:])
+            component = utils.load_component(section_keyword, section_name)
+            logger.log_quiet(f"Parse configuration of section [{section}] ...")
+            if component.parse_config_section(config[section]):
+                sect_objs.append(component)
+                logger.log_quiet(f"Configuration of section [{section}] looks good. Continue ...")
+            else:
+                logger.log_error(f"Failed to parse config for section [{section}]! Skipping ...")
+
+        logger.log_quiet("Try to start configured Cams / Services ...")
+        if sect_objs:
+            lock = asyncio.Lock()
+            for section_object in sect_objs:
+                task = asyncio.create_task(section_object.execute(lock))
+                sect_exec_tasks.add(task)
+
+            # Lets sect_exec_tasks finish first
+            await asyncio.sleep(0)
+            async with lock:
+                logger.log_quiet("... Done!")
+        else:
+            logger.log_quiet("No Service started! Exiting ...")
+
+        for task in sect_exec_tasks:
+            if task is not None:
+                await task
+    except Exception as e:
+        logger.log_multiline(traceback.format_exc().strip(), logger.log_error)
+    finally:
+        for task in sect_exec_tasks:
+            if task is not None:
+                task.cancel()
+        watchdog.running = False
+        logger.log_quiet("Shutdown or Killed by User!")
+        logger.log_quiet("Please come again :)")
+        logger.log_quiet("Goodbye...")
+
+async def exit_gracefully(signum, frame):
+    asyncio.sleep(1)
+
+async def main():
+    global args, crowsnest
+    logger.setup_logging(args.log_path)
+    logging_helper.log_initial()
+
+    initial_parse_config()
+
+    if crowsnest.parameters['delete_log'].value:
+        logger.logger.handlers.clear()
+        logger.setup_logging(args.log_path, 'w')
+        logging_helper.log_initial()
+
+    logger.set_log_level(crowsnest.parameters['log_level'].value)
+
+    logging_helper.log_host_info()
+    logging_helper.log_streamer()
+    logging_helper.log_config(args.config)
+    logging_helper.log_cams()
+
+    task1 = asyncio.create_task(start_sections())
+    await asyncio.sleep(0)
+    task2 = asyncio.create_task(watchdog.run_watchdog())
+
+    await task1
+    if task2:
+        task2.cancel()
+
+if __name__ == "__main__":
+    loop = asyncio.get_event_loop()
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()

pylibs/__init__.py

@@ -0,0 +1 @@
+#!/usr/bin/python3

pylibs/camera/__init__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/python3
+
+from .types.uvc import UVC
+from .types.legacy import Legacy
+from .types.libcamera import Libcamera
+from .camera import Camera
+from . import camera_manager

pylibs/camera/camera.py

@@ -0,0 +1,26 @@
+#!/usr/bin/python3
+
+import os
+from abc import ABC, abstractmethod
+
+class Camera(ABC):
+    def __init__(self, path: str, *args, **kwargs) -> None:
+        self.path = path
+        self.control_values = {}
+        self.formats = {}
+
+    def path_equals(self, path: str) -> bool:
+        return self.path == os.path.realpath(path)
+
+    @abstractmethod
+    def get_formats_string(self) -> str:
+        pass
+
+    @abstractmethod
+    def get_controls_string(self) -> str:
+        pass
+
+    @staticmethod
+    @abstractmethod
+    def init_camera_type() -> list:
+        pass

pylibs/camera/camera_manager.py

@@ -0,0 +1,24 @@
+#!/usr/bin/python3
+
+from .camera import Camera
+
+def get_all_cameras() -> list:
+    global cameras
+    try:
+        cameras
+    except NameError:
+        cameras = []
+    return cameras
+
+def get_cam_by_path(path: str) -> Camera:
+    global cameras
+    for camera in get_all_cameras():
+        if camera.path_equals(path):
+            return camera
+    return None
+
+def init_camera_type(obj: Camera) -> list:
+    global cameras
+    cams = obj.init_camera_type()
+    get_all_cameras().extend(cams)
+    return cams

pylibs/camera/types/legacy.py

@@ -0,0 +1,10 @@
+from . import uvc
+from ... import v4l2
+
+class Legacy(uvc.UVC):
+    @staticmethod
+    def init_camera_type() -> list:
+        legacy_path = v4l2.ctl.get_dev_path_by_name('mmal')
+        if not legacy_path:
+            return []
+        return [Legacy(legacy_path)]

pylibs/camera/types/libcamera.py

@@ -0,0 +1,83 @@
+import shutil, re
+
+from ... import utils
+from .. import camera
+
+class Libcamera(camera.Camera):
+    def __init__(self, path, *args, **kwargs) -> None:
+        self.path = path
+        self.control_values = self._get_controls()
+        self.formats = []
+
+    def _get_controls(self) -> str:
+        ctrls = {}
+        try:
+            from libcamera import CameraManager, Rectangle
+
+            def rectangle_to_tuple(rectangle):
+                return (rectangle.x, rectangle.y, rectangle.width, rectangle.height)
+
+            libcam_cm = CameraManager.singleton()
+            for cam in libcam_cm.cameras:
+                if cam.id != self.path:
+                    continue
+                for k, v in cam.controls.items():
+                    if isinstance(v.min, Rectangle):
+                        ctrls[k.name] = {
+                            'min': rectangle_to_tuple(v.min),
+                            'max': rectangle_to_tuple(v.max),
+                            'default': rectangle_to_tuple(v.default)
+                        }
+                    else:
+                        ctrls[k.name] = {
+                            'min': v.min,
+                            'max': v.max,
+                            'default': v.default
+                        }
+        except ImportError:
+            pass
+        return ctrls
+
+    def _get_formats(self, libcamera_output: str) -> list:
+        resolutions = re.findall(
+            rf"{self.path}.*?:.*?: (.*?)(?=\n\n|\n *')",
+            libcamera_output, flags=re.DOTALL
+        )
+        res = []
+        if resolutions:
+            res = [r.strip() for r in resolutions[0].split('\n')]
+        return res
+
+    def get_formats_string(self) -> str:
+        message = ''
+        for res in self.formats:
+            message += f"{res}\n"
+        return message[:-1]
+
+    def get_controls_string(self) -> str:
+        if not self.control_values:
+            return "apt package 'python3-libcamera' is not installed! " \
+                   "Make sure to install it to log the controls!"
+        message = ''
+        for name, value in self.control_values.items():
+            min, max, default = value.values()
+            str_first = f"{name} ({self.get_type_str(min)})"
+            str_indent = (30 - len(str_first)) * ' ' + ': '
+            str_second = f"min={min} max={max} default={default}"
+            message += str_first + str_indent + str_second + '\n'
+        return message.strip()
+
+    def get_type_str(self, obj) -> str:
+        return str(type(obj)).split('\'')[1]
+
+    @staticmethod
+    def init_camera_type() -> list:
+        cmd = shutil.which('libcamera-hello')
+        if not cmd:
+            return {}
+        libcam_cmd =f'{cmd} --list-cameras'
+        libcam = utils.execute_shell_command(libcam_cmd, strip=False)
+        cams = [Libcamera(path) for path in re.findall(r'\((/base.*?)\)', libcam)]
+        for cam in cams:
+            cam.formats = cam._get_formats(libcam)
+        return cams