diff --git a/.gitignore b/.gitignore index 935de73..e2e4275 100644 --- a/.gitignore +++ b/.gitignore @@ -26,3 +26,5 @@ share/python-wheels/ .installed.cfg *.egg MANIFEST + +frames/ \ No newline at end of file diff --git a/.ruff.toml b/.ruff.toml new file mode 100644 index 0000000..918bba0 --- /dev/null +++ b/.ruff.toml @@ -0,0 +1,41 @@ +# The contents of this file is based on https://github.com/home-assistant/core/blob/dev/pyproject.toml + +target-version = "py313" + +[lint] +select = [ + "ALL", +] + +ignore = [ + "ANN401", # Dynamically typed expressions (typing.Any) are disallowed + "ANN001", + "ANN201", + "ANN202", + "D203", # no-blank-line-before-class (incompatible with formatter) + "D212", # multi-line-summary-first-line (incompatible with formatter) + "COM812", # incompatible with formatter + "ISC001", # incompatible with formatter + "E501", # Line length is not important + "EM101", + "TRY003", # Allow exception messages to be used directly + "ERA001", # Commented out code is allowed for examples + "T201", # Prints are allowed in examples + "PLR0913", + "FBT001", # Boolean positional args are part of the existing public API + "FBT002", + "N999", + "CPY001", # disable copyright check + "D104", # disable file docstring + "D100", # disable file docstring + "S311" +] + +[lint.flake8-pytest-style] +fixture-parentheses = false + +[lint.pyupgrade] +keep-runtime-typing = true + +[lint.mccabe] +max-complexity = 25 \ No newline at end of file diff --git a/DOC.md b/DOC.md new file mode 100644 index 0000000..0666f22 --- /dev/null +++ b/DOC.md @@ -0,0 +1,121 @@ +# Reverse-Engineered Protocol Notes + +This document explains the technical details of the protocol used by the cameras, as discovered through the decompilation of SDKs found on the internet ([here](https://github.com/jameshilliard/android-p2p-sdk3.0) and [here](https://github.com/jameshilliard/HKiPhoneSDKDemo20160621)) and the analysis of network packets using Wireshark. It may therefore contain approximations resulting from reverse engineering, whilst the manufacturer has undoubtedly defined its protocol more precisely. +All the data presented here is implemented in the files `p2pcam/lan_scanner.py` and `p2pcam/lan_video.py`. + +## Discovery Process + +Camera discovery uses a UDP broadcast exchange on port `2627`. + +The scanner emits a discovery packet with this exact layout: + +```text +[0-1] 00 00 +[2-3] struct.pack("; +MainCmd=LocalData; +userType=hkclient; +status=1; +Prot=; +MacIP=; +``` + +The code sends that packet to `255.255.255.255` and to each interface broadcast address derived from the host IPv4 interfaces. The exact packet template is: + +```text +00 00 + 0C 1D 00 00 00 00> +TIME=3600;endTime=...;MainCmd=LocalData;userType=hkclient;status=1;Prot=;MacIP=; +``` + +Responses are collected on the source port and, optionally, a separate listen port. The decoder accepts two packet families: + +1. framed replies that start with `00 00` and store the packet length in the shifted outer-length field +2. inner discovery packets where `data[0] >> 4 == COMMAND_LAN_REFRESH` and the dictionary body begins at byte 9 + +Parsed fields are normalized into `LanDevice` objects by reading aliases for the same concept. For example, `HKID`, `hkid`, `DevID`, and `DSTHKID` are treated as the device identifier family, while `Prot`, `UDPPort`, `Port`, and `port` are treated as the network port family. + +The scanner ignores its own discovery echo by checking the `MacIP` field against the locally generated marker. It also recognizes 13-byte ACK packets using `_decode_ack()` and turns them into a simple online/status marker: + +```text +00 00 d0 00 20 09 00 00 00 +``` + +The result of discovery is a list of devices sorted by device ID and HKID. + +## Streaming + +Video streaming uses UDP port `5000` and follows a strict handshake/state machine before MJPEG data starts flowing. + +Every framed packet in this phase uses `_build_packet()`: + +```text +[0-1] packet counter (uint16 LE) +[2-3] outer length = total_packet_len << 4 +[4] inner_cmd +[5] inner_flag1 +[6] inner_flag2 +[7-8] inner payload length +[9-12] inner_extra (4 bytes) +[13+] payload body +``` + +The streaming sequence is exactly: + +1. send the connection ping packets several times until the camera acknowledges +2. send `HK_RES_REQ` to request the video session +3. poll with `ICMD2` until the camera answers with `SessionCreate` +4. send `SessionStart` +5. receive MJPEG chunks, ACK camera `ICMD1` polls, and periodically send continue packets +6. stop cleanly with `SessionDelete` + +The exact packet builders are: + +```text +Ping 1: +00 00 d0 00 82 0c 00 09 00 d1 07 00 00 + +Ping 2: +00 00 d0 00 a2 0c 40 09 00 d1 07 00 00 + +HK_RES_REQ body: +id=;ftN0=video.vbVideo.MPEG4;ftN1=net.0;ftN2=HKPCPresent.HKPCPresent;opN2=;Callid=;sidN=;AsCode=337;MainCmd=HK_RES_REQ;user=Lan user; + +ICMD2 poll body: +d4:ICMD2:293:SEQ1::GUARDSEQ1: + +SessionStart body: +MainCmd=SessionStart;sidN=;ftN0=HKPCPresent.HKPCPresent;FD0=4;ftN1=net.1024;FD1=1024; + +ICMD1 ACK body: +d4:ICMD1:293:lastreq1::SEQ3:e + +SessionDelete body: +sidN=;MainCmd=SessionDelete;coz=; +``` + +Those bodies are encoded exactly as follows: + +- dictionary-style packets keep the first 2 bytes in plain ASCII and XOR the remaining bytes with `0xE9` +- `ICMD`-style packets XOR every byte with `0xE9` + +The `HK_RES_REQ`, `SessionStart`, `ICMD1` ACK, `ICMD2` poll, and `SessionDelete` builders all follow those rules. The code uses a fixed session identifier and call identifier because those values are hard-coded defaults in the implementation. + +After `SessionCreate`, the camera begins sending MJPEG fragments. The assembler takes `data[4:]` as the payload, looks for `FF D8` to start a frame, and looks for `FF D9` to close it. If `_in_frame` is already true, the payload is appended before searching for EOI. + +Two additional keepalive mechanisms are required while streaming: + +- the client replies to camera `ICMD1` polls with `_build_icmd1_ack(hkid, seq, session_id)` +- the client sends `cont_state.next_packet()` every 5 received fragments + +The continue packet builder is intentionally stateful because the camera expects the exact `_ContinueState` sequence: `nb_digits`, `idx`, `base_index`, and `_fragment_index` are mutated across calls, bytes `[2]` and `[7]` are rewritten for the current digit width, and the packet ends with `_CONT_END`. + +When the stream ends, the client sends `SessionDelete` so the camera releases the session and the next connection attempt starts from a clean state. \ No newline at end of file diff --git a/README.md b/README.md index 4ddd10b..896e5c8 100644 --- a/README.md +++ b/README.md @@ -1,57 +1,71 @@ +

+ +

+ # P2PCam -Class to retrieve camera images from cameras using the p2p protocol + +Classes to retrieve camera images from cameras using the p2p protocol First of all i just wrote it to work as a class, the original connection and retrieval process has been made by [Jheyman](https://github.com/jheyman/) in his [videosurveillance script](https://github.com/jheyman/videosurveillance/). I rewrote it to run as a class instead of an application. -So i had this [chinese camera](https://nl.aliexpress.com/item/Phone-monitor-P2P-Free-DDNS-Ontop-RT8633-HD-1-4-CMOS-1-0MP-Network-IP-Camera/990524792.html) laying around, it had this feature that you could access it from outside your home without the need for port forwarding. However after a couple of years this brand dissappeared and with it their services so i couldn't connect to it outside of my own network using [this app](https://play.google.com/store/apps/details?id=x.p2p.cam). +So i had this [chinese camera](https://nl.aliexpress.com/item/Phone-monitor-P2P-Free-DDNS-Ontop-RT8633-HD-1-4-CMOS-1-0MP-Network-IP-Camera/990524792.html) laying around, it had this feature that you could access it from outside your home without the need for port forwarding. However after a couple of years this brand dissappeared and with it their services so i couldn't connect to it outside of my own network using [this app](https://apkpure.com/p2pcamviewer/x.p2p.cam). Which made owning this camera quite useless. But i had since gotten into Home Asssistant and got the idea to get it working in there since my instance ran locally so it should be able to access the camera. -## Usage -``` -import p2pcam -import cv2 -import numpy as np +It has been confirmed that this script works with cameras labelled MD81 and MD81S. + +## Quick start + +You can use the cli.py script to quickly test your camera. -def saveFile(cam, jpeg): - RGBImage = cv2.imdecode(np.fromstring(jpeg, dtype=np.uint8), cv2.IMREAD_COLOR) - cv2.imwrite('image.jpg', RGBImage) +```bash +# Detect cameras on your local network +python3 cli.py -camera = p2pcam.P2PCam(, ) -saveFile(camera, camera.retrieveImage()) +# Detect a camera on your network, connect to it and save 10 JPEG frames in the frames folder +python3 cli.py --video --max-frames 10 --outdir frames/ + +# Detect a camera on your network, connect to it and start an HTTP MJPEG server on port 8080 +python3 cli.py --video --serve --port 8080 + +# If you want to use image transformations first install pillow +pip install pillow +# Then you can append --vertical-flip, --horizontal-flip or --add-timestamp to any command ``` -## Methods and Variables -### Methods -Any methods that may be useful. -`camera.initialize()` Set some variables and attempt to connect to the camera for the first time. +## API + +### LanScanner + +#### `refresh(timeout: float = 3.0) -> list[LanDevice]` + +Broadcasts a LAN refresh packet, waits for camera responses, and returns the discovered devices sorted by device ID and HKID. -`camera.retrieveImage()` Retrieve a jpeg string from the camera. +### LanVideoClient -`camera.start()` Start a while true loop staying connected, this will not do anything if `onJpegReceived` isn't set. +#### `stream(timeout: float = 60.0) -> Iterator[bytes]` -`camera.loop()` Start a while loop doing `retrieveImage()` until a socket error occurs. -### Variables -Some variables you may want to set. +Opens the UDP session, performs the full camera handshake, and yields complete JPEG frames as they become available. -`camera.horizontal_flip` Flip camera horizontally. (if true requires numpy and cv2) +#### `close() -> None` -`camera.vertical_flip` Flip camera vertically. (if true requires numpy and cv2) +Stops the stream and closes the UDP socket. Call this when you want to end the session without waiting for the generator to finish. -`camera.addTimeStamp` Add a timestamp to the image. (if true requires numpy and cv2) +### MJPEGServer -`camera.debug` If true prints out some debugging information. +#### `update_frame(frame: bytes) -> None` +Replaces the currently broadcast frame and notifies all connected HTTP clients waiting on the next image. -The port information will have to be set before initialisation. +#### `start() -> None` -`camera.UDP_PORT_HOST` Host udp port default: 5123 +Starts the threaded HTTP server and exposes the MJPEG stream on `/stream`. -`camera.UDP_PORT_TARGET` Target udp port default: 5000 +#### `stop() -> None` -`camera.SOCKET_TIMEOUT` Sets the socket timeout in seconds. +Stops the HTTP server cleanly and closes its socket. -`camera.NB_FRAGMENTS_TO_ACCUMULATE` How many packets to get a full image. If you put this number high you will get a higher quality image but it will take longer to retrieve. Default: 80 +## Protocol documentation -`camera.onJpegReceived` Callback that will be executed if a jpeg image is retrieved. first argument will be the camera class, the second argument will be the jpeg image string. +You can find more documentation about the protocol [in the DOC file](DOC.md). \ No newline at end of file diff --git a/cli.py b/cli.py new file mode 100644 index 0000000..acc6617 --- /dev/null +++ b/cli.py @@ -0,0 +1,162 @@ +import time +from collections.abc import Iterable +from pathlib import Path + +from p2pcam import LanDevice, LanScanner + + +def format_devices(devices: Iterable[LanDevice]) -> str: + """Format discovered devices as human-readable text.""" + lines = [] + for device in devices: + state = "online" if device.online else f"status={device.status}" + lines.append( + f"{device.device_id} type={device.device_type} hkid={device.hkid} " + f"channels={device.channel_count} audio={device.audio_type} {state}" + ) + return "\n".join(lines) + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Discover HeKai LAN devices") + + parser.add_argument("--timeout", type=float, default=3.0) + parser.add_argument("--encoding", default="utf-8") + parser.add_argument( + "--video", + action="store_true", + help="Start video stream on first discovered camera", + ) + parser.add_argument( + "--outdir", default="frames", help="Directory to save video frames" + ) + parser.add_argument( + "--max-frames", + type=int, + default=0, + help="Maximum number of frames to capture (0 = unlimited)", + ) + parser.add_argument( + "--serve", + action="store_true", + help="Start an HTTP MJPEG server to expose the stream", + ) + parser.add_argument( + "--port", + type=int, + default=8080, + help="Port for the HTTP MJPEG server (default: 8080)", + ) + parser.add_argument( + "--vertical-flip", + action="store_true", + help="Vertically flip the video stream", + ) + parser.add_argument( + "--horizontal-flip", + action="store_true", + help="Horizontally flip the video stream", + ) + parser.add_argument( + "--add-timestamp", + action="store_true", + help="Add date related text to the video stream", + ) + + args = parser.parse_args() + + scanner = LanScanner(encoding=args.encoding) + found = scanner.refresh(timeout=args.timeout) + + print("Found devices:") + print(format_devices(found)) + + if args.video and found: + from p2pcam import LanVideoClient + + # The IP address is stored in the ip attribute + target = found[0] + ip = target.ip or target.device_id.split(":")[0] + print(f"\nStarting video stream from {ip} (HKID: {target.hkid})...") + + server = None + if args.serve: + from p2pcam import MJPEGServer + + server = MJPEGServer(port=args.port) + server.start() + else: + Path(args.outdir).mkdir(parents=True, exist_ok=True) + + count = 0 + try: + with LanVideoClient(camera_ip=ip, hkid=target.hkid) as client: + for raw_frame in client.stream(timeout=10.0): + count += 1 + frame = raw_frame + + if args.vertical_flip or args.horizontal_flip or args.add_timestamp: + try: + from io import BytesIO + + from PIL import Image, ImageDraw, ImageFont + + input_frame = Image.open(BytesIO(raw_frame)) + output_frame = BytesIO() + # Image flips + if args.vertical_flip: + input_frame = input_frame.transpose( + Image.FLIP_TOP_BOTTOM + ) + if args.horizontal_flip: + input_frame = input_frame.transpose( + Image.FLIP_LEFT_RIGHT + ) + # Timestamp + if args.add_timestamp: + draw = ImageDraw.Draw(input_frame) + try: + font = ImageFont.truetype("arial.ttf", 15) + except OSError: + font = ImageFont.load_default() + draw.text( + (10, 10), + time.strftime("%Y-%m-%d %H:%M:%S"), + font=font, + fill=(255, 255, 255), + stroke_width=1, + stroke_fill=(0, 0, 0), + ) + input_frame.save(output_frame, format="JPEG") + # Replace the frame content + frame = output_frame.getvalue() + except (OSError, ValueError): + # Some frames may be corrupted so PIL cannot work with it + # We can simply ignore these frames + count -= 1 + continue + + if server: + server.update_frame(frame) + if count % 30 == 0: + print(f"Streamed {count} frames...") + else: + path = Path(args.outdir) / f"frame_{count:04d}.jpg" + with path.open("wb") as f: + f.write(frame) + print(f"Captured frame {count} to {path} ({len(frame)} bytes)") + + if args.max_frames > 0 and count >= args.max_frames: + break + + except KeyboardInterrupt: + print("\nStreaming stopped by user.") + finally: + client.close() + if server: + server.stop() + + if not server: + print(f"Captured {count} frames in total.") diff --git a/example.py b/example.py deleted file mode 100644 index 60ea06b..0000000 --- a/example.py +++ /dev/null @@ -1,39 +0,0 @@ -#!/usr/bin/env python3 -import p2pcam as Camera -import cv2 -import numpy as np - -def saveFile(cam, jpeg): - RGBImageNext = cv2.imdecode(np.fromstring(jpeg, dtype=np.uint8), cv2.IMREAD_COLOR) - cv2.imwrite('image.jpg', RGBImageNext) - - -camera = Camera.P2PCam("192.168.178.28", "192.168.178.9") -camera.NB_FRAGMENTS_TO_ACCUMULATE = 20 -# camera.SOCKET_TIMEOUT = 20 -# camera.debug = True - -# Loop in scripts loop. Not recommended since you are lacking control. - -# camera.onJpegReceived = saveFile -# camera.start() - -# Synchronously fetch image from camera (asynchronous is not in the package yet.) -# -# Initialisation is not needed since it initialises in the retrieveImage function as well -# camera.initialize() -# saveFile(camera, camera.retrieveImage()) - -# A loop to keep retrieving pictures. initialisation is needed since otherwise socket_error will not be set. - -camera.initialize() -while camera.socket_error == False: - try: - jpeg = camera.retrieveImage() - print('got an image!') - saveFile(camera, jpeg) - except KeyboardInterrupt: - raise - except Exception as e: - print(("[ERROR] " + str(e))) - pass diff --git a/github/logo.png b/github/logo.png new file mode 100644 index 0000000..2b958b4 Binary files /dev/null and b/github/logo.png differ diff --git a/github/workflows/ruff.yml b/github/workflows/ruff.yml new file mode 100644 index 0000000..90d3f83 --- /dev/null +++ b/github/workflows/ruff.yml @@ -0,0 +1,14 @@ +name: Ruff +on: + - push + - pull_request + +jobs: + ruff: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/ruff-action@v3 + with: + src: "./p2pcam" + version: "0.15.16" \ No newline at end of file diff --git a/p2pcam/__init__.py b/p2pcam/__init__.py index 212eea8..3c45fc7 100644 --- a/p2pcam/__init__.py +++ b/p2pcam/__init__.py @@ -1,530 +1,13 @@ -name = "p2pcam" -import sys, traceback -import socket -import time -import random - -class RestartException(Exception): - def __init__(self, msg, delay=1): - self.msg = msg - self.delay = delay - - def __str__(self): - return repr(self.msg) - -class P2PCam(): - def __init__(self, host_ip, target_ip): - self.horizontal_flip = False - self.vertical_flip = False - self.addTimeStamp = False - self.debug = False - - self.global_loop_iteration = 0 - # Our IP address - self.UDP_IP_HOST = host_ip - self.UDP_PORT_HOST = 5123 # arbitrary - # Camera IP address and UDP port - self.UDP_IP_TARGET = target_ip - self.UDP_PORT_TARGET = 5000 - # network socket timeout in seconds - self.SOCKET_TIMEOUT = 2 - # Nb of UDP packets to receive to get one full image. - # Packets are typically 904 bytes, 80 packets is about 70kB, which is enough - # for the camera setup I used (640x480 medium quality jpeg) - self.NB_FRAGMENTS_TO_ACCUMULATE = 80 - self.jpeg = None - self.MESSAGE_43 = bytearray( - [0x00, 0x00, 0xb0, 0x02, 0x82, 0x00, 0x00, 0x27, 0x00, 0x01, 0x00, 0x00, 0x00, 0x4d, 0x61, 0x63, 0x49, 0x50, - 0x3d, 0x42, 0x43, 0x2d, 0x41, 0x45, 0x2d, 0x43, 0x35, 0x2d, 0x37, 0x43, 0x2d, 0x37, 0x37, 0x2d, 0x37, 0x42, - 0x2b, 0x31, 0x36, 0x34, 0x36, 0x37, 0x3b]) - self.MESSAGE_13_1 = bytearray([0x00, 0x00, 0xd0, 0x00, 0x82, 0x00, 0x06, 0x09, 0x00, 0x01, 0x00, 0x00, 0x00]) - self.MESSAGE_13_2 = bytearray([0x00, 0x00, 0xd0, 0x00, 0xa2, 0x00, 0x06, 0x09, 0x00, 0x01, 0x00, 0x00, 0x00]) - self.MESSAGE_13_3 = bytearray([0x00, 0x00, 0xd0, 0x00, 0x62, 0x00, 0x06, 0x09, 0x00, 0x01, 0x00, 0x00, 0x00]) - self.MESSAGE_212 = bytearray( - [0x01, 0x00, 0x40, 0x0d, 0x32, 0x00, 0x00, 0xd0, 0x00, 0x51, 0x01, 0x00, 0x00, 0x69, 0x64, 0xd4, 0xd8, 0xd8, - 0xd2, 0x8f, 0x9d, 0xa7, 0xd9, 0xd4, 0x9f, 0x80, 0x8d, 0x8c, 0x86, 0xc7, 0x9f, 0x8b, 0xbf, 0x80, 0x8d, 0x8c, - 0x86, 0xc7, 0xa4, 0xb9, 0xac, 0xae, 0xdd, 0xd2, 0x8f, 0x9d, 0xa7, 0xd8, 0xd4, 0x87, 0x8c, 0x9d, 0xc7, 0xd9, - 0xd2, 0x8f, 0x9d, 0xa7, 0xdb, 0xd4, 0xa1, 0xa2, 0xb9, 0xaa, 0xb9, 0x9b, 0x8c, 0x9a, 0x8c, 0x87, 0x9d, 0xc7, - 0xa1, 0xa2, 0xb9, 0xaa, 0xb9, 0x9b, 0x8c, 0x9a, 0x8c, 0x87, 0x9d, 0xd2, 0x86, 0x99, 0xa7, 0xdb, 0xd4, 0xdc, - 0x98, 0x8d, 0xdf, 0xa6, 0xa3, 0xdf, 0xda, 0xda, 0xdf, 0xde, 0x8f, 0x8f, 0x8f, 0xd2, 0xaa, 0x88, 0x85, 0x85, - 0x80, 0x8d, 0xd4, 0xdd, 0x85, 0x90, 0xd9, 0x81, 0x8f, 0xdc, 0x82, 0xd8, 0xde, 0xa8, 0xd9, 0xd9, 0xae, 0xb3, - 0xd8, 0xd0, 0x8f, 0xda, 0x85, 0xdc, 0xde, 0xad, 0x8a, 0xdf, 0xda, 0xda, 0xdf, 0xd9, 0x8f, 0x8f, 0xd9, 0xd2, - 0x9a, 0x80, 0x8d, 0xa7, 0xd4, 0xdc, 0x98, 0x8d, 0xdf, 0xa6, 0xa3, 0xdf, 0xda, 0xda, 0xdf, 0xde, 0x8f, 0x8f, - 0x8f, 0xd2, 0xa8, 0x9a, 0xaa, 0x86, 0x8d, 0x8c, 0xd4, 0xda, 0xda, 0xde, 0xd2, 0xa4, 0x88, 0x80, 0x87, 0xaa, - 0x84, 0x8d, 0xd4, 0xa1, 0xa2, 0xb6, 0xbb, 0xac, 0xba, 0xb6, 0xbb, 0xac, 0xb8, 0xd2, 0x9c, 0x9a, 0x8c, 0x9b, - 0xd4, 0xd8, 0xd0, 0xdb, 0xc7, 0xd8, 0xdf, 0xd1, 0xc7, 0xd9, 0xc7, 0xda, 0xda, 0xd2]) - self.MESSAGE_34 = bytearray( - [0x00, 0x00, 0x20, 0x02, 0x12, 0x00, 0x00, 0x1e, 0x00, 0x01, 0x00, 0x00, 0x00, 0xa0, 0xaa, 0xa4, 0xad, 0xd4, - 0xd8, 0xd2, 0xba, 0xac, 0xb8, 0xd4, 0xd8, 0xd2, 0xbd, 0xa0, 0xa4, 0xac, 0xd4, 0xd9, 0xd2, 0xe9]) - self.MESSAGE_119 = bytearray( - [0x02, 0x00, 0x70, 0x07, 0x32, 0x00, 0x00, 0x73, 0x00, 0x64, 0x00, 0x00, 0x00, 0x4d, 0x61, 0x80, 0x87, 0xaa, - 0x84, 0x8d, 0xd4, 0xba, 0x8c, 0x9a, 0x9a, 0x80, 0x86, 0x87, 0xba, 0x9d, 0x88, 0x9b, 0x9d, 0xd2, 0x9a, 0x80, - 0x8d, 0xa7, 0xd4, 0xdc, 0x98, 0x8d, 0xdf, 0xa6, 0xa3, 0xdf, 0xda, 0xda, 0xdf, 0xde, 0x8f, 0x8f, 0x8f, 0xd2, - 0x8f, 0x9d, 0xa7, 0xd9, 0xd4, 0xa1, 0xa2, 0xb9, 0xaa, 0xb9, 0x9b, 0x8c, 0x9a, 0x8c, 0x87, 0x9d, 0xc7, 0xa1, - 0xa2, 0xb9, 0xaa, 0xb9, 0x9b, 0x8c, 0x9a, 0x8c, 0x87, 0x9d, 0xd2, 0xaf, 0xad, 0xd9, 0xd4, 0xdb, 0xdc, 0xd8, - 0xdd, 0xd0, 0xdb, 0xd1, 0xd1, 0xd2, 0x8f, 0x9d, 0xa7, 0xd8, 0xd4, 0x87, 0x8c, 0x9d, 0xc7, 0xd8, 0xd9, 0xdb, - 0xdc, 0xd2, 0xaf, 0xad, 0xd8, 0xd4, 0xd8, 0xd9, 0xdb, 0xdc, 0xd2]) - # Allowed byte length received after MESSAGE_119, since not all cameras send the same byte length in return. - self.allowedPacketLengths = [368, 334, 372] - # The continue packet is composed of a first part where the 0xff below get dynamically replaced by the appropriate value at runtime, and a second part that is invariable - self.MESSAGE_CONTINUE_BEGIN = bytearray( - [0x00, 0x00, 0xff, 0x02, 0x12, 0x00, 0x00, 0xff, 0x00, 0x01, 0x00, 0x00, 0x00, 0xa0, 0xaa, 0xa4, 0xad, 0xd4, - 0xd8, 0xd2, 0xba, 0xac, 0xb8, 0xd4]) - self.MESSAGE_CONTINUE_END = bytearray([0xd2, 0xbd, 0xa0, 0xa4, 0xac, 0xd4, 0xd9, 0xd2, 0xe9]) - # in the continue packet, each digit goes through this sequence - self.CONTINUE_LIST_2 = bytearray([0xd9, 0xd8, 0xdb, 0xda, 0xdd, 0xdc, 0xdf, 0xde, 0xd1, 0xd0]) - # in the continue packet, the last digit toggles between two values (e.g. 0xd8 and 0xdf) - # periodically, change the toggle set (e.g move to 0xdb/0xde) - self.CONTINUE_LIST_1 = bytearray( - [0xd8, 0xdf, 0xdb, 0xde, 0xda, 0xd1, 0xdd, 0xd0, 0xdc, 0xd9, 0xdf, 0xd8, 0xde, 0xdb, 0xd1, 0xda, 0xd0, 0xdd, - 0xd9, 0xdc]) - self.sock = None - self.onJpegReceived = None - self.latestErrorMsg = None - # buffer for control/initialization packets reception - self.buffer = bytearray(1024) - self.global_loop_iteration = 0 - self.msg = bytearray() - self.timeout_iteration = 0 - self.hasInitialised = False - - def byteToInt(self, byteVal): - return byteVal - - def sendControlPacket(self, packet): - if self.debug: - print(("[CONTROL] sending %d bytes " % len(packet))) - pass - self.sock.sendto(packet, (self.UDP_IP_TARGET, self.UDP_PORT_TARGET)) - - def sendContinuePacket(self, packet): - self.sock.sendto(packet, (self.UDP_IP_TARGET, self.UDP_PORT_TARGET)) - - def receiveControlPacket(self, output): - self.sock.settimeout(self.SOCKET_TIMEOUT) - try: - nbbytes, addr = self.sock.recvfrom_into(output, 1024) - if self.debug: - print(("[CONTROL] received %d bytes " % nbbytes)) - pass - return nbbytes - except socket.timeout: - raise socket.timeout - except KeyboardInterrupt: - raise - - def start(self): - try: - self.initialize() - while True: - self.loop() - except KeyboardInterrupt: - print(("[CONTROL] manually interrupted, %s" % time.strftime("%Y-%m-%d @ %H:%M:%S"))) - except NameError as n: - print(("[ERROR] NameError %s" % n)) - except: - exc_type, exc_value, exc_traceback = sys.exc_info() - traceback.print_exception(exc_type, exc_value, exc_traceback, limit=2, file=sys.stdout) - del exc_traceback - print("[CONTROL] exiting surveillance") - - def initialize(self): - try: - if self.horizontal_flip and self.vertical_flip: - self.flipcode = -1 - elif self.horizontal_flip: - self.flipcode = 0 - elif self.vertical_flip: - self.flipcode = 1 - self.global_loop_iteration += 1 - if self.debug: - print("*****************************************************************") - print(("Global loop iteration #%d started on %s" % ( - self.global_loop_iteration, time.strftime("%Y-%m-%d @ %H:%M:%S")))) - print("*****************************************************************") - pass - - ######################### - # VARIOUS INITIALIZATIONS - ######################### - # the 7th byte in the 13 byte msg seems to be arbitrary: pick any random value for which bit 4 is not already set - val = random.randint(0, 16) - self.MESSAGE_13_1[6] = val - self.MESSAGE_13_2[6] = val - self.MESSAGE_13_3[6] = val - self.msg = b'' - self.imageIndex = 0 - self.lastFragmentId = 0 - self.fragmentIndex = 0 - self.nbDigits = 1 - self.continue_index = [0, 0, 0, 0, 0] - self.base_index = 0 - self.fragments_received = 0 - self.bytes = '' - self.socket_error = False - ####################### - # NETWORK RELATED SETUP - ####################### - # In case this is not the first run - if self.sock: - self.sock.close() - # Open UDP socket to talk to camera - try: - self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - self.sock.bind((self.UDP_IP_HOST, self.UDP_PORT_HOST)) - except socket.error as e: - if self.sock: - self.sock.close() - self.latestErrorMsg = e.strerror - raise RestartException("Could not open socket: " + e.strerror, 15) - except KeyboardInterrupt: - raise - ################################ - # CAMERA INITIALIZATION SEQUENCE - ################################ - self.sendControlPacket(self.MESSAGE_43) - self.sendControlPacket(self.MESSAGE_13_1) - - for i in range(5): - try: - nbReceived = self.receiveControlPacket(self.buffer) - if ((nbReceived == 13) and (self.buffer[4] == (self.MESSAGE_13_1[4] | 0b00010000)) and ( - self.buffer[6] != (self.MESSAGE_13_1[6] | 0b01000000))): - if self.debug: - print("[CONTROL] status ok") - pass - break - else: - if self.debug: - print("[CONTROL] status ko, repeating") - pass - # print("status ko: repeat (buff4=%x, mess4and=%x)" % (self.buffer[4],(MESSAGE_13_1[4] | 0b00010000))) - if (i == 9): - raise RestartException("Max number of status check loops reached", 15) - except socket.timeout: - if self.timeout_iteration > 3: - raise Exception('Not responding') - self.timeout_iteration +=1 - raise RestartException("Socket timeout 1", 1) - except KeyboardInterrupt: - raise - self.sendControlPacket(self.MESSAGE_13_2) - self.sendControlPacket(self.MESSAGE_13_3) - try: - nbReceived = self.receiveControlPacket(self.buffer) - if (nbReceived != 13): - raise RestartException("Expected 13 bytes, received %d" % nbReceived, 5) - except socket.timeout: - raise RestartException("Socket timeout 2", 15) - except KeyboardInterrupt: - raise - self.sendControlPacket(self.MESSAGE_212) - try: - nbReceived = self.receiveControlPacket(self.buffer) - # Sometimes the 42 bytes packet comes before the 115: discard it and re-read - if (nbReceived == 42): - if self.debug: - print("[CONTROL] received 42 early, re-reading") - pass - nbReceived = self.receiveControlPacket(self.buffer) - if (nbReceived != 155 and nbReceived != 156): - raise RestartException("Expected 155 bytes, received %d" % nbReceived, 15) - except socket.timeout: - raise RestartException("Socket timeout 3", 15) - except KeyboardInterrupt: - raise - self.sendControlPacket(self.MESSAGE_34) - self.sendControlPacket(self.MESSAGE_119) - try: - nbReceived = self.receiveControlPacket(self.buffer) - # Sometimes the 42 bytes packet comes at this point: discard it and re-read - if (nbReceived == 42): - if self.debug: - print("[CONTROL] received 42 late, re-reading") - pass - nbReceived = self.receiveControlPacket(self.buffer) - if (nbReceived not in self.allowedPacketLengths): - raise RestartException( - "Expected one of %(allowedPacketLengths)s bytes, received %(receivedBytes)d" % { - "allowedPacketLengths": self.allowedPacketLengths, "receivedBytes": nbReceived}, 15) - except socket.timeout: - raise RestartException("Socket timeout 4", 15) - except KeyboardInterrupt: - raise - # reception packet TAPA 410 & paquet 42 bytes - # receiveControlPacket(self.buffer) - except RestartException as resExc: - if self.debug: - print(("[ERROR] restarting global loop in %d seconds due to exception: %s" % (resExc.delay, resExc.msg))) - pass - # Let the camera breathe a bit before trying again - time.sleep(resExc.delay) - self.initialize() - pass - except KeyboardInterrupt: - raise - - - def retrieveImage(self): - if not self.hasInitialised: - self.initialize() - self.hasInitialised = True - try: - ############################ - # BEGIN IMAGE RECEPTION LOOP - ############################ - foundImage = False - - while self.socket_error == False and foundImage == False: - self.sock.settimeout(self.SOCKET_TIMEOUT) - # Receive UDP fragment - try: - chunk = self.sock.recv(1024) - except KeyboardInterrupt: - raise - except Exception as e: - self.socket_error = True - if self.debug: - print("[DATA] self.sock.recv error: " + str(e)) - pass - self.initialize() - return self.retrieveImage() - - nbbytes = len(chunk) - self.fragments_received += 1 - self.fragmentIndex += 1 - if (self.fragments_received <= self.NB_FRAGMENTS_TO_ACCUMULATE): - # Filter out any potential non-image-data packets (e.g. 13 bytes statuses) - if nbbytes >= 17: - # First frame / Start of Image : get rid of the 15 bytes header - if (chunk[15] == 255) and (chunk[16] == 216): - self.lastFragmentId = self.byteToInt(chunk[0]) - self.msg += chunk[15:] - # additional data fragment : just drop the 4 bytes header and concatenate to already received data - else: - # Check for sequence number continuity - if ((self.byteToInt(chunk[0]) == self.lastFragmentId + 1) or ( - self.byteToInt(chunk[0]) == 0) and (self.lastFragmentId == 255)): - self.msg += chunk[4:] - # If we lost a fragment, no point in continuing accumulating data for this frame so restart another data grab - else: - self.msg = b'' - self.fragments_received = 0 - # Keep track of sequence number - self.lastFragmentId = self.byteToInt(chunk[0]) - # If we received an unexpected packet in the middle of the image data, something is wrong : just drop the ongoing image capture & restart - else: - self.msg = b'' - self.fragments_received = 0 - self.manageContinuePackets() - else: - # We now normally have enough data so that a full image is present in the buffer: search for SOI and EOI markers - # SOI = 0xffd8 - # EOI = 0xffd9 - SOI_index = -1 - EOI_index = -1 - for index in range(0, len(self.msg) - 1): - if (self.msg[index] == 255): - if self.msg[index + 1] == 216: - SOI_index = index - for index in range(index + 2, len(self.msg) - 1): - if (self.msg[index] == 255): - if self.msg[index + 1] == 217: - EOI_index = index - break - break - if SOI_index != -1 and EOI_index != -1: - # A complete image was indeed found in the data buffer : isolate the image data in a dedicated buffer - # Keep the rest of data for next iterations - self.jpeg = self.msg[SOI_index:EOI_index + 2] - self.msg = self.msg[EOI_index + 2:] - try: - foundImage = True - if callable(self.onJpegReceived): - self.onJpegReceived(self, self.jpeg) - except KeyboardInterrupt: - raise - except: - exc_type, exc_value, exc_traceback = sys.exc_info() - traceback.print_exception(exc_type, exc_value, exc_traceback, limit=2, file=sys.stdout) - del exc_traceback - pass - # Log a trace every 5 min or so (5*60s*3img/sec) - if (self.imageIndex % 900 == 0): - if self.debug: - print(("[DATA] Still alive %s, image index %d" % ( - time.strftime("%Y-%m-%d @ %H:%M:%S"), self.imageIndex))) - pass - self.imageIndex += 1 - # else: - # print("no image found in stream among %d bytes"% len(self.msg)) - # Restart another data grab - self.msg = b'' - self.fragments_received = 0 - self.manageContinuePackets() - if hasattr(self, 'flipcode') or self.addTimeStamp: - import cv2, numpy - image = cv2.imdecode(numpy.fromstring(self.jpeg, dtype=numpy.uint8),cv2.IMREAD_COLOR) - if hasattr(self, 'flipcode'): - cv2.flip(image, self.flipcode, image) - if self.addTimeStamp: - cv2.putText(image, time.strftime("%Y-%m-%d %H:%M:%S"), (10, 460), cv2.FONT_HERSHEY_SIMPLEX, 0.85, (0, 0, 255), 2, 8) - return cv2.imencode('.jpg', image)[1].tostring() - return self.jpeg - if self.socket_error: - self.initialize() - return self.retrieveImage() - except RestartException as resExc: - if self.debug: - print(("[ERROR] restarting global loop in %d seconds due to exception: %s" % (resExc.delay, resExc.msg))) - pass - # Let the camera breathe a bit before trying again - time.sleep(resExc.delay) - pass - except KeyboardInterrupt: - raise - - def manageContinuePackets(self): - #################################### - # MANAGE "CONTINUE" PACKETS SEQUENCE - #################################### - # Send out a feedback message every 5 fragments received, to tell the camera to keep sending frames. - if (self.fragmentIndex % 5) == 0: - tmp = bytearray() - if (self.nbDigits == 1): - self.MESSAGE_CONTINUE_BEGIN[2] = 0x20 - self.MESSAGE_CONTINUE_BEGIN[7] = 0x1e - tmp.append(self.CONTINUE_LIST_1[self.base_index + self.continue_index[0]]) - self.continue_index[0] += 1 - if self.continue_index[0] == 2: - self.nbDigits += 1 - self.continue_index[1] = 1 # start at d8 - self.continue_index[0] = 0 - elif (self.nbDigits == 2): - self.MESSAGE_CONTINUE_BEGIN[2] = 0x30 - self.MESSAGE_CONTINUE_BEGIN[7] = 0x1f - tmp.append(self.CONTINUE_LIST_2[self.continue_index[1]]) - tmp.append(self.CONTINUE_LIST_1[self.base_index + self.continue_index[0]]) - self.continue_index[0] += 1 - if self.continue_index[0] == 2: - self.continue_index[1] += 1 - self.continue_index[0] = 0 - if self.continue_index[1] == len(self.CONTINUE_LIST_2): - self.nbDigits += 1 - self.continue_index[2] = 1 # start at d8 - self.continue_index[1] = 0 # start at d9 - self.continue_index[0] = 0 - elif (self.nbDigits == 3): - self.MESSAGE_CONTINUE_BEGIN[2] = 0x40 - self.MESSAGE_CONTINUE_BEGIN[7] = 0x20 - tmp.append(self.CONTINUE_LIST_2[self.continue_index[2]]) - tmp.append(self.CONTINUE_LIST_2[self.continue_index[1]]) - tmp.append(self.CONTINUE_LIST_1[self.base_index + self.continue_index[0]]) - # update digit 0 - self.continue_index[0] += 1 - # update digit 1 - if self.continue_index[0] == 2: - self.continue_index[1] += 1 - self.continue_index[0] = 0 - # update digit 2 - if self.continue_index[1] == len(self.CONTINUE_LIST_2): - self.continue_index[2] += 1 - self.continue_index[1] = 0 # start at d9 - self.continue_index[0] = 0 - # check for adding one more digit - if self.continue_index[2] == len(self.CONTINUE_LIST_2): - self.nbDigits += 1 - self.continue_index[3] = 1 # start at d8 - self.continue_index[2] = 0 # start at d9 - self.continue_index[1] = 0 # start at d9 - self.continue_index[0] = 0 # start at d9 - elif (self.nbDigits == 4): - self.MESSAGE_CONTINUE_BEGIN[2] = 0x50 - self.MESSAGE_CONTINUE_BEGIN[7] = 0x21 - tmp.append(self.CONTINUE_LIST_2[self.continue_index[3]]) - tmp.append(self.CONTINUE_LIST_2[self.continue_index[2]]) - tmp.append(self.CONTINUE_LIST_2[self.continue_index[1]]) - tmp.append(self.CONTINUE_LIST_1[self.base_index + self.continue_index[0]]) - # update digit 0 - self.continue_index[0] += 1 - # update digit 1 - if self.continue_index[0] == 2: - self.continue_index[1] += 1 - self.continue_index[0] = 0 - # update digit 2 - if self.continue_index[1] == len(self.CONTINUE_LIST_2): - self.continue_index[2] += 1 - self.continue_index[1] = 0 # start at d9 - self.continue_index[0] = 0 - # update digit 3 - if self.continue_index[2] == len(self.CONTINUE_LIST_2): - self.continue_index[3] += 1 # start at d8 - self.continue_index[2] = 0 # start at d9 - self.continue_index[1] = 0 # start at d9 - self.continue_index[0] = 0 # start at d9 - # check for adding one more digit - if self.continue_index[3] == len(self.CONTINUE_LIST_2): - self.nbDigits += 1 - self.continue_index[4] = 1 # start at d8 - self.continue_index[3] = 0 # start at d9 - self.continue_index[2] = 0 # start at d9 - self.continue_index[1] = 0 # start at d9 - elif (self.nbDigits == 5): - self.MESSAGE_CONTINUE_BEGIN[2] = 0x60 - self.MESSAGE_CONTINUE_BEGIN[7] = 0x22 - tmp.append(self.CONTINUE_LIST_2[self.continue_index[4]]) - tmp.append(self.CONTINUE_LIST_2[self.continue_index[3]]) - tmp.append(self.CONTINUE_LIST_2[self.continue_index[2]]) - tmp.append(self.CONTINUE_LIST_2[self.continue_index[1]]) - tmp.append(self.CONTINUE_LIST_1[self.base_index + self.continue_index[0]]) - # update digit 0 - self.continue_index[0] += 1 - # update digit 1 - if self.continue_index[0] == 2: - self.continue_index[1] += 1 - self.continue_index[0] = 0 - # update digit 2 - if self.continue_index[1] == len(self.CONTINUE_LIST_2): - self.continue_index[2] += 1 - self.continue_index[1] = 0 # start at d9 - self.continue_index[0] = 0 - # update digit 3 - if self.continue_index[2] == len(self.CONTINUE_LIST_2): - self.continue_index[3] += 1 # start at d8 - self.continue_index[2] = 0 # start at d9 - self.continue_index[1] = 0 # start at d9 - self.continue_index[0] = 0 # start at d9 - # update digit 4 - if self.continue_index[3] == len(self.CONTINUE_LIST_2): - self.continue_index[4] += 1 # start at d8 - self.continue_index[3] = 0 # start at d9 - self.continue_index[2] = 0 # start at d9 - self.continue_index[1] = 0 # start at d9 - self.continue_index[0] = 0 # start at d9 - if self.continue_index[4] == len(self.CONTINUE_LIST_2): - # restart sequence - # print("RESTARTING SEQUENCE") - self.nbDigits = 1 - # horrible reverse-engineered condition to restart sequence at 1 digit - if len(tmp) == 5 and tmp.startswith(b'\xdf\xdc\xdd\xd0'): - self.nbDigits = 1 - # horrible experimentally-determined condition to change the toggle data set for the last byte - if (self.fragmentIndex % 100 == 0): - self.base_index = (self.base_index + 2) % 20 - packet = self.MESSAGE_CONTINUE_BEGIN + tmp + self.MESSAGE_CONTINUE_END - self.sendContinuePacket(packet) - - def loop(self): - try: - while (self.socket_error == False): - self.retrieveImage() - except RestartException as resExc: - if self.debug: - print(("[ERROR] restarting global loop in %d seconds due to exception: %s" % (resExc.delay, resExc.msg))) - pass - # Let the camera breathe a bit before trying again - time.sleep(resExc.delay) - pass - except KeyboardInterrupt: - raise - # end of global loop +from .lan_device import LanDevice +from .lan_scanner import LanScanner +from .lan_video import LanVideoClient +from .mjpeg_server import MJPEGServer + +__all__ = [ + "LanDevice", + "LanScanner", + "LanVideoClient", + "MJPEGServer", + "P2PCam", + "RestartException", +] diff --git a/p2pcam/lan_device.py b/p2pcam/lan_device.py new file mode 100644 index 0000000..47374cf --- /dev/null +++ b/p2pcam/lan_device.py @@ -0,0 +1,21 @@ +from dataclasses import dataclass + +ONLINE_STATUSES = {1, 2} + + +@dataclass(frozen=True) +class LanDevice: + """Device entry returned by the SDK LAN callback.""" + + device_id: str + device_type: str + hkid: int + channel_count: int + status: int + audio_type: str + ip: str = "" + + @property + def online(self) -> bool: + """True when the SDK reports the device as online.""" + return self.status in ONLINE_STATUSES diff --git a/p2pcam/lan_scanner.py b/p2pcam/lan_scanner.py new file mode 100644 index 0000000..b6bca4c --- /dev/null +++ b/p2pcam/lan_scanner.py @@ -0,0 +1,254 @@ +import random +import select +import socket +import struct +import time + +from .lan_device import LanDevice + +SOURCE_PORT = 2726 +BROADCAST_PORT = 2627 +LISTEN_PORT = 5000 +COMMAND_LAN_REFRESH = 0x0B +DISCOVERY_HEADER_SIZE = 13 +DISCOVERY_MIN_PACKET_LEN = 12 +MIN_DISCOVERY_PACKET_SIZE = 9 +ACK_PACKET_SIZE = 13 +IP_OCTET_COUNT = 4 + + +class LanScanner: + """ + Scan the local network for compatible P2P camera devices. + + This class broadcasts discovery packets on the local network and listens + for device responses. It decodes responses into LanDevice instances. + """ + + def __init__( + self, + listen_port: int = LISTEN_PORT, + source_port: int = SOURCE_PORT, + broadcast_port: int = BROADCAST_PORT, + encoding: str = "utf-8", + ) -> None: + """Initialize scanner networking ports and payload encoding.""" + self.listen_port = listen_port + self.source_port = source_port + self.broadcast_port = broadcast_port + self.encoding = encoding + self.mac_ip = self._make_mac_ip() + + def refresh(self, timeout: float = 3.0) -> list[LanDevice]: + """Broadcast a LAN refresh packet and collect device responses.""" + devices: dict[str, LanDevice] = {} + source_sock = self._open_socket(self.source_port) + listen_sock = None + if self.listen_port != self.source_port: + listen_sock = self._open_socket(self.listen_port) + sockets = [source_sock] + ([listen_sock] if listen_sock is not None else []) + try: + packet = self._build_refresh_packet() + source_sock.sendto(packet, ("255.255.255.255", self.broadcast_port)) + for broadcast in self._interface_broadcasts(): + source_sock.sendto(packet, (broadcast, self.broadcast_port)) + + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + readable, _, _ = select.select(sockets, [], [], remaining) + if not readable: + break + data, (host, _port) = readable[0].recvfrom(65535) + + ack = self._decode_ack(data) + if ack is not None: + devices.setdefault( + f"{host}:0:0", + LanDevice( + device_id=host, + device_type="", + hkid=0, + channel_count=0, + status=ack, + audio_type="", + ip=host, + ), + ) + + for fields in self._extract_dicts(data): + if fields.get("MacIP") == self.mac_ip: + continue + fields.setdefault("Ip", host) + device = self._device_from_fields(fields) + if device is not None: + devices[ + f"{device.device_id}:{device.hkid}:{device.channel_count}" + ] = device + finally: + for sock in sockets: + sock.close() + + return sorted(devices.values(), key=lambda item: (item.device_id, item.hkid)) + + def _open_socket(self, port: int) -> socket.socket: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + try: + sock.bind(("", port)) + except OSError: + sock.bind(("", 0)) + if port == self.source_port: + self.source_port = sock.getsockname()[1] + return sock + + def _build_refresh_packet(self) -> bytes: + end_time = int(time.time()) + 3600 + body = self._encode_old_dict( + { + "TIME": "3600", + "endTime": str(end_time), + "MainCmd": "LocalData", + "userType": "hkclient", + "status": "1", + "Prot": str(self.listen_port), + "MacIP": self.mac_ip, + } + ) + inner = self._build_inner_packet(COMMAND_LAN_REFRESH, body) + total_len = len(inner) + 4 + return b"\x00\x00" + struct.pack(" bytes: + inner_len = len(body) + 9 + header = bytearray(9) + header[0] = (command << 4) | 0x02 + header[1] = 0x0C + header[2] = 0x1D + header[3] = inner_len & 0xFF + header[4] = (inner_len >> 8) & 0xFF + return bytes(header) + body + + @staticmethod + def _build_ack_packet( + command: int = 0x08, flag: int = 0x0B, pipe: int = 0x07D1 + ) -> bytes: + return ( + b"\x00\x00\xd0\x00" + + bytes([(command << 4) | 0x02, flag, 0x20, 0x09, 0x00]) + + struct.pack(" bytes: + return "".join(f"{key}={value};" for key, value in fields.items()).encode( + "ascii" + ) + + def _extract_dicts(self, data: bytes) -> list[dict[str, str]]: + candidates = [] + if len(data) >= DISCOVERY_HEADER_SIZE and data[0:2] == b"\x00\x00": + packet_len = struct.unpack_from("> 4 + if DISCOVERY_MIN_PACKET_LEN < packet_len <= len(data): + candidates.append(data[DISCOVERY_HEADER_SIZE:packet_len]) + if len(data) >= MIN_DISCOVERY_PACKET_SIZE: + command = data[0] >> 4 + if command == COMMAND_LAN_REFRESH: + candidates.append(data[9:]) + candidates.append(data) + + decoded = [] + seen = set() + for candidate in candidates: + fields = self._decode_old_dict(candidate) + marker = tuple(sorted(fields.items())) + if fields and marker not in seen: + decoded.append(fields) + seen.add(marker) + return decoded + + def _decode_ack(self, data: bytes) -> int | None: + if len(data) != ACK_PACKET_SIZE or data[:4] != b"\x00\x00\xd0\x00": + return None + if data[4] >> 4 in {0x08, 0x09, 0x0A} and data[7:9] == b"\x09\x00": + return 1 + return None + + def _decode_old_dict(self, data: bytes) -> dict[str, str]: + text = data.split(b"\x00", 1)[0].decode(self.encoding, errors="replace") + fields: dict[str, str] = {} + for part in text.split(";"): + if "=" not in part: + continue + key, value = part.split("=", 1) + key = key.strip() + if key: + fields[key] = self._unescape_old_value(value.strip()) + return fields + + @staticmethod + def _unescape_old_value(value: str) -> str: + return ( + value.replace("^equal", "=").replace("^scolon", ";").replace("^vivi", "^") + ) + + @staticmethod + def _make_mac_ip() -> str: + return f"0x0x{random.getrandbits(32):08x}:{random.randint(0, 0x7FFFFFFF)}" + + def _device_from_fields(self, fields: dict[str, str]) -> LanDevice | None: + hkid = self._int_field(fields, "HKID", "hkid", "DevID", "DSTHKID") + port = self._int_field(fields, "Prot", "UDPPort", "Port", "port") + status = self._int_field(fields, "status", "Status", default=1) + device_id = self._first_field(fields, "devid", "DevID", "HKID", "id", "Ip") + if not device_id and hkid: + device_id = str(hkid) + if not device_id and not hkid: + return None + + return LanDevice( + device_id=device_id or "unknown", + device_type=self._first_field( + fields, "DevFlag", "devtype", "type", "Protocol" + ), + hkid=hkid or port, + channel_count=self._int_field(fields, "Count", "count", default=0), + status=status, + audio_type=self._first_field(fields, "audio", "audiotype", "AudioType"), + ip=fields.get("Ip", ""), + ) + + @staticmethod + def _first_field(fields: dict[str, str], *names: str) -> str: + for name in names: + value = fields.get(name) + if value: + return value + return "" + + @classmethod + def _int_field(cls, fields: dict[str, str], *names: str, default: int = 0) -> int: + value = cls._first_field(fields, *names) + try: + return int(value, 0) + except (TypeError, ValueError): + return default + + @staticmethod + def _interface_broadcasts() -> list[str]: + broadcasts = set() + hostname = socket.gethostname() + try: + for item in socket.getaddrinfo(hostname, None, socket.AF_INET): + ip = item[4][0] + parts = ip.split(".") + if len(parts) == IP_OCTET_COUNT and not ip.startswith("127."): + broadcasts.add(".".join([*parts[:3], "255"])) + except OSError: + pass + return sorted(broadcasts) diff --git a/p2pcam/lan_video.py b/p2pcam/lan_video.py new file mode 100644 index 0000000..f51dcfd --- /dev/null +++ b/p2pcam/lan_video.py @@ -0,0 +1,802 @@ +import contextlib +import select +import socket +import struct +import threading +import time +from collections.abc import Iterator +from typing import Self + +# --------------------------------------------------------------------------- +# Protocol constants +# --------------------------------------------------------------------------- + +BROADCAST_PORT = 2627 # LocalData broadcast port (LAN discovery) +VIDEO_PORT = 5000 # UDP port for video exchange + +# XOR obfuscation key for dict body (bytes 2+ are XOR'd with this value) +_BODY_XOR_KEY = 0xE9 + +# Session ID and call ID used in the video handshake. +# These can remain fixed strings since we observed them hard-coded in the SDK. +_DEFAULT_SID = "j882Tm1a108000" +_DEFAULT_CALLID = "I0.JvIZbLTnL7MpGdBuLRVmA1a100ff1" + +# Polling intervals +_POLL_INTERVAL = 0.17 # seconds between ICMD2 polls before stream +_ACK_INTERVAL = 0.17 # seconds between ICMD1 acks during stream + +# MJPEG reassembly +_MAX_UDP_PAYLOAD = 65535 +_MJPEG_HDR_OFFSET = 4 # JPEG SOI is at offset 4 within video UDP payload +_JPEG_SOI = b"\xff\xd8" +_JPEG_EOI = b"\xff\xd9" +_XOR_PREFIX_LEN = 2 +_CONTINUE_LIST_STEP = 2 +_CONTINUE_DIGITS_3 = 3 +_CONTINUE_DIGITS_4 = 4 +_MAX_CONTINUE_DIGITS = 5 +_MIN_SESSION_CREATE_LEN = 14 +_ICMD1_POLL_LEN = 51 +_MJPEG_MIN_LEN = 60 +_PING_ACK_LEN = 13 +_PING_ACK_BYTE = 0x92 + + +# --------------------------------------------------------------------------- +# Low-level packet builders +# --------------------------------------------------------------------------- + + +def _xor_encode(s: str) -> bytes: + """Encode a dict-style string: first 2 bytes plain, rest XOR 0xe9.""" + raw = s.encode("ascii") + if len(raw) <= _XOR_PREFIX_LEN: + return raw + return raw[:_XOR_PREFIX_LEN] + bytes( + b ^ _BODY_XOR_KEY for b in raw[_XOR_PREFIX_LEN:] + ) + + +def _xor_encode_all(s: str) -> bytes: + """Encode an ICMD-style string: ALL bytes XOR 0xe9.""" + return bytes(b ^ _BODY_XOR_KEY for b in s.encode("ascii")) + + +def _build_packet( # noqa: PLR0917 + counter: int, + inner_cmd: int, + inner_flag1: int, + inner_flag2: int, + inner_extra: bytes, + body: bytes, +) -> bytes: + """ + Build a complete packet with the 2-byte counter prefix. + + Layout (verified byte-by-byte against the Android app dumps): + [0-1] packet counter (uint16 LE) + [2-3] outer len field: (total_packet_len << 4) as uint16 LE + [4] inner_cmd + [5] inner_flag1 + [6] inner_flag2 + [7-8] inner payload length (9-byte inner header + body) as uint16 LE + [9-12] inner_extra (4 bytes) + [13+] body + """ + inner_len = 9 + len(body) # 9-byte inner header + body + outer_total = 4 + inner_len # 4-byte outer header + inner + outer_len_field = outer_total << 4 + + pkt = bytearray() + pkt += struct.pack(" bytes: + """Build the first connection ping (``82 0c 00 09 00 d1 07 00 00``).""" + # Exactly as seen in frames 2-4 of the dump, with counter=0x0000 + return bytes.fromhex("0000d000820c000900d1070000") + + +def _build_ping2() -> bytes: + """Build the second connection ping (``a2 0c 40 09 00 d1 07 00 00``).""" + # Exactly as seen in frames 8-16 of the dump, with counter=0x0000 + return bytes.fromhex("0000d000a20c400900d1070000") + + +def _build_hk_res_req( + hkid: int, sid: str = _DEFAULT_SID, callid: str = _DEFAULT_CALLID +) -> bytes: + """ + Build the HK_RES_REQ video-init packet (counter=0x0001). + + Exact layout from frame 17 of the dump: + counter = 0x0001 + inner_cmd = 0x32 + inner_flag1= 0x8B + inner_flag2= 0xC5 + inner_extra= 51 01 00 00 + """ + body_str = ( + f"id={hkid};ftN0=video.vbVideo.MPEG4;ftN1=net.0;" + f"ftN2=HKPCPresent.HKPCPresent;opN2={sid};" + f"Callid={callid};sidN={sid};" + "AsCode=337;MainCmd=HK_RES_REQ;user=Lan user;" + ) + body = _xor_encode(body_str) + return _build_packet( + counter=0x0001, + inner_cmd=0x32, + inner_flag1=0x8B, + inner_flag2=0xC5, + inner_extra=bytes([0x51, 0x01, 0x00, 0x00]), + body=body, + ) + + +def _build_icmd2_poll(hkid: int, seq: int, session_id: int = 293) -> bytes: + """ + Build the 47-byte ICMD2 poll packet sent while waiting for SessionCreate. + + Frame 18 layout: + counter = 0x0000 + inner_cmd = 0x10 + inner_flag1= 0x03 + inner_flag2= 0x00 + inner_extra= 00 d1 07 00 + Body (XOR 0xe9 from byte 2): ``d4:ICMD2:293:SEQ1::GUARDSEQ1:`` + """ + body_str = f"d4:ICMD2:{session_id}:SEQ1:{hkid}:GUARDSEQ1:{seq:02x}" + body = _xor_encode_all(body_str) + bytes([0xE9 ^ 0x00]) # trailing null, XOR'd + return _build_packet( + counter=0x0000, + inner_cmd=0x10, + inner_flag1=0x03, + inner_flag2=0x00, + inner_extra=bytes([0xD1, 0x07, 0x00, 0x00]), + body=body, + ) + + +def _build_session_start(sid: str = _DEFAULT_SID) -> bytes: + """ + Build the SessionStart packet (counter=0x0002). + + Frame 25 layout: + counter = 0x0002 + inner_cmd = 0x32 + inner_flag1= 0x51 + inner_flag2= 0xC4 + inner_extra= 64 00 00 00 + """ + body_str = ( + f"MainCmd=SessionStart;sidN={sid};" + "ftN0=HKPCPresent.HKPCPresent;FD0=4;ftN1=net.1024;FD1=1024;" + ) + body = _xor_encode(body_str) + return _build_packet( + counter=0x0002, + inner_cmd=0x32, + inner_flag1=0x51, + inner_flag2=0xC4, + inner_extra=bytes([0x64, 0x00, 0x00, 0x00]), + body=body, + ) + + +def _build_icmd1_ack(hkid: int, seq: int, session_id: int = 293) -> bytes: + """ + Build the 47-byte ICMD1 ACK sent during streaming. + + counter = 0x0000, inner_cmd = 0x10, inner_flag1 = 0x00, inner_flag2 = 0x00 + Body (XOR 0xe9): ``d4:ICMD1::lastreq1::SEQ3:e`` + """ + body_str = f"d4:ICMD1:{session_id}:lastreq1:{hkid}:SEQ3:{seq:x}e" + body = _xor_encode_all(body_str) + bytes([0xE9 ^ 0x00]) # trailing null, XOR'd + return _build_packet( + counter=0x0000, + inner_cmd=0x10, + inner_flag1=0x00, + inner_flag2=0x00, + inner_extra=bytes([0xD1, 0x07, 0x00, 0x00]), + body=body, + ) + + +# --------------------------------------------------------------------------- +# Continue-packet state machine. +# --------------------------------------------------------------------------- + +_CONTINUE_LIST_1 = bytes( + [ + 0xD8, + 0xDF, # index 0,1 + 0xDB, + 0xDE, # index 2,3 + 0xDA, + 0xD1, # index 4,5 + 0xDD, + 0xD0, # index 6,7 + 0xDC, + 0xD9, # index 8,9 + 0xDF, + 0xD8, # index 10,11 + 0xDE, + 0xDB, # index 12,13 + 0xD1, + 0xDA, # index 14,15 + 0xD0, + 0xDD, # index 16,17 + 0xD9, + 0xDC, # index 18,19 + ] +) +_CONTINUE_LIST_2 = bytes([0xD9, 0xD8, 0xDB, 0xDA, 0xDD, 0xDC, 0xDF, 0xDE, 0xD1, 0xD0]) + +# Fixed prefix (bytes 0-23) and suffix (bytes 25+) of every continue packet. +# Bytes [2] and [7] are overwritten by _ContinueState.next_packet() to signal +# the digit count to the camera. +_CONT_BEGIN = bytearray( + [ + 0x00, + 0x00, + 0xFF, # [2] overwritten: 0x20/0x30/0x40/0x50/0x60 for nbDigits 1..5 + 0x02, + 0x12, + 0x00, + 0x00, + 0xFF, # [7] overwritten: 0x1e/0x1f/0x20/0x21/0x22 for nbDigits 1..5 + 0x00, + 0x01, + 0x00, + 0x00, + 0x00, + 0xA0, + 0xAA, + 0xA4, + 0xAD, + 0xD4, + 0xD8, + 0xD2, + 0xBA, + 0xAC, + 0xB8, + 0xD4, + ] +) +_CONT_END = bytes([0xD2, 0xBD, 0xA0, 0xA4, 0xAC, 0xD4, 0xD9, 0xD2, 0xE9]) + + +class _ContinueState: + """ + Stateful generator for the continue-packet payload sequence. + + Mirrors the MANAGE "CONTINUE" PACKETS SEQUENCE block from jheyman. + Call next_packet() to obtain the next raw UDP bytes to send. + """ + + def __init__(self) -> None: + self.nb_digits: int = 1 + self.idx: list[int] = [0, 0, 0, 0, 0] # continue_index + self.base_index: int = 0 + self._fragment_index: int = 0 # total calls to next_packet() + + def next_packet(self) -> bytes: # noqa: PLR0912, PLR0915 + """Advance the state and return the complete continue packet bytes.""" + self._fragment_index += 1 + + # Rotate base_index every 100 calls (experimentally determined) + if self._fragment_index % 100 == 0: + self.base_index = (self.base_index + 2) % 20 + + hdr = bytearray(_CONT_BEGIN) # mutable copy + tmp = bytearray() + + nd = self.nb_digits + list_len_2 = len(_CONTINUE_LIST_2) + + if nd == 1: + hdr[2] = 0x20 + hdr[7] = 0x1E + tmp.append(_CONTINUE_LIST_1[self.base_index + self.idx[0]]) + self.idx[0] += 1 + if self.idx[0] == _CONTINUE_LIST_STEP: + self.nb_digits += 1 + self.idx[1] = 1 # start at d8 + self.idx[0] = 0 + + elif nd == _CONTINUE_LIST_STEP: + hdr[2] = 0x30 + hdr[7] = 0x1F + tmp.append(_CONTINUE_LIST_2[self.idx[1]]) + tmp.append(_CONTINUE_LIST_1[self.base_index + self.idx[0]]) + self.idx[0] += 1 + if self.idx[0] == _CONTINUE_LIST_STEP: + self.idx[1] += 1 + self.idx[0] = 0 + if self.idx[1] == list_len_2: + self.nb_digits += 1 + self.idx[2] = 1 + self.idx[1] = 0 + self.idx[0] = 0 + + elif nd == _CONTINUE_DIGITS_3: + hdr[2] = 0x40 + hdr[7] = 0x20 + tmp.append(_CONTINUE_LIST_2[self.idx[2]]) + tmp.append(_CONTINUE_LIST_2[self.idx[1]]) + tmp.append(_CONTINUE_LIST_1[self.base_index + self.idx[0]]) + self.idx[0] += 1 + if self.idx[0] == _CONTINUE_LIST_STEP: + self.idx[1] += 1 + self.idx[0] = 0 + if self.idx[1] == list_len_2: + self.idx[2] += 1 + self.idx[1] = 0 + self.idx[0] = 0 + if self.idx[2] == list_len_2: + self.nb_digits += 1 + self.idx[3] = 1 + self.idx[2] = 0 + self.idx[1] = 0 + self.idx[0] = 0 + + elif nd == _CONTINUE_DIGITS_4: + hdr[2] = 0x50 + hdr[7] = 0x21 + tmp.append(_CONTINUE_LIST_2[self.idx[3]]) + tmp.append(_CONTINUE_LIST_2[self.idx[2]]) + tmp.append(_CONTINUE_LIST_2[self.idx[1]]) + tmp.append(_CONTINUE_LIST_1[self.base_index + self.idx[0]]) + self.idx[0] += 1 + if self.idx[0] == _CONTINUE_LIST_STEP: + self.idx[1] += 1 + self.idx[0] = 0 + if self.idx[1] == list_len_2: + self.idx[2] += 1 + self.idx[1] = 0 + self.idx[0] = 0 + if self.idx[2] == list_len_2: + self.idx[3] += 1 + self.idx[2] = 0 + self.idx[1] = 0 + self.idx[0] = 0 + if self.idx[3] == list_len_2: + self.nb_digits += 1 + self.idx[4] = 1 + self.idx[3] = 0 + self.idx[2] = 0 + self.idx[1] = 0 + + elif nd == _MAX_CONTINUE_DIGITS: + hdr[2] = 0x60 + hdr[7] = 0x22 + tmp.append(_CONTINUE_LIST_2[self.idx[4]]) + tmp.append(_CONTINUE_LIST_2[self.idx[3]]) + tmp.append(_CONTINUE_LIST_2[self.idx[2]]) + tmp.append(_CONTINUE_LIST_2[self.idx[1]]) + tmp.append(_CONTINUE_LIST_1[self.base_index + self.idx[0]]) + self.idx[0] += 1 + if self.idx[0] == _CONTINUE_LIST_STEP: + self.idx[1] += 1 + self.idx[0] = 0 + if self.idx[1] == list_len_2: + self.idx[2] += 1 + self.idx[1] = 0 + self.idx[0] = 0 + if self.idx[2] == list_len_2: + self.idx[3] += 1 + self.idx[2] = 0 + self.idx[1] = 0 + self.idx[0] = 0 + if self.idx[3] == list_len_2: + self.idx[4] += 1 + self.idx[3] = 0 + self.idx[2] = 0 + self.idx[1] = 0 + self.idx[0] = 0 + if self.idx[4] == list_len_2: + self.nb_digits = 1 # restart + self.idx = [0, 0, 0, 0, 0] + + # Horrible reverse-engineered condition: restart at 1 digit + if len(tmp) == _MAX_CONTINUE_DIGITS and tmp[:4] == bytes( + [0xDF, 0xDC, 0xDD, 0xD0] + ): + self.nb_digits = 1 + + return bytes(hdr) + bytes(tmp) + _CONT_END + + +def _build_session_delete(sid: str = _DEFAULT_SID) -> bytes: + """ + Build the SessionDelete packet to cleanly stop the stream. + + Frame 558 layout (client->camera, len=60): + counter = 0x0003 + inner_cmd = 0xC0 + inner_flag1= 0x03 + inner_flag2= 0x32 + inner_extra= 1c c7 38 00 + """ + body_str = f"sidN={sid};MainCmd=SessionDelete;coz=;" + body = _xor_encode(body_str) + return _build_packet( + counter=0x0003, + inner_cmd=0x32, + inner_flag1=0x1C, + inner_flag2=0xC7, + inner_extra=bytes([0x64, 0x00, 0x00, 0x00]), + body=body, + ) + + +# --------------------------------------------------------------------------- +# MJPEG frame reassembly +# --------------------------------------------------------------------------- + + +class _FrameAssembler: + """ + Reassemble fragmented JPEG frames from camera UDP packets. + + From the dump, each video UDP payload begins with a 4-byte header: + [0-1] chunk sequence number (uint16 LE, increases per chunk) + [2-3] 0x84 0x3d (magic) + Then at byte 4 onwards is raw JPEG data. A new frame starts when + ``ff d8`` appears at offset 4. The frame ends when ``ff d9`` is seen. + """ + + def __init__(self) -> None: + self._buf: bytearray = bytearray() + self._in_frame: bool = False + + def feed(self, data: bytes) -> bytes | None: + """Feed one raw UDP payload. Returns a complete JPEG if assembled.""" + if len(data) < _MJPEG_HDR_OFFSET + 2: + return None + + payload = data[_MJPEG_HDR_OFFSET:] + frame = None + + if self._in_frame: + self._buf.extend(payload) + # Check if we now have the EOI for the current frame + eoi_idx = self._buf.find(_JPEG_EOI) + if eoi_idx != -1: + frame = bytes(self._buf[: eoi_idx + 2]) + payload = self._buf[eoi_idx + 2 :] + self._buf = bytearray() + self._in_frame = False + + # If we are not in a frame, or we just finished one, look for the next SOI + if not self._in_frame: + soi_idx = payload.find(_JPEG_SOI) + if soi_idx != -1: + self._buf = bytearray(payload[soi_idx:]) + self._in_frame = True + + # Check if this same payload also contains the EOI for the new frame + eoi_idx2 = self._buf.find(_JPEG_EOI) + if eoi_idx2 != -1: + frame = bytes(self._buf[: eoi_idx2 + 2]) + self._buf = bytearray() + self._in_frame = False + + return frame + + +# --------------------------------------------------------------------------- +# Session-layer response parser +# --------------------------------------------------------------------------- + + +def _decode_camera_msg(data: bytes) -> str | None: + """Try to XOR-decode a camera message body (offset 13, key 0xe9).""" + if len(data) < _MIN_SESSION_CREATE_LEN: + return None + body = data[13:] + decoded = body[:2] + bytes(b ^ _BODY_XOR_KEY for b in body[2:]) + return decoded.decode("ascii", "ignore") + + +def _is_session_create(data: bytes) -> bool: + """Return True if this packet is the camera's ``SessionCreate`` response.""" + msg = _decode_camera_msg(data) + if msg is None: + return False + return "SessionCreate" in msg + + +def _is_icmd1_poll(data: bytes) -> bool: + """Return True if this is a camera ICMD1 poll during streaming.""" + # Camera ICMD1 polls are 51 bytes; header starts with 00 00 30 03 + return len(data) == _ICMD1_POLL_LEN and data[2:4] == bytes([0x30, 0x03]) + + +def _is_mjpeg(data: bytes) -> bool: + """ + Return True if this looks like a video data chunk from the camera. + + We just check the packet size to ignore 51-byte ICMD polls. + """ + return len(data) > _MJPEG_MIN_LEN + + +# --------------------------------------------------------------------------- +# Public interface +# --------------------------------------------------------------------------- + + +class LanVideoClient: + """ + Connect to a HeKai/HK P2P camera on the LAN and receive MJPEG frames. + + Usage:: + + client = LanVideoClient("192.168.1.53", hkid=12) + for frame_jpeg in client.stream(timeout=15): + with open("frame.jpg", "wb") as f: + f.write(frame_jpeg) + + Parameters + ---------- + camera_ip: + LAN IP address of the camera (from ``LanScanner``). + hkid: + Device HKID from ``LanScanner``. Sent in handshake packets. + port: + UDP port for video exchange (default 5000). + session_id: + Arbitrary session identifier (default 293, as observed in dumps). + local_port: + Local UDP port to bind. 0 = let the OS choose. + sid: + Session string used in protocol messages (default ``j882Tm1a108000``). + + """ + + def __init__( # noqa: PLR0917 + self, + camera_ip: str, + hkid: int = 0, + port: int = VIDEO_PORT, + session_id: int = 293, + local_port: int = VIDEO_PORT, + sid: str = _DEFAULT_SID, + ) -> None: + """Initialize the LAN video client with the camera handshake settings.""" + self.camera_ip = camera_ip + self.camera_port = port + self.hkid = hkid + self.session_id = session_id + self.local_port = local_port + self.sid = sid + + self._sock: socket.socket | None = None + self._running = False + self._lock = threading.Lock() + self._seq = 0 + + # ------------------------------------------------------------------ + # Context manager support + # ------------------------------------------------------------------ + + def __enter__(self) -> Self: + """Enter the client context by opening the UDP socket.""" + self._open() + return self + + def __exit__(self, *_: object) -> None: + """Exit the context manager and close the client socket.""" + self.close() + + # ------------------------------------------------------------------ + # Public methods + # ------------------------------------------------------------------ + + def stream(self, timeout: float = 60.0) -> Iterator[bytes]: # noqa: PLR0912, PLR0915 + """ + Yield complete JPEG frames from the camera. + + Performs the full handshake, waits for ``SessionCreate``, sends + ``SessionStart``, then enters the receive / ACK loop. + + Parameters + ---------- + timeout: + Seconds to wait for a response at each blocking step before + giving up. + + """ + self._open() + self._running = True + assembler = _FrameAssembler() + + try: + # ------------------------------------------------------- + # Phase 1: Connection pings (mirror frames 2-7 from dump) + # ------------------------------------------------------- + # Send 3x ping-type-1, wait for camera to respond + for _ in range(3): + self._send(_build_ping1()) + time.sleep(0.002) + + self._wait_for_ping_ack(timeout=2.0) + + # Send 9x ping-type-2 (as seen in frames 8-16) + for _ in range(9): + self._send(_build_ping2()) + time.sleep(0.001) + + # ------------------------------------------------------- + # Phase 2: Send HK_RES_REQ + ICMD2 polls, wait for SessionCreate + # ------------------------------------------------------- + self._send( + _build_hk_res_req( + hkid=self.hkid, + sid=self.sid, + ) + ) + + # Send a few immediate ICMD2 polls (as seen in frames 18-20) + for _ in range(3): + self._seq += 1 + self._send(_build_icmd2_poll(self.hkid, self._seq, self.session_id)) + time.sleep(0.001) + + # Wait for SessionCreate with polling to keep camera happy + session_created = self._wait_for_session_create(timeout=timeout) + if not session_created: + return # camera never replied + + # ------------------------------------------------------- + # Phase 3: Send SessionStart + # ------------------------------------------------------- + self._send(_build_session_start(self.sid)) + + # ------------------------------------------------------- + # Phase 4: Streaming loop + # ------------------------------------------------------- + last_poll = time.monotonic() + deadline = time.monotonic() + timeout + fragments_received = 0 + cont_state = _ContinueState() + + while self._running: + now = time.monotonic() + if now > deadline: + break + + # Periodic ICMD1 ACK to keep the session alive + if now - last_poll >= _ACK_INTERVAL: + self._seq += 1 + self._send(_build_icmd1_ack(self.hkid, self._seq, self.session_id)) + last_poll = now + + readable, _, _ = select.select( + [self._sock], [], [], min(0.05, deadline - now) + ) + if not readable: + continue + + data, (src_ip, _) = self._sock.recvfrom(_MAX_UDP_PAYLOAD) + if src_ip != self.camera_ip: + continue + + # Extend deadline whenever the camera speaks to us + deadline = time.monotonic() + timeout + + # Camera ICMD1 poll → reply with ICMD1 ACK immediately + if _is_icmd1_poll(data): + self._seq += 1 + self._send(_build_icmd1_ack(self.hkid, self._seq, self.session_id)) + last_poll = time.monotonic() + continue + + fragments_received += 1 + # Send continue packet every 5 received fragments using the + # faithful port of the reverse-engineered state machine. + if fragments_received % 5 == 0: + self._send(cont_state.next_packet()) + + # Video chunk → try to assemble a full JPEG + if _is_mjpeg(data): + frame = assembler.feed(data) + if frame is not None: + yield frame + + finally: + # Cleanly close the session before exiting + if self._running: + with contextlib.suppress(Exception): + self._send(_build_session_delete(self.sid)) + self._running = False + + def close(self) -> None: + """Stop the stream and close the UDP socket.""" + self._running = False + with self._lock: + if self._sock is not None: + self._sock.close() + self._sock = None + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _open(self) -> None: + with self._lock: + if self._sock is not None: + return + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + sock.bind(("", self.local_port)) + except OSError: + sock.bind(("", 0)) + self._sock = sock + + def _send(self, data: bytes) -> None: + with self._lock: + if self._sock is not None: + self._sock.sendto(data, (self.camera_ip, self.camera_port)) + + def _wait_for_ping_ack(self, timeout: float = 2.0) -> bool: + """Wait for camera's ``92 0c`` ping-ack response.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + readable, _, _ = select.select([self._sock], [], [], min(0.1, remaining)) + if not readable: + continue + data, (src_ip, _) = self._sock.recvfrom(_MAX_UDP_PAYLOAD) + if src_ip != self.camera_ip: + continue + # Camera ping-ack: 13 bytes starting with 00 00 d0 00 92 + if len(data) == _PING_ACK_LEN and data[4] == _PING_ACK_BYTE: + return True + return False + + def _wait_for_session_create(self, timeout: float = 10.0) -> bool: + """Poll the camera with ICMD2 while waiting for ``SessionCreate``.""" + deadline = time.monotonic() + timeout + last_poll = 0.0 + + while time.monotonic() < deadline: + now = time.monotonic() + remaining = deadline - now + + # Keep sending ICMD2 polls + if now - last_poll >= _POLL_INTERVAL: + self._seq += 1 + self._send(_build_icmd2_poll(self.hkid, self._seq, self.session_id)) + last_poll = now + + readable, _, _ = select.select([self._sock], [], [], min(0.05, remaining)) + if not readable: + continue + + data, (src_ip, _) = self._sock.recvfrom(_MAX_UDP_PAYLOAD) + if src_ip != self.camera_ip: + continue + + if _is_session_create(data): + return True + + # If we get an ICMD1 poll, reply to keep camera happy + if _is_icmd1_poll(data): + self._seq += 1 + self._send(_build_icmd1_ack(self.hkid, self._seq, self.session_id)) + + return False diff --git a/p2pcam/mjpeg_server.py b/p2pcam/mjpeg_server.py new file mode 100644 index 0000000..31fb815 --- /dev/null +++ b/p2pcam/mjpeg_server.py @@ -0,0 +1,81 @@ +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from socketserver import ThreadingMixIn + + +class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): + """Handle requests in a separate thread.""" + + daemon_threads = True + + +class MJPEGServer: + """A class to stream bytes frames to a HTTP MJPEG server.""" + + def __init__(self, port: int = 8080) -> None: + """Initialize a HTTP MJPEG server on localhost:8080 by default.""" + self.port = port + self.latest_frame = None + self.frame_condition = threading.Condition() + self.server = None + self.server_thread = None + + def update_frame(self, frame: bytes) -> None: + """Update the current frame and notify all connected clients.""" + with self.frame_condition: + self.latest_frame = frame + self.frame_condition.notify_all() + + def start(self) -> None: + """Start the HTTP server in a background thread.""" + server_instance = self + + class StreamingHandler(BaseHTTPRequestHandler): + def log_message(self, format, *args): # noqa: A002, ANN002 + # Suppress default HTTP logging to avoid spamming the console + pass + + def do_GET(self): + if self.path == "/stream": + self.send_response(200) + self.send_header( + "Content-type", + "multipart/x-mixed-replace; boundary=--jpgboundary", + ) + self.end_headers() + + while True: + try: + with server_instance.frame_condition: + # Wait for a new frame to be available + server_instance.frame_condition.wait(timeout=1.0) + frame = server_instance.latest_frame + + if frame is None: + continue + + self.wfile.write(b"--jpgboundary\r\n") + self.send_header("Content-type", "image/jpeg") + self.send_header("Content-length", str(len(frame))) + self.end_headers() + self.wfile.write(frame) + self.wfile.write(b"\r\n") + except Exception: # noqa: BLE001 + # Client disconnected (e.g. Broken pipe) + break + else: + self.send_response(404) + self.end_headers() + + self.server = ThreadedHTTPServer(("", self.port), StreamingHandler) + self.server_thread = threading.Thread( + target=self.server.serve_forever, daemon=True + ) + self.server_thread.start() + print(f"MJPEG server started at http://localhost:{self.port}/stream") + + def stop(self) -> None: + """Stop the HTTP server cleanly.""" + if self.server: + self.server.shutdown() + self.server.server_close() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b37d131 Binary files /dev/null and b/requirements.txt differ diff --git a/setup.py b/setup.py index 0997211..c15299e 100644 --- a/setup.py +++ b/setup.py @@ -1,11 +1,13 @@ +from pathlib import Path + import setuptools -with open("README.md", "r") as fh: +with Path.open("README.md") as fh: long_description = fh.read() setuptools.setup( name="p2pcam", - version="0.0.3", + version="1.0.0", author="IndyKoning", description="A package to talk to p2p cameras", long_description=long_description,