Created
April 16, 2026 16:50
-
-
Save Tronix117/e7b50274bb2323c39566696eaafb3008 to your computer and use it in GitHub Desktop.
PTZ ONVIF Proxy for Frigate for camera missing STOP onvif ptz action (like Arenti baby cam)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| ONVIF PTZ Proxy — single process, multiple cameras. | |
| Workaround for cameras (Arenti, some Tuya-based) that declare PTZ support | |
| via ONVIF but don't implement Stop or GetStatus commands. | |
| Intercepts: | |
| Stop → rewrites SOAP body to ContinuousMove(0,0,0) | |
| GetStatus → returns synthetic IDLE response | |
| * → transparent forward with URL rewriting | |
| Config: | |
| CAMERAS proxy_port:cam_host:cam_port[,...] | |
| LOG_LEVEL DEBUG | INFO (default) | WARNING | |
| """ | |
| import os, re, logging, threading | |
| from http.server import HTTPServer, BaseHTTPRequestHandler | |
| from urllib.request import Request, urlopen | |
| from urllib.error import URLError | |
| LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper() | |
| logging.basicConfig(level=getattr(logging, LOG_LEVEL, logging.INFO), | |
| format="%(asctime)s [%(levelname)s] %(message)s") | |
| log = logging.getLogger("onvif-proxy") | |
| # --- SOAP helpers --- | |
| BODY_RE = re.compile(r"<([\w-]*:?)Body\b.*?</([\w-]*:?)Body>", re.DOTALL) | |
| ACTION_RE = re.compile(r"<[\w-]*:?(Stop|ContinuousMove|AbsoluteMove|RelativeMove|GetStatus)\b") | |
| TOKEN_RE = re.compile(r"<[\w-]*:?ProfileToken>([^<]+)</") | |
| GET_STATUS_RESPONSE = ( | |
| '<?xml version="1.0" encoding="UTF-8"?>' | |
| '<s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope">' | |
| "<s:Body>" | |
| '<GetStatusResponse xmlns="http://www.onvif.org/ver20/ptz/wsdl">' | |
| "<PTZStatus>" | |
| '<Position><PanTilt xmlns="http://www.onvif.org/ver10/schema" x="0" y="0"/>' | |
| '<Zoom xmlns="http://www.onvif.org/ver10/schema" x="0"/></Position>' | |
| "<MoveStatus>" | |
| '<PanTilt xmlns="http://www.onvif.org/ver10/schema">IDLE</PanTilt>' | |
| '<Zoom xmlns="http://www.onvif.org/ver10/schema">IDLE</Zoom>' | |
| "</MoveStatus>" | |
| "</PTZStatus>" | |
| "</GetStatusResponse>" | |
| "</s:Body></s:Envelope>" | |
| ) | |
| def rewrite_stop_body(body, profile): | |
| """Replace <Body>…Stop…</Body> with ContinuousMove(0,0). | |
| Preserves Envelope, xmlns declarations, and WS-Security Header.""" | |
| m = BODY_RE.search(body) | |
| if not m: | |
| return body | |
| prefix = m.group(1) | |
| replacement = ( | |
| f"<{prefix}Body>" | |
| f'<ContinuousMove xmlns="http://www.onvif.org/ver20/ptz/wsdl">' | |
| f"<ProfileToken>{profile}</ProfileToken>" | |
| f"<Velocity>" | |
| f'<PanTilt xmlns="http://www.onvif.org/ver10/schema" x="0" y="0"/>' | |
| f'<Zoom xmlns="http://www.onvif.org/ver10/schema" x="0"/>' | |
| f"</Velocity>" | |
| f"</ContinuousMove>" | |
| f"</{prefix}Body>" | |
| ) | |
| return BODY_RE.sub(replacement, body, count=1) | |
| # --- HTTP handler --- | |
| def make_handler(camera_url, camera_host, camera_port, proxy_port): | |
| class Handler(BaseHTTPRequestHandler): | |
| def do_POST(self): | |
| raw = self.rfile.read(int(self.headers.get("Content-Length", 0))) | |
| body = raw.decode("utf-8") | |
| action = (ACTION_RE.search(body) or type("", (), {"group": lambda s, n: "other"})()).group(1) | |
| token = (TOKEN_RE.search(body) or type("", (), {"group": lambda s, n: "PROFILE_000"})()).group(1) | |
| if action == "Stop": | |
| log.info(f":{proxy_port} Stop → ContinuousMove(0,0)") | |
| self._send_to_camera(rewrite_stop_body(body, token)) | |
| elif action == "GetStatus": | |
| log.info(f":{proxy_port} GetStatus → IDLE") | |
| self._respond(200, GET_STATUS_RESPONSE) | |
| else: | |
| log.debug(f":{proxy_port} {action} → forward") | |
| self._send_to_camera(body, rewrite_urls=True) | |
| def _send_to_camera(self, body, rewrite_urls=False): | |
| try: | |
| req = Request(f"{camera_url}{self.path}", | |
| data=body.encode(), method="POST", | |
| headers={"Content-Type": "application/soap+xml"}) | |
| with urlopen(req, timeout=10) as resp: | |
| data = resp.read().decode("utf-8") | |
| if rewrite_urls: | |
| host = self.headers.get("Host", f"127.0.0.1:{proxy_port}") | |
| data = data.replace(f"http://{camera_host}:{camera_port}", | |
| f"http://{host}") | |
| self._respond(200, data) | |
| except URLError as e: | |
| log.error(f":{proxy_port} forward error: {e}") | |
| self._respond(502, str(e)) | |
| def _respond(self, code, body): | |
| self.send_response(code) | |
| self.send_header("Content-Type", "application/soap+xml") | |
| self.end_headers() | |
| self.wfile.write(body.encode()) | |
| def log_message(self, *_): | |
| pass | |
| return Handler | |
| # --- Main --- | |
| def serve(proxy_port, cam_host, cam_port): | |
| url = f"http://{cam_host}:{cam_port}" | |
| HTTPServer(("0.0.0.0", proxy_port), | |
| make_handler(url, cam_host, cam_port, proxy_port)).serve_forever() | |
| if __name__ == "__main__": | |
| entries = os.environ.get("CAMERAS", "").strip() | |
| if not entries: | |
| log.error("CAMERAS env var required (e.g. 8001:192.168.1.10:8000,8002:192.168.1.11:8000)") | |
| raise SystemExit(1) | |
| cameras = [] | |
| for e in entries.split(","): | |
| p = e.strip().split(":") | |
| cameras.append((int(p[0]), p[1], int(p[2]))) | |
| log.info(f"Starting {len(cameras)} proxy(ies)") | |
| threads = [] | |
| for port, host, cam_port in cameras: | |
| log.info(f" :{port} → {host}:{cam_port}") | |
| t = threading.Thread(target=serve, args=(port, host, cam_port), daemon=True) | |
| t.start() | |
| threads.append(t) | |
| for t in threads: | |
| t.join() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment