Merge branch 'master' of ch1p.io:homekit

1 files changed, 544 insertions, 0 deletions

diff --git a/bin/ipcam_server.py b/bin/ipcam_server.py
new file mode 100755
index 0000000..71d5ea1
--- /dev/null
+++ b/bin/ipcam_server.py
@@ -0,0 +1,544 @@
+#!/usr/bin/env python3
+import logging
+import os
+import asyncio
+import time
+import shutil
+import __py_include
+
+import homekit.telegram.aio as telegram
+
+from socket import gethostname
+from argparse import ArgumentParser
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from asyncio import Lock
+
+from homekit.config import config as homekit_config, LinuxBoardsConfig
+from homekit.util import Addr
+from homekit import http
+from homekit.database.sqlite import SQLiteBase
+from homekit.camera import util as camutil, IpcamConfig
+from homekit.camera.types import (
+    TimeFilterType,
+    TelegramLinkType,
+    VideoContainerType
+)
+from homekit.camera.util import (
+    get_recordings_path,
+    get_motion_path,
+    is_valid_recording_name,
+    datetime_from_filename
+)
+
+from typing import Optional, Union, List, Tuple
+from datetime import datetime, timedelta
+from functools import cmp_to_key
+
+
+ipcam_config = IpcamConfig()
+lbc_config = LinuxBoardsConfig()
+
+
+# ipcam database
+# --------------
+
+class IpcamServerDatabase(SQLiteBase):
+    SCHEMA = 4
+
+    def __init__(self, path=None):
+        super().__init__(path=path)
+
+    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 ipcam_config.get_all_cam_names_for_this_server():
+                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 ''")
+
+        if version < 4:
+            cursor.execute("ALTER TABLE timestamps ADD COLUMN motion_start_time INTEGER NOT NULL DEFAULT 0")
+            cursor.execute("UPDATE timestamps SET motion_start_time=motion_time")
+
+        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, motion_start_time FROM timestamps")
+        for cam, fix_time, motion_time, motion_start_time in cur.fetchall():
+            data[int(cam)] = {
+                'fix': int(fix_time),
+                'motion': int(motion_time),
+                'motion_start': int(motion_start_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', self.get_motion_queue)
+        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/fix', self.debug_fix)
+        self.post('/api/debug/cleanup', self.debug_cleanup)
+
+        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)
+
+        # self.get('/api/motion/params/{name}', self.get_motion_params)
+        # self.get('/api/motion/params/{name}/roi', self.get_motion_roi_params)
+
+        self.queue_lock = Lock()
+
+    async def get_camera_recordings(self, req):
+        camera = int(req.match_info['name'])
+        try:
+            filter = TimeFilterType(req.query['filter'])
+        except KeyError:
+            filter = None
+
+        try:
+            limit = int(req.query['limit'])
+        except KeyError:
+            limit = 0
+
+        files = get_recordings_files(camera, filter, limit)
+        if files:
+            time = datetime_from_filename(files[len(files)-1]['name'])
+            db.set_timestamp(camera, TimeFilterType.MOTION_START, time)
+        return self.ok({'files': files})
+
+    async def get_motion_queue(self, req):
+        try:
+            limit = int(req.query['limit'])
+        except KeyError:
+            limit = 0
+
+        async with self.queue_lock:
+            files = get_recordings_files(None, TimeFilterType.MOTION_START, limit)
+            if files:
+                times_by_cam = {}
+                for file in files:
+                    time = datetime_from_filename(file['name'])
+                    if file['cam'] not in times_by_cam or times_by_cam[file['cam']] < time:
+                        times_by_cam[file['cam']] = time
+                for cam, time in times_by_cam.items():
+                    db.set_timestamp(cam, TimeFilterType.MOTION_START, time)
+
+        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(get_recordings_path(cam), 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(ipcam_config.get_all_cam_names_for_this_server())
+
+    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 = datetime_from_filename(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 camutil.DVRScanInvalidTimecodes as e:
+            db.add_motion_failure(camera, filename, str(e))
+            db.set_timestamp(camera, TimeFilterType.MOTION, time)
+            return self.ok('invalid timecodes')
+
+    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, datetime_from_filename(filename))
+
+        return self.ok()
+
+    async def debug_fix(self, req: http.Request):
+        asyncio.ensure_future(fix_job())
+        return self.ok()
+
+    async def debug_cleanup(self, req: http.Request):
+        asyncio.ensure_future(cleanup_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())
+
+    # async def get_motion_params(self, req: http.Request):
+    #     data = config['motion_params'][int(req.match_info['name'])]
+    #     lines = [
+    #         f'threshold={data["threshold"]}',
+    #         f'min_event_length=3s',
+    #         f'frame_skip=2',
+    #         f'downscale_factor=3',
+    #     ]
+    #     return self.plain('\n'.join(lines)+'\n')
+    #
+    # async def get_motion_roi_params(self, req: http.Request):
+    #     data = config['motion_params'][int(req.match_info['name'])]
+    #     return self.plain('\n'.join(data['roi'])+'\n')
+
+    @staticmethod
+    def _getset_timestamp_params(req: http.Request, need_time=False):
+        values = []
+
+        cam = int(req.match_info['name'])
+        assert cam in ipcam_config.get_all_cam_names_for_this_server(), 'invalid camera'
+
+        values.append(cam)
+        values.append(TimeFilterType(req.match_info['type']))
+
+        if need_time:
+            time = req.query['time']
+            if time.startswith('record_'):
+                time = datetime_from_filename(time)
+            elif time.isnumeric():
+                time = int(time)
+            else:
+                raise ValueError('invalid time')
+            values.append(time)
+
+        return values
+
+
+# other global stuff
+# ------------------
+
+def open_database(database_path: str):
+    global db
+    db = IpcamServerDatabase(database_path)
+
+    # update cams list in database, if needed
+    stored_cams = db.get_all_timestamps().keys()
+    for cam in ipcam_config.get_all_cam_names_for_this_server():
+        if cam not in stored_cams:
+            db.add_camera(cam)
+
+
+def get_recordings_files(cam: Optional[int] = None,
+                         time_filter_type: Optional[TimeFilterType] = None,
+                         limit=0) -> List[dict]:
+    from_time = 0
+    to_time = int(time.time())
+
+    cams = [cam] if cam is not None else ipcam_config.get_all_cam_names_for_this_server()
+    files = []
+    for cam in cams:
+        if time_filter_type:
+            from_time = db.get_timestamp(cam, time_filter_type)
+            if time_filter_type in (TimeFilterType.MOTION, TimeFilterType.MOTION_START):
+                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)
+        cam_files = [{
+            'cam': cam,
+            'name': file,
+            'size': os.path.getsize(os.path.join(recdir, file))}
+                 for file in os.listdir(recdir)
+                 if is_valid_recording_name(file) and from_time < datetime_from_filename(file) <= to_time]
+        cam_files.sort(key=lambda file: file['name'])
+
+        if cam_files:
+            last = cam_files[len(cam_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')
+                cam_files.pop()
+            files.extend(cam_files)
+
+    if limit > 0:
+        files = files[:limit]
+
+    return files
+
+
+async def process_fragments(camera: int,
+                            filename: str,
+                            fragments: List[Tuple[int, int]]) -> None:
+    time = datetime_from_filename(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 -= ipcam_config['motion_padding']
+        end += ipcam_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)
+
+    if fragments and ipcam_config['motion_telegram']:
+        asyncio.ensure_future(motion_notify_tg(camera, filename, fragments))
+
+
+async def motion_notify_tg(camera: int,
+                           filename: str,
+                           fragments: List[Tuple[int, int]]):
+    dt_file = datetime_from_filename(filename)
+    fmt = '%H:%M:%S'
+
+    text = f'Camera: <b>{camera}</b>\n'
+    text += f'Original file: <b>{filename}</b> '
+    text += _tg_links(TelegramLinkType.ORIGINAL_FILE, camera, filename)
+
+    for start, end in fragments:
+        start -= ipcam_config['motion_padding']
+        end += ipcam_config['motion_padding']
+
+        if start < 0:
+            start = 0
+
+        duration = end - start
+        if duration < 0:
+            duration = 0
+
+        dt1 = dt_file + timedelta(seconds=start)
+        dt2 = dt_file + timedelta(seconds=end)
+
+        text += f'\nFragment: <b>{duration}s</b>, {dt1.strftime(fmt)}-{dt2.strftime(fmt)} '
+        text += _tg_links(TelegramLinkType.FRAGMENT, camera, f'{dt1.strftime(datetime_format)}__{dt2.strftime(datetime_format)}.mp4')
+
+    await telegram.send_message(text)
+
+
+def _tg_links(link_type: TelegramLinkType,
+              camera: int,
+              file: str) -> str:
+    links = []
+    for link_name, link_template in ipcam_config[f'{link_type.value}_url_templates']:
+        link = link_template.replace('{camera}', str(camera)).replace('{file}', file)
+        links.append(f'<a href="{link}">{link_name}</a>')
+    return ' '.join(links)
+
+
+async def fix_job() -> None:
+    global fix_job_running
+    logger.debug('fix_job: starting')
+
+    if fix_job_running:
+        logger.error('fix_job: already running')
+        return
+
+    try:
+        fix_job_running = True
+        for cam in ipcam_config.get_all_cam_names_for_this_server():
+            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 = datetime_from_filename(file['name'])
+                if timestamp:
+                    db.set_timestamp(cam, TimeFilterType.FIX, timestamp)
+
+    finally:
+        fix_job_running = False
+
+
+async def cleanup_job() -> None:
+    def compare(i1: str, i2: str) -> int:
+        dt1 = datetime_from_filename(i1)
+        dt2 = datetime_from_filename(i2)
+
+        if dt1 < dt2:
+            return -1
+        elif dt1 > dt2:
+            return 1
+        else:
+            return 0
+
+    global cleanup_job_running
+    logger.debug('cleanup_job: starting')
+
+    if cleanup_job_running:
+        logger.error('cleanup_job: already running')
+        return
+
+    try:
+        cleanup_job_running = True
+
+        gb = float(1 << 30)
+        disk_number = 0
+        for storage in lbc_config.get_board_disks(gethostname()):
+            disk_number += 1
+            if os.path.exists(storage['mountpoint']):
+                total, used, free = shutil.disk_usage(storage['mountpoint'])
+                free_gb = free // gb
+                if free_gb < ipcam_config['cleanup_min_gb']:
+                    cleaned = 0
+                    files = []
+                    for cam in ipcam_config.get_all_cam_names_for_this_server(filter_by_disk=disk_number):
+                        for _dir in (get_recordings_path(cam), get_motion_path(cam)):
+                            files += list(map(lambda file: os.path.join(_dir, file), os.listdir(_dir)))
+                        files = list(filter(lambda path: os.path.isfile(path) and path.endswith(tuple([f'.{t.value}' for t in VideoContainerType])), files))
+                        files.sort(key=cmp_to_key(compare))
+
+                    for file in files:
+                        size = os.stat(file).st_size
+                        try:
+                            os.unlink(file)
+                            cleaned += size
+                        except OSError as e:
+                            logger.exception(e)
+                        if (free + cleaned) // gb >= ipcam_config['cleanup_min_gb']:
+                            break
+            else:
+                logger.error(f"cleanup_job: {storage['mountpoint']} not found")
+    finally:
+        cleanup_job_running = False
+
+
+fix_job_running = False
+cleanup_job_running = False
+
+datetime_format = '%Y-%m-%d-%H.%M.%S'
+datetime_format_re = r'\d{4}-\d{2}-\d{2}-\d{2}\.\d{2}.\d{2}'
+db: Optional[IpcamServerDatabase] = None
+server: Optional[IpcamWebServer] = None
+logger = logging.getLogger(__name__)
+
+
+# start of the program
+# --------------------
+
+if __name__ == '__main__':
+    parser = ArgumentParser()
+    parser.add_argument('--listen', type=str, required=True)
+    parser.add_argument('--database-path', type=str, required=True)
+    arg = homekit_config.load_app(no_config=True, parser=parser)
+
+    open_database(arg.database_path)
+
+    loop = asyncio.get_event_loop()
+
+    try:
+        scheduler = AsyncIOScheduler(event_loop=loop)
+        if ipcam_config['fix_enabled']:
+            scheduler.add_job(fix_job, 'interval',
+                              seconds=ipcam_config['fix_interval'],
+                              misfire_grace_time=None)
+
+        scheduler.add_job(cleanup_job, 'interval',
+                          seconds=ipcam_config['cleanup_interval'],
+                          misfire_grace_time=None)
+        scheduler.start()
+    except KeyError:
+        pass
+
+    asyncio.ensure_future(fix_job())
+    asyncio.ensure_future(cleanup_job())
+
+    server = IpcamWebServer(Addr.fromstring(arg.listen))
+    server.run()