Fix yaml

This reverts commit 077bdcffd4.

.drone.yml

@@ -0,0 +1,21 @@
+---
+kind: pipeline
+type: kubernetes
+name: default
+
+steps:
+- name: lint
+  image: python
+  commands:
+    - pip install pre_commit
+    - pre-commit run --all-files
+- name: docker
+  image: plugins/docker
+  settings:
+    repo: harbor.k-space.ee/${DRONE_REPO}
+    registry: harbor.k-space.ee
+    mtu: 1300
+    username:
+      from_secret: docker_username
+    password:
+      from_secret: docker_password

.flake8

@@ -0,0 +1,3 @@
+[flake8]
+inline-quotes = "
+indent-size = 4

.gitignore

@@ -1,2 +1 @@
 .env
-*.save

.gitmodules

@@ -1,4 +0,0 @@
-[submodule "log-viewer"]
-	path = log-viewer
-	url = https://git.k-space.ee/k-space/log-viewer.git
-	branch = wip

.woodpecker/build.yaml

@@ -1,26 +0,0 @@
----
-matrix:
-  ARCH:
-    - amd64
-    - arm64
-
-steps:
-  - name: build
-    image: woodpeckerci/plugin-kaniko
-    backend_options:
-      kubernetes:
-        nodeSelector:
-          kubernetes.io/arch: ${ARCH}
-        tolerations:
-          - key: arch
-            operator: Equal
-            value: ${ARCH}
-            effect: 'NoSchedule'
-    settings:
-      repo: ${CI_REPO}
-      registry: harbor.k-space.ee
-      tags: latest-${ARCH}
-      username:
-        from_secret: docker_username
-      password:
-        from_secret: docker_password

.woodpecker/manifest.yaml

@@ -1,31 +0,0 @@
----
-skip_clone: true
-
-steps:
-  - name: manifest
-    image: mirror.gcr.io/mplatform/manifest-tool:alpine-v2.1.6
-    secrets:
-      - docker_username
-      - docker_password
-    commands:
-      - set -u
-      - ls -lash
-      - env
-      - |
-        cat << EOF > spec.yaml
-        image: "harbor.k-space.ee/${CI_REPO}:latest"
-        manifests:
-          - image: "harbor.k-space.ee/${CI_REPO}:latest-amd64"
-            platform:
-              architecture: amd64
-              os: linux
-          - image: "harbor.k-space.ee/${CI_REPO}:latest-arm64"
-            platform:
-              architecture: arm64
-              os: linux
-        EOF
-      - /manifest-tool --username $docker_username --password $docker_password push from-spec spec.yaml > stdout
-      - cat stdout
-
-depends_on:
-  - build

Dockerfile

@@ -1,31 +1,21 @@
-FROM mirror.gcr.io/library/ubuntu:focal
+FROM ubuntu
 WORKDIR /app
 ENV DEBIAN_FRONTEND=noninteractive
-RUN apt-get update \
- && apt-get install --no-install-recommends -y \
-    build-essential \
+RUN apt-get update && apt-get install -y \
+    gstreamer1.0-libav \
+    gstreamer1.0-plugins-bad \
+    gstreamer1.0-plugins-base \
+    gstreamer1.0-plugins-good \
+    gstreamer1.0-plugins-ugly \
+    gstreamer1.0-tools \
     libjpeg-dev \
-    libpython3-dev \
     python3-gevent \
     python3-numpy \
     python3-opencv \
+    python3-flask \
     python3-pip \
-    git \
- && pip3 install \
-    aioboto3 \
-    aiohttp \
-    jpeg2dct \
-    motor \
-    prometheus_client \
-    sanic==21.6.2 \
-    sanic-json-logging \
-    git+https://github.com/Assarius/sanic-prometheus@Sanic_22 \
- && apt-get remove -y \
-    build-essential \
-    libjpeg-dev \
-    libpython3-dev \
- && apt-get autoremove -y \
  && apt-get clean
+RUN pip3 install boto3 prometheus_client pymongo==3.12.2 aiohttp jpeg2dct sanic==21.6.2 sanic_prometheus motor
 COPY camdetect.py /app
 ENTRYPOINT /app/camdetect.py
 EXPOSE 5000

README.md

@@ -8,26 +8,8 @@ In a nutshell:
 - It brings the MJPEG stream into the cluster
 - Performs highly optimal JPEG DCT coefficient based motion detection
   without actually decoding the JPEG frame to a bitmap
-- Writes events to MongoDB
-- Generates thumbnails based on JPEG DCT coefficents
-- Uploads screenshots and corresponding thumbnails to S3
+- WIP: Writes events to MongoDB
+- WIP: Uploads screenshots to S3
 - Exposes endpoint for distributing MJPEG stream inside the cluster,
-  eg for the `camera-tiler`
+  eg by the `camera-tiler`
 - Exposes endpoint for inspecting DCT blocks where motion has been detected
-
-# Developing
-
-Bundled `docker-compose.yml` brings up:
-
-* [camdetect bypass stream](http://localhost:5000/bypass)
-* [camdetect debug](http://localhost:5000/debug)
-* [Minio](http://localhost:9001/buckets/camdetect/browse)
-* [Mongoexpress](http://localhost:8081/db/default/eventlog)
-* [Prometheus](http://localhost:9090/graph)
-* [mjpeg-streamer](http://user:123456@localhost:8080/?action=stream)
-
-To manually trigger event:
-
-```
-docker kill -sUSR1 camera-motion-detect_camdetect_1
-```

camdetect.py

@@ -1,390 +1,86 @@
 #!/usr/bin/env python3
-import aioboto3
 import aiohttp
 import asyncio
 import cv2
-import hashlib
-import io
-import json
 import numpy as np
 import os
-import pymongo
-import signal
+import json
+import socket
 import sys
-import botocore.exceptions
-from datetime import datetime, timedelta
+from datetime import datetime
 from jpeg2dct.numpy import loads
-from math import inf
-from motor.motor_asyncio import AsyncIOMotorClient
-from prometheus_client import Counter, Gauge, Histogram
-from pymongo import ReturnDocument
+from prometheus_client import Counter, Gauge
 from sanic import Sanic, response
-from sanic_json_logging import setup_json_logging
-from sanic.log import logger
-from sanic_prometheus import monitor
 from sanic.response import stream
+from sanic_prometheus import monitor
 from time import time
-from urllib.parse import urlparse
 
-_, target = sys.argv
+_, url = sys.argv
 
-# Override basic auth password from env var
-basic_auth_password = os.getenv("BASIC_AUTH_PASSWORD")
-if basic_auth_password:
-    o = urlparse(target)
-    netloc = o.netloc
-    username = ""
-    if "@" in netloc:
-        username, netloc = o.netloc.split("@", 1)
-        if ":" in username:
-            username, _ = username.split(":")
-    target = o._replace(netloc="%s:%s@%s" % (username, basic_auth_password, netloc)).geturl()
-
-AWS_ACCESS_KEY_ID = os.environ["AWS_ACCESS_KEY_ID"]
-AWS_SECRET_ACCESS_KEY = os.environ["AWS_SECRET_ACCESS_KEY"]
-
-S3_ENDPOINT_URL = os.environ["S3_ENDPOINT_URL"]
-S3_BUCKET_NAME = os.getenv("S3_BUCKET_NAME", "camdetect")
 MONGO_URI = os.getenv("MONGO_URI", "mongodb://127.0.0.1:27017/default")
-MONGO_COLLECTION = os.getenv("MONGO_COLLECTION", "eventlog")
-SOURCE_NAME = os.environ["SOURCE_NAME"]
-SLIDE_WINDOW = 4
+FQDN = socket.getfqdn()
+SLIDE_WINDOW = 2
 DCT_BLOCK_SIZE = 8
-UPLOAD_FRAMESKIP = 3
-CLOCK_SKEW_TOLERANCE = timedelta(seconds=3)
-
-# Percentage of blocks active to consider movement in whole frame
-THRESHOLD_RATIO = int(os.getenv("THRESHOLD_RATIO", "5"))
-
-# Days to keep events
-TTL_DAYS = int(os.getenv("TTL_DAYS", "3"))
 
+# How many blocks have changes to consider movement in frame
+THRESHOLD_BLOCKS = 20
+THRESHOLD_MOTION_START = 2
 CHUNK_BOUNDARY = b"\n--frame\nContent-Type: image/jpeg\n\n"
 
-hist_active_blocks_ratio = Histogram(
-    "camtiler_active_blocks_ratio",
-    "Ratio of active DCT blocks",
-    ["roi"],
-    buckets=(0.01, 0.02, 0.05, 0.1, 0.5, inf))
-hist_processing_latency = Histogram(
-    "camtiler_frame_processing_latency_seconds",
-    "Frame processing latency",
-    buckets=(0.01, 0.05, 0.1, 0.5, 1, inf))
-hist_upload_latency = Histogram(
-    "camtiler_frame_upload_latency_seconds",
-    "Frame processing latency",
-    buckets=(0.1, 0.5, 1, 5, 10, inf))
-counter_events = Counter(
-    "camtiler_events",
-    "Count of successfully processed events")
-counter_frames = Counter(
-    "camtiler_frames",
-    "Count of frames",
-    ["stage"])
-counter_dropped_frames = Counter(
-    "camtiler_dropped_frames",
-    "Frames that were dropped due to one of queues being full",
-    ["stage"])
-counter_discarded_bytes = Counter(
-    "camtiler_discarded_bytes",
+counter_dropped_bytes = Counter(
+    "camdetect_dropped_bytes",
     "Bytes that were not not handled or part of actual JPEG frames")
-counter_receive_bytes = Counter(
-    "counter_receive_bytes",
+counter_rx_bytes = Counter(
+    "camdetect_rx_bytes",
     "Bytes received over HTTP stream")
-counter_transmit_bytes = Counter(
-    "camtiler_transmit_bytes",
+counter_tx_bytes = Counter(
+    "camdetect_tx_bytes",
     "Bytes transmitted over HTTP streams")
-counter_receive_frames = Counter(
-    "camtiler_receive_frames",
-    "Frames received from upstream")
-counter_transmit_frames = Counter(
-    "camtiler_transmit_frames",
-    "Frames transmitted to downstream consumers")
-counter_emitted_events = Counter(
-    "camtiler_emitted_events",
+counter_rx_frames = Counter(
+    "camdetect_rx_frames",
+    "Frames received")
+counter_tx_frames = Counter(
+    "camdetect_tx_frames",
+    "Frames transmitted")
+counter_tx_events = Counter(
+    "camdetect_tx_events",
     "Events emitted")
-counter_receive_chunks = Counter(
-    "camtiler_receive_chunks",
+counter_rx_chunks = Counter(
+    "camdetect_rx_chunks",
     "HTTP chunks received")
 counter_errors = Counter(
-    "camtiler_errors",
+    "camdetect_errors",
     "Upstream connection errors",
-    ["stage", "exception"])
+    ["exception"])
+counter_movement_frames = Counter(
+    "camdetect_movement_frames",
+    "Frames with movement detected in them")
 gauge_last_frame = Gauge(
-    "camtiler_last_frame_timestamp_seconds",
-    "Timestamp of last frame",
-    ["stage"])
-gauge_queue_frames = Gauge(
-    "camtiler_queue_frames",
-    "Numer of frames in a queue",
-    ["stage"])
-gauge_build_info = Gauge(
-    "docker_build_info",
-    "Build info",
-    ["git_commit", "git_commit_timestamp"])
-
-gauge_build_info.labels(
-    os.getenv("GIT_COMMIT", "null"),
-    os.getenv("GIT_COMMIT_TIMESTAMP", "null")).set(1)
-
-# Reset some gauges
-gauge_queue_frames.labels("download").set(0)
-gauge_queue_frames.labels("hold").set(0)
-gauge_queue_frames.labels("upload").set(0)
-counter_frames.labels("motion").inc(0)
-counter_frames.labels("downloaded").inc(0)
-
-assert SLIDE_WINDOW <= 8  # This is 256 frames which should be enough
+    "camdetect_last_frame",
+    "Timestamp of last frame")
+gauge_total_blocks = Gauge(
+    "camdetect_total_blocks",
+    "Total DCT blocks")
+gauge_active_blocks = Gauge(
+    "camdetect_active_blocks",
+    "Total active, threshold exceeding DCT blocks")
 
 
-async def upload(bucket, blob: bytes, thumb: bytes, event_id):
-    """
-    Upload single frame to S3 bucket
-    """
-
-    # Generate S3 path based on the JPEG blob SHA512 digest
-    fp = hashlib.sha512(blob).hexdigest()
-    path = "%s/%s.jpg" % (fp[:4], fp[4:])
-
-    try:
-        await bucket.upload_fileobj(io.BytesIO(thumb), "thumb/%s" % path)
-        await bucket.upload_fileobj(io.BytesIO(blob), path)
-    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
-        j = "%s.%s" % (e.__class__.__module__, e.__class__.__name__)
-        counter_errors.labels("upload", j).inc()
-
-    # Add screenshot path to the event
-    app.ctx.coll.update_one({
-        "_id": event_id
-    }, {
-        "$addToSet": {
-            "screenshots": path,
-        }
-    })
-
-    counter_frames.labels("stored").inc()
-    now = datetime.utcnow()
-    gauge_last_frame.labels("upload").set(now.timestamp())
-    # TODO: Handle 16MB maximum document size
+class Frame(object):
+    def __init__(self, blob):
+        self.blob = blob
+        self.y, self.cb, self.cr = loads(blob)
+        self.mask = np.int16(self.y[:, :, 0])
 
 
-async def uploader(queue):
-    """
-    Uploader task grabs JPEG blobs from upload queue and uploads them to S3
-    """
-    session = aioboto3.Session()
-    async with session.resource("s3",
-                                aws_access_key_id=AWS_ACCESS_KEY_ID,
-                                aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
-                                endpoint_url=S3_ENDPOINT_URL) as s3:
-        bucket = await s3.Bucket(S3_BUCKET_NAME)
-        while True:
-            dt, blob, thumb, event_id = await queue.get()
-            gauge_queue_frames.labels("upload").set(queue.qsize())
-            await upload(bucket, blob, thumb, event_id)
-            counter_frames.labels("uploaded").inc()
-            hist_upload_latency.observe(
-                (datetime.utcnow() - dt).total_seconds())
-
-
-class ReferenceFrame():
-    """
-    ReferenceFrame keeps last 2 ^ size frames to infer the background scene
-    compared to which motion is detected
-
-    This is pretty much what background subtractor does in OpenCV,
-    only difference is that we want have better performance instead of
-    accuracy
-    """
-    class NotEnoughFrames(Exception):
-        pass
-
-    def __init__(self, size=SLIDE_WINDOW):
-        self.y = []
-        self.cumulative = None
-        self.size = size
-
-    def put(self, y):
-        if self.cumulative is None:
-            self.cumulative = np.copy(y)
-        else:
-            self.cumulative += y
-        self.y.append(y)
-        if len(self.y) > 2 ** self.size:
-            self.cumulative -= self.y[0]
-            self.y = self.y[1:]
-
-    def get(self):
-        if len(self.y) == 2 ** self.size:
-            return self.cumulative >> SLIDE_WINDOW
-        else:
-            raise self.NotEnoughFrames()
-
-
-async def motion_detector(reference_frame, download_queue, upload_queue):
-    """
-    Motion detector grabs JPEG blobs and Y channel coefficients
-    from download queue, performs motion detection and pushes relevant
-    JPEG blobs to upload queue going to S3
-    """
-    event_id = None
-    differing_blocks = []
-    uploads_skipped = 0
-
-    # Hold queue keeps frames that we have before motion event start timestamp
-    hold_queue = asyncio.Queue(2 ** SLIDE_WINDOW)
-
-    while True:
-        dt, blob, dct, thumb = await download_queue.get()
-        gauge_queue_frames.labels("download").set(download_queue.qsize())
-        app.ctx.last_frame, app.ctx.dct = blob, dct
-
-        # Signal /bypass and /debug handlers about new frame
-        app.ctx.event_frame.set()
-        app.ctx.event_frame.clear()
-
-        # Separate most significant luma value for each DCT (8x8 pixel) block
-        y = np.int16(dct[0][:, :, 0])
-
-        reference_frame.put(y)
-        try:
-            app.ctx.mask = cv2.inRange(cv2.absdiff(y,
-                                       reference_frame.get()), 50, 65535)
-        except ReferenceFrame.NotEnoughFrames:
-            app.ctx.mask = None
-            motion_detected = False
-        else:
-            # Implement dumb Kalman filter
-            active_blocks = np.count_nonzero(app.ctx.mask)
-            differing_blocks.append(active_blocks)
-            differing_blocks[:] = differing_blocks[-10:]
-            total_blocks = app.ctx.mask.shape[0] * app.ctx.mask.shape[1]
-            threshold_blocks = THRESHOLD_RATIO * total_blocks / 100
-            average_blocks = sum(differing_blocks) / len(differing_blocks)
-            hist_active_blocks_ratio.labels("main").observe(active_blocks / total_blocks)
-            motion_detected = average_blocks > threshold_blocks
-
-        now = datetime.utcnow()
-        gauge_last_frame.labels("processed").set(now.timestamp())
-        hist_processing_latency.observe((now - dt).total_seconds())
-
-        # Propagate SIGUSR1 signal handler
-        if app.ctx.manual_trigger:
-            logger.info("Manually triggering event via SIGUSR1")
-            motion_detected = True
-            app.ctx.manual_trigger = False
-
-        # Handle event start
-        if motion_detected and not event_id:
-            result = await app.ctx.coll.find_one_and_update({
-                "@timestamp": {
-                    "$lte": dt + CLOCK_SKEW_TOLERANCE,
-                    "$gte": dt - CLOCK_SKEW_TOLERANCE,
-                },
-                "source": SOURCE_NAME,
-            }, {
-                "$setOnInsert": {
-                    "@timestamp": dt,
-                    "source": SOURCE_NAME,
-                    "event": "motion-detected",
-                    "started": dt,
-                    "finished": dt + timedelta(minutes=2),
-                    "component": "camdetect",
-                    "screenshots": [],
-                    "action": "event",
-                }
-            }, upsert=True, return_document=ReturnDocument.AFTER)
-            app.ctx.event_id = event_id = result["_id"]
-
-        # Handle buffering frames prior event start
-        if hold_queue.full():
-            await hold_queue.get()
-        hold_queue.put_nowait((blob, thumb))
-        gauge_queue_frames.labels("hold").set(hold_queue.qsize())
-
-        # Handle image upload
-        if motion_detected and event_id:
-            counter_frames.labels("motion").inc()
-            while True:
-                if not uploads_skipped:
-                    uploads_skipped = UPLOAD_FRAMESKIP
-                else:
-                    uploads_skipped -= 1
-                    continue
-
-                # Drain queue of frames prior event start
-                try:
-                    blob, thumb = hold_queue.get_nowait()
-                except asyncio.QueueEmpty:
-                    break
-
-                try:
-                    # Push JPEG blob into upload queue
-                    upload_queue.put_nowait((dt, blob, thumb, event_id))
-                except asyncio.QueueFull:
-                    counter_dropped_frames.labels("upload").inc()
-                gauge_queue_frames.labels("upload").set(upload_queue.qsize())
-
-        # Handle event end
-        if not motion_detected and event_id:
-            app.ctx.coll.update_one({
-                "_id": event_id
-            }, {
-                "$set": {
-                    "finished": dt,
-                }
-            })
-            app.ctx.event_id = event_id = None
-            counter_events.inc()
-
-
-def generate_thumbnail(dct):
-    """
-    This is highly efficient and highly inaccurate function to generate
-    thumbnail based purely on JPEG coefficients
-    """
-    y, cr, cb = dct
-
-    # Determine aspect ratio and minimum dimension for cropping
-    ar = y.shape[0] < y.shape[1]
-    dm = (y.shape[0] if ar else y.shape[1]) & 0xfffffff8
-
-    # Determine cropping slices to make it square
-    jl = ((y.shape[1] >> 1) - (dm >> 1) if ar else 0)
-    jt = (0 if ar else (y.shape[0] >> 1) - (dm >> 1))
-    jr = jl + dm
-    jb = jt + dm
-
-    # Do the actual crop
-    ty = y[jt:jb, jl:jr, 0]
-    tb = cb[jt >> 1:jb >> 1, jl >> 1:jr >> 1, 0]
-    tr = cr[jt >> 1:jb >> 1, jl >> 1:jr >> 1, 0]
-
-    # Upsample chroma, dummy convert first coeff and stack all channels
-    m = np.dstack((
-        np.array((ty >> 3) + 127, dtype=np.uint8),
-        np.array(
-            (tb.repeat(2, 1).repeat(2, 0) >> 3) + 127, dtype=np.uint8)[:dm],
-        np.array(
-            (tr.repeat(2, 1).repeat(2, 0) >> 3) + 127, dtype=np.uint8)[:dm]))
-    _, jpeg = cv2.imencode(".jpg",
-                           cv2.cvtColor(m, cv2.COLOR_YCrCb2BGR),
-                           (cv2.IMWRITE_JPEG_QUALITY, 80))
-    return jpeg
-
-
-async def download(resp, queue):
-    """
-    This coroutine iterates over HTTP connection chunks
-    assembling the original JPEG blobs and decodes the
-    DCT coefficients of the frames
-    """
+async def client_connect(resp):
     buf = b""
-    logger.info("Upstream connection opened with status: %d", resp.status)
+    print("Upstream connection opened with status:", resp.status)
     async for data, end_of_http_chunk in resp.content.iter_chunks():
-        counter_receive_bytes.inc(len(data))
+        counter_rx_bytes.inc(len(data))
         if end_of_http_chunk:
-            counter_receive_chunks.inc()
+            counter_rx_chunks.inc()
+
         if buf:
             # seek end
             marker = data.find(b"\xff\xd9")
@@ -392,82 +88,100 @@ async def download(resp, queue):
                 buf += data
                 continue
             else:
-                # Assemble JPEG blob
-                blob = buf + data[:marker + 2]
+                app.ctx.last_frame = Frame(buf + data[:marker+2])
+                gauge_last_frame.set(time())
 
-                # Parse DCT coefficients
-                try:
-                    dct = loads(blob)
-                except RuntimeError:
-                    counter_frames.labels("corrupted").inc()
+                reference = app.ctx.last_frame.mask
+                app.ctx.frames.append(reference)
+                if app.ctx.avg is None:
+                    app.ctx.avg = np.copy(reference)
                 else:
-                    now = datetime.utcnow()
-                    gauge_last_frame.labels("download").set(now.timestamp())
-                    try:
-                        queue.put_nowait((
-                            now,
-                            blob,
-                            dct,
-                            generate_thumbnail(dct)))
-                    except asyncio.QueueFull:
-                        counter_dropped_frames.labels("download").inc()
-                    else:
-                        counter_frames.labels("downloaded").inc()
-                    gauge_queue_frames.labels("download").set(queue.qsize())
-                data = data[marker + 2:]
+                    app.ctx.avg += reference
+
+                if len(app.ctx.frames) > 2 ** SLIDE_WINDOW:
+                    app.ctx.avg -= app.ctx.frames[0]
+                    app.ctx.frames = app.ctx.frames[1:]
+
+                if len(app.ctx.frames) == 2 ** SLIDE_WINDOW:
+                    app.ctx.thresh = cv2.inRange(cv2.absdiff(
+                        app.ctx.last_frame.mask,
+                        app.ctx.avg >> SLIDE_WINDOW), 25, 65535)
+                else:
+                    app.ctx.thresh = None
+                gauge_total_blocks.set(app.ctx.last_frame.mask.shape[0] *
+                                       app.ctx.last_frame.mask.shape[1])
+
+                movement_detected = False
+                if app.ctx.thresh is not None:
+                    differing_blocks = np.count_nonzero(app.ctx.thresh)
+                    gauge_active_blocks.set(differing_blocks)
+                    if differing_blocks > THRESHOLD_BLOCKS:
+                        counter_movement_frames.inc()
+                        movement_detected = True
+
+                if movement_detected:
+                    if app.ctx.motion_frames < 30:
+                        app.ctx.motion_frames += 1
+                else:
+                    if app.ctx.motion_frames > 0:
+                        app.ctx.motion_frames -= 1
+
+                if app.ctx.motion_frames > 20:
+                    if not app.ctx.motion_start:
+                        app.ctx.motion_start = datetime.utcnow()
+                        print("Movement start")
+                elif app.ctx.motion_frames < 5:
+                    app.ctx.motion_start = None
+                    print("Movement end")
+
+                app.ctx.event_frame.set()
+                app.ctx.event_frame.clear()
+
+                data = data[marker+2:]
                 buf = b""
-                counter_receive_frames.inc()
+                counter_rx_frames.inc()
 
         # seek begin
-        marker = data.rfind(b"\xff\xd8")
+        marker = data.find(b"\xff\xd8")
         if marker >= 0:
             buf = data[marker:]
         else:
-            counter_discarded_bytes.inc(len(data))
+            counter_dropped_bytes.inc(len(data))
 
 
-async def downloader(queue: asyncio.Queue):
-    """
-    Downloader task connects to MJPEG source and
-    pushes the JPEG frames to download queue
-    """
+async def client():
     while True:
         to = aiohttp.ClientTimeout(connect=5, sock_read=2)
         async with aiohttp.ClientSession(timeout=to) as session:
-            logger.info("Opening connection to %s", target)
+            print("Opening upstream connection to %s" % url)
             try:
-                async with session.get(target) as resp:
-                    await download(resp, queue)
+                async with session.get(url) as resp:
+                    await client_connect(resp)
             except (aiohttp.ClientError, asyncio.exceptions.TimeoutError) as e:
                 j = "%s.%s" % (e.__class__.__module__, e.__class__.__name__)
-                logger.info("Caught exception %s", j)
-                counter_errors.labels("download", j).inc()
+                print("Caught exception %s" % j)
+                counter_errors.labels(exception=j).inc()
         await asyncio.sleep(1)
 
-app = Sanic("camdetect")
-setup_json_logging(app)
+
+app = Sanic("lease")
+app.config["WTF_CSRF_ENABLED"] = False
 
 
 @app.route("/bypass")
 async def bypass_stream_wrapper(request):
-    # Desired frame interval, by default 500ms
-    interval = float(request.args.get("interval", 500)) / 1000.0
-
     async def stream_camera(response):
-        ts = 0
         while True:
-            while True:
-                await app.ctx.event_frame.wait()
-                if time() > ts + interval:
-                    break
-            ts = time()
-            data = CHUNK_BOUNDARY + app.ctx.last_frame
+            await app.ctx.event_frame.wait()
+            data = CHUNK_BOUNDARY + app.ctx.last_frame.blob
             await response.write(data)
-            counter_transmit_bytes.inc(len(data))
-            counter_transmit_frames.inc()
+            counter_tx_bytes.inc(len(data))
+            counter_tx_frames.inc()
+
     return response.stream(
-        stream_camera,
-        content_type="multipart/x-mixed-replace; boundary=frame")
+                stream_camera,
+                content_type="multipart/x-mixed-replace; boundary=frame"
+            )
 
 
 @app.route("/debug")
@@ -475,30 +189,21 @@ async def stream_wrapper(request):
     async def stream_camera(response):
         while True:
             await app.ctx.event_frame.wait()
-
-            # Parse JPEG blob
-            arr = np.frombuffer(app.ctx.last_frame, dtype=np.uint8)
+            arr = np.frombuffer(app.ctx.last_frame.blob, dtype=np.uint8)
             img = cv2.imdecode(arr, cv2.IMREAD_UNCHANGED)
+            if len(app.ctx.frames) == 2 ** SLIDE_WINDOW:
+                for y in range(0, len(app.ctx.last_frame.mask)):
+                    for x in range(0, len(app.ctx.last_frame.mask[0])):
+                        if app.ctx.thresh[y][x] > 0:
+                            img[y*DCT_BLOCK_SIZE:(y+1)*DCT_BLOCK_SIZE,
+                                x*DCT_BLOCK_SIZE:(x+1)*DCT_BLOCK_SIZE, 2] = 255
 
-            # Highlight green or red channel depending on whether
-            # motion event is in progress or not
-            channel = 2 if app.ctx.event_id else 1
-            if app.ctx.mask is not None:
-                for y in range(0, app.ctx.mask.shape[0]):
-                    for x in range(0, app.ctx.mask.shape[1]):
-                        if app.ctx.mask[y][x] > 0:
-                            img[y * DCT_BLOCK_SIZE:(y + 1) * DCT_BLOCK_SIZE,
-                                x * DCT_BLOCK_SIZE:(x + 1) * DCT_BLOCK_SIZE,
-                                channel] = 255
-
-            # Compress modified frame as JPEG frame
             _, jpeg = cv2.imencode(".jpg", img, (cv2.IMWRITE_JPEG_QUALITY, 80))
             data = CHUNK_BOUNDARY + jpeg.tobytes()
             await response.write(data)
-            counter_transmit_bytes.inc(len(data))
-            counter_transmit_frames.inc()
+            counter_tx_bytes.inc(len(data))
+            counter_tx_frames.inc()
 
-    # Transmit as chunked MJPEG stream
     return response.stream(
         stream_camera,
         content_type="multipart/x-mixed-replace; boundary=frame"
@@ -507,32 +212,9 @@ async def stream_wrapper(request):
 
 @app.route("/readyz")
 async def ready_check(request):
-    logger.info("Testing if Mongo is accessible")
-    try:
-        async for i in app.ctx.coll.find().limit(1):
-            break
-    except pymongo.errors.ServerSelectionTimeoutError:
-        return response.text("MongoDB server selection timeout", status=503)
-
-    session = aioboto3.Session()
-    logger.info("Testing if S3 is writable")
-    async with session.resource("s3",
-                                aws_access_key_id=AWS_ACCESS_KEY_ID,
-                                aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
-                                endpoint_url=S3_ENDPOINT_URL) as s3:
-        bucket = await s3.Bucket(S3_BUCKET_NAME)
-        await bucket.upload_fileobj(io.BytesIO(b"test"), "test")
-    return response.text("OK")
-
-    print("Checking if there are any frames received")
-    if app.ctx.mask is None:
-        return response.text("Not enough frames", status=503)
-
-
-@app.route("/healthz")
-async def health_check(request):
-    if app.ctx.mask is None:
-        return response.text("Not enough frames", status=503)
+    if len(app.ctx.frames) == 2 ** SLIDE_WINDOW:
+        return response.text("OK")
+    return response.text("Not enough frames", status=503)
 
 
 @app.route("/event")
@@ -540,50 +222,28 @@ async def wrapper_stream_event(request):
     async def stream_event(response):
         while True:
             await app.ctx.event_frame.wait()
-            if app.ctx.mask is not None:
+            if len(app.ctx.frames) < 2 ** SLIDE_WINDOW:
                 continue
-            s = "data: " + json.dumps(app.ctx.mask.tolist()) + "\r\n\r\n"
+            s = "data: " + json.dumps(app.ctx.thresh.tolist()) + "\r\n\r\n"
             await response.write(s.encode())
-            counter_emitted_events.inc()
+            counter_tx_events.inc()
     return stream(stream_event, content_type="text/event-stream")
 
 
-def handler(signum, frame):
-    # SIGUSR1 handler for manually triggering an event
-    app.ctx.manual_trigger = True
-
-
 @app.listener("before_server_start")
 async def setup_db(app, loop):
-    app.ctx.mask = None
-    app.ctx.db = AsyncIOMotorClient(MONGO_URI).get_default_database()
-    app.ctx.coll = app.ctx.db[MONGO_COLLECTION]
-    app.ctx.coll.create_index(
-        "@timestamp", expireAfterSeconds=TTL_DAYS * 60 * 60 * 24)
-    app.ctx.coll.create_index([
-        ("source", pymongo.ASCENDING),
-        ("@timestamp", pymongo.ASCENDING)], unique=True)
     app.ctx.last_frame = None
     app.ctx.event_frame = asyncio.Event()
-    app.ctx.event_id = None
-    app.ctx.manual_trigger = False
-    signal.signal(signal.SIGUSR1, handler)
-
-    # Set up processing pipeline
-    download_queue = asyncio.Queue(50)
-    upload_queue = asyncio.Queue(50)
-    asyncio.create_task(uploader(
-        upload_queue))
-    asyncio.create_task(downloader(
-        download_queue))
-    asyncio.create_task(motion_detector(
-        ReferenceFrame(),
-        download_queue,
-        upload_queue))
-
+    app.ctx.frames = []
+    app.ctx.avg = None
+    app.ctx.motion_frames = 0
+    app.ctx.motion_start = None
+    app.ctx.motion_end = None
+    asyncio.create_task(client())
 
 monitor(app).expose_endpoint()
 
-app.run(host="0.0.0.0",
-        port=5000,
-        debug=bool(os.getenv("DEBUG", 0)))
+try:
+    app.run(host="0.0.0.0", port=5000)
+except KeyboardInterrupt:
+    asyncio.get_event_loop().close()

docker-compose.yml

@@ -1,20 +1,7 @@
 version: '3.7'
 
 # All keys here are for dev instance only, do not put prod keys here
-x-common: &common
-  AWS_ACCESS_KEY_ID: camdetect
-  MINIO_ROOT_USER: camdetect
-  AWS_SECRET_ACCESS_KEY: 2mSI6HdbJ8
-  MINIO_ROOT_PASSWORD: 2mSI6HdbJ8
-  ME_CONFIG_MONGODB_ENABLE_ADMIN: 'true'
-  ME_CONFIG_MONGODB_SERVER: '127.0.0.1'
-  ME_CONFIG_MONGODB_AUTH_DATABASE: admin
-  MINIO_DEFAULT_BUCKETS: camdetect
-  MINIO_URI: 'http://camdetect:2mSI6HdbJ8@127.0.0.1:9000/camdetect'
-  S3_ENDPOINT_URL: http://127.0.0.1:9000
-  MINIO_CONSOLE_PORT_NUMBER: 9001
-  MJPEGSTREAMER_CREDENTIALS: user:123456
-  SOURCE_NAME: dummy
+# To override and use inventory from prod use .env file
 
 services:
   camdetect:
@@ -24,13 +11,18 @@ services:
       context: .
     entrypoint: /app/camdetect.py
     command: http://user:123456@127.0.0.1:8080?action=stream
-    environment: *common
+    environment:
+      - MJPEGSTREAMER_CREDENTIALS=user:123456
+    env_file: .env
 
   mongoexpress:
     restart: always
     image: mongo-express
     network_mode: host
-    environment: *common
+    environment:
+      - ME_CONFIG_MONGODB_ENABLE_ADMIN=true
+      - ME_CONFIG_MONGODB_SERVER=127.0.0.1
+      - ME_CONFIG_MONGODB_AUTH_DATABASE=admin
     logging:
       driver: none
 
@@ -50,16 +42,16 @@ services:
       - --config.file=/config/prometheus.yml
     volumes:
       - ./config:/config:ro
-    logging:
-      driver: none
 
   minio:
     restart: always
     network_mode: host
     image: bitnami/minio:latest
-    environment: *common
-    logging:
-      driver: none
+    environment:
+      - MINIO_ACCESS_KEY=kspace-mugshot
+      - MINIO_SECRET_KEY=2mSI6HdbJ8
+      - MINIO_DEFAULT_BUCKETS=kspace-mugshot:download
+      - MINIO_CONSOLE_PORT_NUMBER=9001
 
   mjpg-streamer:
     network_mode: host
@@ -67,17 +59,4 @@ services:
     image: kvaps/mjpg-streamer
     devices:
       - /dev/video0
-    command: -i "/usr/lib64/input_uvc.so -y -d /dev/video0 -r 1280x720 -f 5" -o "output_http.so -c user:123456"
-
-  log-viewer-backend:
-    restart: always
-    network_mode: host
-    build:
-      context: ./log-viewer/backend
-    environment: *common
-
-  log-viewer-frontend:
-    restart: always
-    network_mode: host
-    build:
-      context: ./log-viewer/frontend
+    command: -i "/usr/lib64/input_uvc.so -y -d /dev/video0 -r 1280x720 -f 30" -o "output_http.so -c user:123456"

log-viewer

@@ -1 +0,0 @@
-Subproject commit b8e9f03b86ff63a5cdc77bae18caf7d7efd077b9

mongo-init.sh

@@ -1,4 +1,11 @@
 #!/bin/bash
 mongo <<EOF
-rs.initiate()
+
+rs.initiate({
+    _id: 'rs0',
+    members: [
+        {_id: 0, host: '127.0.0.1:27017'}
+    ]
+})
+
 EOF