You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
519 lines
17 KiB
519 lines
17 KiB
#!/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 sys |
|
from datetime import datetime, timedelta |
|
from jpeg2dct.numpy import loads |
|
from motor.motor_asyncio import AsyncIOMotorClient |
|
from prometheus_client import Counter, Gauge |
|
from sanic import Sanic, response |
|
from sanic.response import stream |
|
from sanic_prometheus import monitor |
|
from time import time |
|
|
|
_, url = sys.argv |
|
|
|
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID", "camdetect") |
|
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_COLLETION", "eventlog") |
|
SOURCE_NAME = os.environ["SOURCE_NAME"] |
|
SLIDE_WINDOW = 2 |
|
DCT_BLOCK_SIZE = 8 |
|
UPLOAD_FRAMESKIP = 3 |
|
|
|
# Percentage of blocks active to consider movement in whole frame |
|
THRESHOLD_RATIO = int(os.getenv("THRESHOLD_RATIO", "5")) |
|
|
|
CHUNK_BOUNDARY = b"\n--frame\nContent-Type: image/jpeg\n\n" |
|
|
|
counter_dropped_bytes = Counter( |
|
"camdetect_dropped_bytes", |
|
"Bytes that were not not handled or part of actual JPEG frames") |
|
counter_rx_bytes = Counter( |
|
"camdetect_rx_bytes", |
|
"Bytes received over HTTP stream") |
|
counter_tx_bytes = Counter( |
|
"camdetect_tx_bytes", |
|
"Bytes transmitted over HTTP streams") |
|
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_rx_chunks = Counter( |
|
"camdetect_rx_chunks", |
|
"HTTP chunks received") |
|
counter_errors = Counter( |
|
"camdetect_errors", |
|
"Upstream connection errors", |
|
["exception"]) |
|
counter_movement_frames = Counter( |
|
"camdetect_movement_frames", |
|
"Frames with movement detected in them") |
|
counter_uploaded_frames = Counter( |
|
"camdetect_uploaded_frames", |
|
"Frames uploaded via S3") |
|
counter_upload_errors = Counter( |
|
"camdetect_upload_errors", |
|
"Frames upload errors related to S3") |
|
counter_upload_dropped_frames = Counter( |
|
"camdetect_upload_dropped_frames", |
|
"Frames that were dropped due to S3 upload queue being full") |
|
counter_download_dropped_frames = Counter( |
|
"camdetect_download_dropped_frames", |
|
"Frames that were downloaded from camera, but not processed") |
|
|
|
gauge_last_frame = Gauge( |
|
"camdetect_last_frame", |
|
"Timestamp of last frame") |
|
gauge_frame_motion_detected = Gauge( |
|
"camdetect_frame_motion_detected", |
|
"Motion detected in frame") |
|
gauge_event_active = Gauge( |
|
"camdetect_event_active", |
|
"Motion event in progress") |
|
gauge_total_blocks = Gauge( |
|
"camdetect_total_blocks", |
|
"Total DCT blocks") |
|
gauge_active_blocks = Gauge( |
|
"camdetect_active_blocks", |
|
"Total active, threshold exceeding DCT blocks") |
|
gauge_upload_queue_size = Gauge( |
|
"camdetect_upload_queue_size", |
|
"Number of frames awaiting to be uploaded via S3") |
|
gauge_download_queue_size = Gauge( |
|
"camdetect_download_queue_size", |
|
"Number of frames awaiting to be processed by motion detection loop") |
|
|
|
# Reset some gauges |
|
gauge_frame_motion_detected.set(0) |
|
gauge_upload_queue_size.set(0) |
|
gauge_download_queue_size.set(0) |
|
|
|
assert SLIDE_WINDOW <= 8 # This is 256 frames which should be enough |
|
|
|
|
|
async def upload(bucket, blob: bytes, thumb: bytes, event_id): |
|
""" |
|
Upload single JPEG blob to S3 bucket |
|
""" |
|
|
|
# Generate S3 path based on the JPEG blob SHA512 digest |
|
fp = hashlib.sha512(blob).hexdigest() |
|
path = "%s/%s/%s/%s.jpg" % (fp[:4], fp[4:8], fp[8:12], fp[12:]) |
|
|
|
# First upload the thumbnail |
|
await bucket.upload_fileobj(io.BytesIO(thumb), "thumb/%s" % path) |
|
|
|
# Proceed to upload the original JPEG frame |
|
await bucket.upload_fileobj(io.BytesIO(blob), path) |
|
|
|
# Add screenshot path to the event |
|
app.ctx.coll.update_one({ |
|
"_id": event_id |
|
}, { |
|
"$addToSet": { |
|
"screenshots": path, |
|
} |
|
}) |
|
|
|
# TODO: Handle 16MB maximum document size |
|
|
|
|
|
async def uploader(queue): |
|
""" |
|
Uploader task grabs JPEG blobs from upload queue and uploads them to S3 |
|
""" |
|
session = aioboto3.Session( |
|
aws_access_key_id=AWS_ACCESS_KEY_ID, |
|
aws_secret_access_key=AWS_SECRET_ACCESS_KEY) |
|
async with session.resource("s3", endpoint_url=S3_ENDPOINT_URL) as s3: |
|
bucket = await s3.Bucket(S3_BUCKET_NAME) |
|
while True: |
|
blob, thumb, event_id = await queue.get() |
|
await upload(bucket, blob, thumb, event_id) |
|
counter_uploaded_frames.inc() |
|
gauge_upload_queue_size.set(queue.qsize()) |
|
|
|
|
|
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() |
|
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]) |
|
|
|
# Update metrics |
|
gauge_total_blocks.set(y.shape[0] * y.shape[1]) |
|
gauge_last_frame.set(dt.timestamp()) |
|
|
|
reference_frame.put(y) |
|
try: |
|
app.ctx.mask = cv2.inRange(cv2.absdiff(y, |
|
reference_frame.get()), 25, 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) |
|
motion_detected = average_blocks > threshold_blocks |
|
|
|
# Update metrics |
|
gauge_active_blocks.set(active_blocks) |
|
gauge_total_blocks.set(total_blocks) |
|
|
|
# Propagate SIGUSR1 signal handler |
|
if app.ctx.manual_trigger: |
|
print("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.insert_one({ |
|
"timestamp": dt, |
|
"event": "motion-detected", |
|
"started": dt, |
|
"finished": dt + timedelta(minutes=2), |
|
"component": "camdetect", |
|
"source": SOURCE_NAME, |
|
"screenshots": [], |
|
"action": "event", |
|
}) |
|
app.ctx.event_id = event_id = result.inserted_id |
|
gauge_event_active.set(1) |
|
|
|
# Handle buffering frames prior event start |
|
if hold_queue.full(): |
|
await hold_queue.get() |
|
hold_queue.put_nowait((blob, thumb)) |
|
|
|
# Handle image upload |
|
if motion_detected and event_id: |
|
counter_movement_frames.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((blob, thumb, event_id)) |
|
except asyncio.QueueFull: |
|
counter_upload_dropped_frames.inc() |
|
gauge_upload_queue_size.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 |
|
gauge_event_active.set(0) |
|
|
|
|
|
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 |
|
""" |
|
buf = b"" |
|
print("Upstream connection opened with status:", resp.status) |
|
async for data, end_of_http_chunk in resp.content.iter_chunks(): |
|
counter_rx_bytes.inc(len(data)) |
|
if end_of_http_chunk: |
|
counter_rx_chunks.inc() |
|
if buf: |
|
# seek end |
|
marker = data.find(b"\xff\xd9") |
|
if marker < 0: |
|
buf += data |
|
continue |
|
else: |
|
# Assemble JPEG blob |
|
blob = buf + data[:marker+2] |
|
|
|
# Parse DCT coefficients |
|
dct = loads(blob) |
|
|
|
try: |
|
# Convert Y component to 16 bit for easier handling |
|
queue.put_nowait(( |
|
datetime.utcnow(), |
|
blob, |
|
dct, |
|
generate_thumbnail(dct))) |
|
except asyncio.QueueFull: |
|
counter_download_dropped_frames.inc() |
|
data = data[marker+2:] |
|
buf = b"" |
|
counter_rx_frames.inc() |
|
|
|
# seek begin |
|
marker = data.find(b"\xff\xd8") |
|
if marker >= 0: |
|
buf = data[marker:] |
|
else: |
|
counter_dropped_bytes.inc(len(data)) |
|
|
|
|
|
async def downloader(queue: asyncio.Queue): |
|
""" |
|
Downloader task connects to MJPEG source and |
|
pushes the JPEG frames to a queue |
|
""" |
|
while True: |
|
to = aiohttp.ClientTimeout(connect=5, sock_read=2) |
|
async with aiohttp.ClientSession(timeout=to) as session: |
|
print("Opening upstream connection to %s" % url) |
|
try: |
|
async with session.get(url) as resp: |
|
await download(resp, queue) |
|
except (aiohttp.ClientError, asyncio.exceptions.TimeoutError) as e: |
|
j = "%s.%s" % (e.__class__.__module__, e.__class__.__name__) |
|
print("Caught exception %s" % j) |
|
counter_errors.labels(exception=j).inc() |
|
await asyncio.sleep(1) |
|
|
|
app = Sanic("camdetect") |
|
|
|
|
|
@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 response.write(data) |
|
counter_tx_bytes.inc(len(data)) |
|
counter_tx_frames.inc() |
|
return response.stream( |
|
stream_camera, |
|
content_type="multipart/x-mixed-replace; boundary=frame") |
|
|
|
|
|
@app.route("/debug") |
|
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) |
|
img = cv2.imdecode(arr, cv2.IMREAD_UNCHANGED) |
|
|
|
# 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_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" |
|
) |
|
|
|
|
|
@app.route("/readyz") |
|
async def ready_check(request): |
|
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) |
|
if app.ctx.mask is None: |
|
return response.text("Not enough frames", status=503) |
|
return response.text("OK") |
|
|
|
|
|
@app.route("/event") |
|
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: |
|
continue |
|
s = "data: " + json.dumps(app.ctx.mask.tolist()) + "\r\n\r\n" |
|
await response.write(s.encode()) |
|
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.db = AsyncIOMotorClient(MONGO_URI).get_default_database() |
|
app.ctx.coll = app.ctx.db[MONGO_COLLECTION] |
|
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() |
|
upload_queue = asyncio.Queue() |
|
asyncio.create_task(uploader( |
|
upload_queue)) |
|
asyncio.create_task(downloader( |
|
download_queue)) |
|
asyncio.create_task(motion_detector( |
|
ReferenceFrame(), |
|
download_queue, |
|
upload_queue)) |
|
|
|
|
|
monitor(app).expose_endpoint() |
|
|
|
try: |
|
app.run(host="0.0.0.0", |
|
port=5000, |
|
debug=bool(os.getenv("DEBUG", 0))) |
|
except KeyboardInterrupt: |
|
asyncio.get_event_loop().close()
|
|
|