ipcam: rewrite motion detection system

1 files changed, 369 insertions, 0 deletions

diff --git a/src/ipcam_server.py b/src/ipcam_server.py
new file mode 100755
index 0000000..d212be2
--- /dev/null
+++ b/src/ipcam_server.py
@@ -0,0 +1,369 @@
+#!/usr/bin/env python3
+import logging
+import os
+import asyncio
+import time
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+from home.config import config
+from home.util import parse_addr
+from home import http
+from home.database.sqlite import SQLiteBase
+from home.camera import util as camutil
+
+from enum import Enum
+from typing import Optional, Union
+from datetime import datetime, timedelta
+
+
+class TimeFilterType(Enum):
+    FIX = 'fix'
+    MOTION = 'motion'
+
+
+def valid_recording_name(filename: str) -> bool:
+    return filename.startswith('record_') and filename.endswith('.mp4')
+
+
+def filename_to_datetime(filename: str) -> datetime:
+    filename = os.path.basename(filename).replace('record_', '').replace('.mp4', '')
+    return datetime.strptime(filename, datetime_format)
+
+
+# ipcam database
+# --------------
+
+class IPCamServerDatabase(SQLiteBase):
+    SCHEMA = 3
+
+    def __init__(self):
+        super().__init__()
+
+    def schema_init(self, version: int) -> None:
+        cursor = self.cursor()
+
+        if version < 1:
+            # timestamps
+            cursor.execute("""CREATE TABLE IF NOT EXISTS timestamps (
+                camera INTEGER PRIMARY KEY,
+                fix_time INTEGER NOT NULL,
+                motion_time INTEGER NOT NULL
+            )""")
+            for cam in config['camera'].keys():
+                self.add_camera(cam)
+
+        if version < 2:
+            # motion_failures
+            cursor.execute("""CREATE TABLE IF NOT EXISTS motion_failures (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                camera INTEGER NOT NULL,
+                filename TEXT NOT NULL
+            )""")
+
+        if version < 3:
+            cursor.execute("ALTER TABLE motion_failures ADD COLUMN message TEXT NOT NULL DEFAULT ''")
+
+        self.commit()
+
+    def add_camera(self, camera: int):
+        self.cursor().execute("INSERT INTO timestamps (camera, fix_time, motion_time) VALUES (?, ?, ?)",
+                              (camera, 0, 0))
+        self.commit()
+
+    def add_motion_failure(self,
+                           camera: int,
+                           filename: str,
+                           message: Optional[str]):
+        self.cursor().execute("INSERT INTO motion_failures (camera, filename, message) VALUES (?, ?, ?)",
+                              (camera, filename, message or ''))
+        self.commit()
+
+    def get_all_timestamps(self):
+        cur = self.cursor()
+        data = {}
+
+        cur.execute("SELECT camera, fix_time, motion_time FROM timestamps")
+        for cam, fix_time, motion_time in cur.fetchall():
+            data[int(cam)] = {
+                'fix': int(fix_time),
+                'motion': int(motion_time)
+            }
+
+        return data
+
+    def set_timestamp(self,
+                      camera: int,
+                      time_type: TimeFilterType,
+                      time: Union[int, datetime]):
+        cur = self.cursor()
+        if isinstance(time, datetime):
+            time = int(time.timestamp())
+        cur.execute(f"UPDATE timestamps SET {time_type.value}_time=? WHERE camera=?", (time, camera))
+        self.commit()
+
+    def get_timestamp(self,
+                      camera: int,
+                      time_type: TimeFilterType) -> int:
+        cur = self.cursor()
+        cur.execute(f"SELECT {time_type.value}_time FROM timestamps WHERE camera=?", (camera,))
+        return int(cur.fetchone()[0])
+
+
+# ipcam web api
+# -------------
+
+class IPCamWebServer(http.HTTPServer):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self.get('/api/recordings/{name}', self.get_camera_recordings)
+        self.get('/api/recordings/{name}/download/{file}', self.download_recording)
+        self.get('/api/camera/list', self.camlist)
+        self.get('/api/timestamp/{name}/{type}', self.get_timestamp)
+        self.get('/api/timestamp/all', self.get_all_timestamps)
+
+        self.post('/api/debug/migrate-mtimes', self.debug_migrate_mtimes)
+        self.post('/api/debug/fix', self.debug_fix)
+        self.post('/api/timestamp/{name}/{type}', self.set_timestamp)
+
+        self.post('/api/motion/done/{name}', self.submit_motion)
+        self.post('/api/motion/fail/{name}', self.submit_motion_failure)
+
+    async def get_camera_recordings(self, req):
+        cam = int(req.match_info['name'])
+        try:
+            filter = TimeFilterType(req.query['filter'])
+        except KeyError:
+            filter = None
+
+        files = get_recordings_files(cam, filter)
+
+        return self.ok({'files': files})
+
+    async def download_recording(self, req: http.Request):
+        cam = int(req.match_info['name'])
+        file = req.match_info['file']
+
+        fullpath = os.path.join(config['camera'][cam]['recordings_path'], file)
+        if not os.path.isfile(fullpath):
+            raise ValueError(f'file "{fullpath}" does not exists')
+
+        return http.FileResponse(fullpath)
+
+    async def camlist(self, req: http.Request):
+        return self.ok(config['camera'])
+
+    async def submit_motion(self, req: http.Request):
+        data = await req.post()
+
+        camera = int(req.match_info['name'])
+        timecodes = data['timecodes']
+        filename = data['filename']
+
+        time = filename_to_datetime(filename)
+
+        try:
+            if timecodes != '':
+                fragments = camutil.dvr_scan_timecodes(timecodes)
+                asyncio.ensure_future(process_fragments(camera, filename, fragments))
+
+            db.set_timestamp(camera, TimeFilterType.MOTION, time)
+            return self.ok()
+
+        except ValueError as e:
+            db.set_timestamp(camera, TimeFilterType.MOTION, time)
+            raise e
+
+    async def submit_motion_failure(self, req: http.Request):
+        camera = int(req.match_info['name'])
+
+        data = await req.post()
+        filename = data['filename']
+        message = data['message']
+
+        db.add_motion_failure(camera, filename, message)
+        db.set_timestamp(camera, TimeFilterType.MOTION, filename_to_datetime(filename))
+
+        return self.ok()
+
+    async def debug_migrate_mtimes(self, req: http.Request):
+        written = {}
+        for cam in config['camera'].keys():
+            confdir = os.path.join(os.getenv('HOME'), '.config', f'video-util-{cam}')
+            for time_type in TimeFilterType:
+                txt_file = os.path.join(confdir, f'{time_type.value}_mtime')
+                if os.path.isfile(txt_file):
+                    with open(txt_file, 'r') as fd:
+                        data = fd.read()
+                        db.set_timestamp(cam, time_type, int(data.strip()))
+
+                        if cam not in written:
+                            written[cam] = []
+                        written[cam].append(time_type)
+
+        return self.ok({'written': written})
+
+    async def debug_fix(self, req: http.Request):
+        asyncio.ensure_future(fix_job())
+        return self.ok()
+
+    async def set_timestamp(self, req: http.Request):
+        cam, time_type, time = self._getset_timestamp_params(req, need_time=True)
+        db.set_timestamp(cam, time_type, time)
+        return self.ok()
+
+    async def get_timestamp(self, req: http.Request):
+        cam, time_type = self._getset_timestamp_params(req)
+        return self.ok(db.get_timestamp(cam, time_type))
+
+    async def get_all_timestamps(self, req: http.Request):
+        return self.ok(db.get_all_timestamps())
+
+    @staticmethod
+    def _getset_timestamp_params(req: http.Request, need_time=False):
+        values = []
+
+        cam = int(req.match_info['name'])
+        assert cam in config['camera'], 'invalid camera'
+
+        values.append(cam)
+        values.append(TimeFilterType(req.match_info['type']))
+
+        if need_time:
+            time = req.query['time']
+            if time.startswith('record_'):
+                time = filename_to_datetime(time)
+            elif time.isnumeric():
+                time = int(time)
+            else:
+                raise ValueError('invalid time')
+            values.append(time)
+
+        return values
+
+
+# other global stuff
+# ------------------
+
+def open_database():
+    global db
+    db = IPCamServerDatabase()
+
+    # update cams list in database, if needed
+    cams = db.get_all_timestamps().keys()
+    for cam in config['camera']:
+        if cam not in cams:
+            db.add_camera(cam)
+
+
+def get_recordings_path(cam: int) -> str:
+    return config['camera'][cam]['recordings_path']
+
+
+def get_motion_path(cam: int) -> str:
+    return config['camera'][cam]['motion_path']
+
+
+def get_recordings_files(cam: int,
+                         time_filter_type: Optional[TimeFilterType] = None) -> list[dict]:
+    from_time = 0
+    to_time = int(time.time())
+
+    if time_filter_type:
+        from_time = db.get_timestamp(cam, time_filter_type)
+        if time_filter_type == TimeFilterType.MOTION:
+            to_time = db.get_timestamp(cam, TimeFilterType.FIX)
+
+    from_time = datetime.fromtimestamp(from_time)
+    to_time = datetime.fromtimestamp(to_time)
+
+    recdir = get_recordings_path(cam)
+    files = [{
+        'name': file,
+        'size': os.path.getsize(os.path.join(recdir, file))}
+             for file in os.listdir(recdir)
+             if valid_recording_name(file) and from_time < filename_to_datetime(file) <= to_time]
+    files.sort(key=lambda file: file['name'])
+
+    if files:
+        last = files[len(files)-1]
+        fullpath = os.path.join(recdir, last['name'])
+        if camutil.has_handle(fullpath):
+            logger.debug(f'get_recordings_files: file {fullpath} has opened handle, ignoring it')
+            files.pop()
+
+    return files
+
+
+async def process_fragments(camera: int,
+                            filename: str,
+                            fragments: list[tuple[int, int]]) -> None:
+    time = filename_to_datetime(filename)
+
+    rec_dir = get_recordings_path(camera)
+    motion_dir = get_motion_path(camera)
+    if not os.path.exists(motion_dir):
+        os.mkdir(motion_dir)
+
+    for fragment in fragments:
+        start, end = fragment
+
+        start -= config['motion']['padding']
+        end += config['motion']['padding']
+
+        if start < 0:
+            start = 0
+
+        duration = end - start
+
+        dt1 = (time + timedelta(seconds=start)).strftime(datetime_format)
+        dt2 = (time + timedelta(seconds=end)).strftime(datetime_format)
+
+        await camutil.ffmpeg_cut(input=os.path.join(rec_dir, filename),
+                                 output=os.path.join(motion_dir, f'{dt1}__{dt2}.mp4'),
+                                 start_pos=start,
+                                 duration=duration)
+
+
+async def fix_job() -> None:
+    logger.debug('fix_job: starting')
+
+    for cam in config['camera'].keys():
+        files = get_recordings_files(cam, TimeFilterType.FIX)
+        if not files:
+            logger.debug(f'fix_job: no files for camera {cam}')
+            continue
+
+        logger.debug(f'fix_job: got %d files for camera {cam}' % (len(files),))
+
+        for file in files:
+            fullpath = os.path.join(get_recordings_path(cam), file['name'])
+            await camutil.ffmpeg_recreate(fullpath)
+            timestamp = filename_to_datetime(file['name'])
+            if timestamp:
+                db.set_timestamp(cam, TimeFilterType.FIX, timestamp)
+
+
+datetime_format = '%Y-%m-%d-%H.%M.%S'
+db: Optional[IPCamServerDatabase] = None
+server: Optional[IPCamWebServer] = None
+logger = logging.getLogger(__name__)
+
+
+# start of the program
+# --------------------
+
+if __name__ == '__main__':
+    config.load('ipcam_server')
+
+    open_database()
+
+    loop = asyncio.get_event_loop()
+
+    scheduler = AsyncIOScheduler(event_loop=loop)
+    scheduler.add_job(fix_job, 'interval', seconds=config['fix_interval'])
+    scheduler.start()
+
+    server = IPCamWebServer(parse_addr(config['server']['listen']))
+    server.run()