import http.client import json import mimetypes import os import re import shutil import socket import subprocess import sys import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse from configurable_udp_capture import ( DEFAULT_PACKET_LAYOUT, DEFAULT_PACKET_SCHEMA, normalize_packet_layout, normalize_packet_schema, packet_layout_from_schema, packet_schema_from_layout, ) from udp_probe import run_udp_probe APP_DIR = Path(__file__).resolve().parent DATA_DIR = Path(os.environ.get("FPV_DATA_DIR", "/data")) HOST = os.environ.get("FPV_UI_HOST", "127.0.0.1") PORT = int(os.environ.get("FPV_UI_PORT", "8080")) MAX_JSON_BODY_BYTES = 1 * 1024 * 1024 LOG_PATH = Path(os.environ.get("FPV_UI_LOG_PATH", str(DATA_DIR / "logs" / "main.log"))) FRAME_PATH = Path(os.environ.get("FPV_UI_FRAME_PATH", str(DATA_DIR / "ui" / "latest.jpg"))) GUIDANCE_PATH = Path(os.environ.get("FPV_UI_GUIDANCE_PATH", str(DATA_DIR / "guidance" / "guidance_state.json"))) OUT_DIR = Path(os.environ.get("FPV_UI_OUT_DIR", str(DATA_DIR / "out"))) DOWNLOAD_DIR = Path(os.environ.get("FPV_UI_DOWNLOAD_DIR", str(OUT_DIR / ".downloads"))) ACTIVE_VIDEO_PATH = Path(os.environ.get("FPV_UI_ACTIVE_VIDEO_PATH", str(OUT_DIR / ".active_video"))) INPUT_DIR = Path(os.environ.get("FPV_UI_INPUT_DIR", str(DATA_DIR / "input"))) UDP_PROBE_DIR = Path(os.environ.get("FPV_UI_UDP_PROBE_DIR", str(DATA_DIR / "udp-probes"))) CONTROL_PATH = Path(os.environ.get("FPV_UI_CONTROL_PATH", str(DATA_DIR / "ui" / "control_state.json"))) MAIN_SCRIPT = Path(os.environ.get("FPV_UI_MAIN_SCRIPT", str(APP_DIR / "main.py"))) PYTHON_EXE = os.environ.get("FPV_UI_PYTHON", sys.executable) MODEL_PATH = os.environ.get("FPV_MODEL_PATH", str(APP_DIR / "best.pt")) MODEL_DIR = Path(os.environ.get("FPV_UI_MODEL_DIR", str(DATA_DIR / "models"))) MODEL_EXTENSIONS = {".pt"} MODEL_INFO_LOCK = threading.Lock() MODEL_INFO_CACHE = {} NETRON_HOST = "127.0.0.1" NETRON_PORT = int(os.environ.get("FPV_NETRON_PORT", "8092")) NETRON_LOCK = threading.Lock() NETRON_STATE = {"path": ""} CONTROL_LOCK = threading.Lock() UDP_PROBE_LOCK = threading.Lock() CONTROL_PROCESS = None CONTROL_SOURCE = "" QUALITIES = [ {"label": "4K 3840x2160", "width": 3840, "height": 2160}, {"label": "QHD 2560x1440", "width": 2560, "height": 1440}, {"label": "Full HD 1920x1080", "width": 1920, "height": 1080}, {"label": "HD 1280x720", "width": 1280, "height": 720}, {"label": "XGA 1024x768", "width": 1024, "height": 768}, {"label": "PAL 720x576", "width": 720, "height": 576}, {"label": "VGA 640x480", "width": 640, "height": 480}, {"label": "UDP-камера 512x640", "width": 512, "height": 640}, ] FPS_OPTIONS = [120, 60, 50, 30, 25, 24, 15] MIN_FRAME_SIZE = 16 MAX_FRAME_SIZE = 8192 MIN_CAPTURE_FPS = 1 MAX_CAPTURE_FPS = 240 VIDEO_EXTENSIONS = { ".3g2", ".3gp", ".264", ".265", ".asf", ".avi", ".divx", ".dv", ".f4v", ".flv", ".h264", ".h265", ".hevc", ".m2t", ".m2ts", ".m4v", ".mkv", ".mov", ".mp4", ".mpeg", ".mpg", ".mts", ".mxf", ".ogm", ".ogv", ".rm", ".rmvb", ".ts", ".vob", ".webm", ".wmv", ".y4m", } UDP_DUMP_EXTENSIONS = {"", ".udp", ".dump"} INPUT_EXTENSIONS = VIDEO_EXTENSIONS | UDP_DUMP_EXTENSIONS ERROR_PROTOCOLS = ["guidance_v1", "json", "csv", "bin"] ERROR_UNITS = ["px", "norm", "deg", "m"] ARCHIVE_RECORD_MODES = ["fragments", "full"] FILE_SOURCE_MODES = {"file", "udp_dump", "udp_delimited_file"} LIVE_SOURCE_MODES = {"udp_mik_live", "udp_delimited_live", "udp_custom_live"} SOURCE_MODES = {"camera"} | FILE_SOURCE_MODES | LIVE_SOURCE_MODES FRAME_ENCODINGS = {"auto", "bgr24", "rgb24", "gray8", "gray16", "yuyv422"} PACKET_PRESETS = {"auto", "mik", "delimited", "custom"} PACKET_PRESET_BY_SOURCE = { "udp_mik_live": "mik", "udp_delimited_live": "delimited", "udp_custom_live": "custom", } H264_CACHE_VERSION = "v2" DEFAULT_ERROR_HOST = os.environ.get("FPV_ERROR_OUTPUT_HOST", "127.0.0.1" if os.name == "nt" else "host.docker.internal") DEFAULT_CONTROL = { "model_path": MODEL_PATH, "device": 0, "use_half": True, "conf": 0.25, "img_size_roi": 640, "img_size_full": 1280, "max_det": 60, "source_mode": "file", "camera_index": 0, "file_path": str(INPUT_DIR / "source.mp4"), "input_host": "0.0.0.0", "input_port": 59004, "separator_byte": 0, "frame_encoding": "auto", "packet_preset": "auto", "packet_layout": [field.copy() for field in DEFAULT_PACKET_LAYOUT], "packet_schema": DEFAULT_PACKET_SCHEMA.copy(), "quality": "1280x720", "fps": 30, "run_mode": "realtime", "frame_mode": "hd", "save": True, "archive_mode": os.environ.get("FPV_ARCHIVE_RECORD_MODE", "fragments"), "fragment_gap_sec": float(os.environ.get("FPV_DETECTION_CLIP_MAX_GAP_SEC", "15")), "error_output": True, "error_protocol": os.environ.get("FPV_ERROR_OUTPUT_PROTOCOL", "guidance_v1"), "error_host": DEFAULT_ERROR_HOST, "error_port": int(os.environ.get("FPV_ERROR_OUTPUT_PORT", "5010")), "error_object_id": int(os.environ.get("FPV_ERROR_OUTPUT_OBJECT_ID", "1")), "error_units": os.environ.get("FPV_ERROR_OUTPUT_UNITS", "px"), "error_hfov": float(os.environ.get("FPV_ERROR_OUTPUT_HFOV_DEG", "90")), "error_vfov": float(os.environ.get("FPV_ERROR_OUTPUT_VFOV_DEG", "60")), "error_range_m": float(os.environ.get("FPV_ERROR_OUTPUT_RANGE_M", "0")), } def tail_text(path, lines=80): if not path.exists(): return "" data = path.read_bytes()[-65536:] text = data.decode("utf-8", errors="replace") return "\n".join(text.splitlines()[-int(lines):]) def read_json_file(path): if not path.exists(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} def latest_file(root, pattern): if not root.exists(): return None files = [p for p in root.glob(pattern) if p.is_file()] if not files: return None return max(files, key=lambda p: p.stat().st_mtime) def archive_path(root, name): decoded = unquote(str(name)) safe_name = Path(decoded).name if safe_name != decoded or not safe_name.lower().endswith(".mp4"): return None path = root / safe_name try: path.resolve(strict=False).relative_to(Path(root).resolve()) except (OSError, ValueError): return None return path def udp_probe_path(root, name): decoded = unquote(str(name)) safe_name = Path(decoded).name if safe_name != decoded or Path(safe_name).suffix.lower() not in {".udp", ".json"}: return None path = root / safe_name try: path.resolve(strict=False).relative_to(Path(root).resolve()) except (OSError, ValueError): return None return path def archive_files(root, active_name=None): if not root.exists(): return [] rows = [] for path in sorted(root.glob("*.mp4"), key=lambda p: p.stat().st_mtime, reverse=True): stat = path.stat() rows.append({ "name": path.name, "size": stat.st_size, "mtime": int(stat.st_mtime), "active": path.name == active_name, }) return rows def cleanup_empty_recordings(root, max_bytes=128): removed = 0 if not root.exists(): return removed for path in root.glob("*.mp4"): try: if path.stat().st_size <= int(max_bytes): path.unlink() removed += 1 except OSError: pass return removed def active_video_name(root, marker_path=None): marker_path = ACTIVE_VIDEO_PATH if marker_path is None else marker_path try: name = marker_path.read_text(encoding="utf-8").strip() except OSError: return None path = archive_path(root, name) if path and path.exists(): return path.name return None def safe_download_filename(name): filename = Path(str(name)).name ascii_name = re.sub(r"[^A-Za-z0-9_.-]+", "_", filename).strip("._") return ascii_name or "video.mp4" def content_disposition(filename): ascii_name = safe_download_filename(filename) encoded = quote(Path(str(filename)).name) return f'attachment; filename="{ascii_name}"; filename*=UTF-8\'\'{encoded}' def h264_download_name(path): return f"{path.stem}_h264.mp4" def h264_cache_path(path, cache_dir=DOWNLOAD_DIR): stat = path.stat() stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", path.stem).strip("._") or "video" return cache_dir / f"{stem}.{H264_CACHE_VERSION}.{stat.st_mtime_ns}.{stat.st_size}.h264.mp4" def delete_h264_cache(path, cache_dir=DOWNLOAD_DIR): stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", path.stem).strip("._") or "video" if not cache_dir.exists(): return for cached in cache_dir.glob(f"{stem}.*.h264.mp4"): try: cached.unlink() except OSError: pass def h264_download_file(path): ffmpeg = ffmpeg_executable() if not ffmpeg: raise RuntimeError("ffmpeg unavailable") cache = h264_cache_path(path) if cache.exists() and cache.stat().st_size > 0: return cache cache.parent.mkdir(parents=True, exist_ok=True) temp = cache.with_name(f"{cache.name}.{threading.get_ident()}.tmp") command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-y", "-i", str(path), "-an", "-c:v", "libx264", "-preset", "veryfast", "-crf", "18", "-pix_fmt", "yuv420p", "-movflags", "+faststart", "-f", "mp4", str(temp), ] try: result = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True) if result.returncode != 0: raise RuntimeError((result.stderr or "ffmpeg failed").strip()) temp.replace(cache) return cache finally: try: temp.unlink() except OSError: pass def write_json_file(path, data): try: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") except OSError: pass def read_control_state(): state = DEFAULT_CONTROL.copy() data = read_json_file(CONTROL_PATH) if isinstance(data, dict): state.update({k: v for k, v in data.items() if k in state}) if "packet_preset" not in data: state["packet_preset"] = PACKET_PRESET_BY_SOURCE.get(state.get("source_mode"), "auto") if "packet_layout" not in data and "packet_schema" in data: state["packet_layout"] = packet_layout_from_schema(data["packet_schema"]) if state.get("source_mode") in FILE_SOURCE_MODES: path = resolve_video_path(state.get("file_path")) default_path = INPUT_DIR / "source.mp4" if not path.is_file() and default_path.is_file(): state["file_path"] = str(default_path) return state def looks_like_mik_dump(path): try: size = path.stat().st_size with path.open("rb") as stream: header = stream.read(12) except OSError: return False if size < 12 or len(header) < 12: return False port = int.from_bytes(header[:2], "little") payload_size = int.from_bytes(header[2:4], "little") flags = header[5] packet_number = header[7] array_size = int.from_bytes(header[8:12], "little") return ( 0 < port <= 65535 and 8 <= payload_size <= 65507 and size >= 4 + payload_size and flags & 0x02 and not flags & ~0x03 and packet_number == 0 and 0 < array_size <= 256 * 1024 * 1024 ) def input_file_kind(path): if path.suffix.lower() in {".udp", ".dump"} or looks_like_mik_dump(path): return "udp_dump" return "video" def input_video_files(): if not INPUT_DIR.exists(): return [] rows = [] for path in sorted(INPUT_DIR.iterdir(), key=lambda p: p.stat().st_mtime if p.exists() else 0, reverse=True): if not path.is_file() or path.name.startswith("."): continue stat = path.stat() rows.append({ "name": path.name, "path": str(path), "size": stat.st_size, "mtime": int(stat.st_mtime), "kind": input_file_kind(path), }) return rows def parse_quality(value): match = re.fullmatch(r"(\d{1,5})\s*[xх×]\s*(\d{1,5})", str(value or "").strip().lower()) if not match: return 1280, 720 width = max(MIN_FRAME_SIZE, min(MAX_FRAME_SIZE, int(match.group(1)))) height = max(MIN_FRAME_SIZE, min(MAX_FRAME_SIZE, int(match.group(2)))) return width, height def camera_bridge_source(base_url, camera_index, width, height, fps): separator = "&" if "?" in base_url else "?" query = urlencode({ "index": int(camera_index), "width": int(width), "height": int(height), "fps": int(fps), "quality": int(os.environ.get("FPV_CAMERA_BRIDGE_JPEG_QUALITY", "85")), }) return f"{base_url}{separator}{query}" def resolve_video_path(value): path = Path(str(value or "").strip().strip('"')) if not str(path): path = INPUT_DIR / "source.mp4" if not path.is_absolute(): path = APP_DIR / path return path def normalize_control(data): state = read_control_state() if isinstance(data, dict): state.update({k: v for k, v in data.items() if k in state}) model_path = resolve_model_path(state.get("model_path")) state["model_path"] = str(model_path or resolve_model_path(MODEL_PATH) or MODEL_PATH) try: state["device"] = max(-1, min(16, int(state.get("device", 0)))) except (TypeError, ValueError): state["device"] = 0 raw_half = state.get("use_half", True) state["use_half"] = raw_half if isinstance(raw_half, bool) else str(raw_half).strip().lower() in {"1", "true", "yes", "on"} try: state["conf"] = max(0.01, min(1.0, float(state.get("conf", 0.25)))) except (TypeError, ValueError): state["conf"] = 0.25 for key, default in (("img_size_roi", 640), ("img_size_full", 1280)): try: state[key] = max(128, min(4096, int(state.get(key, default)))) except (TypeError, ValueError): state[key] = default try: state["max_det"] = max(1, min(300, int(state.get("max_det", 60)))) except (TypeError, ValueError): state["max_det"] = 60 source_mode = str(state.get("source_mode") or "file").lower() state["source_mode"] = source_mode if source_mode in SOURCE_MODES else "file" state["camera_index"] = max(0, int(state.get("camera_index") or 0)) state["file_path"] = str(resolve_video_path(state.get("file_path"))) state["input_host"] = str(state.get("input_host") or "0.0.0.0").strip() or "0.0.0.0" state["input_port"] = int(max(1, min(65535, int(state.get("input_port") or 59004)))) state["separator_byte"] = int(max(0, min(255, int(state.get("separator_byte") or 0)))) state["frame_encoding"] = str(state.get("frame_encoding") or "auto").lower() if state["frame_encoding"] not in FRAME_ENCODINGS: state["frame_encoding"] = "auto" state["packet_preset"] = str(state.get("packet_preset") or "auto").lower() if isinstance(data, dict) and "source_mode" in data and "packet_preset" not in data: state["packet_preset"] = PACKET_PRESET_BY_SOURCE.get(state["source_mode"], "auto") if state["packet_preset"] not in PACKET_PRESETS: state["packet_preset"] = "auto" if isinstance(data, dict) and "packet_schema" in data and "packet_layout" not in data: state["packet_layout"] = packet_layout_from_schema(state["packet_schema"]) state["packet_layout"] = normalize_packet_layout(state.get("packet_layout")) state["packet_schema"] = packet_schema_from_layout( state["packet_layout"], normalize_packet_schema(state.get("packet_schema")), ) width, height = parse_quality(state.get("quality")) state["quality"] = f"{width}x{height}" fps = int(float(state.get("fps") or 30)) state["fps"] = max(MIN_CAPTURE_FPS, min(MAX_CAPTURE_FPS, fps)) state["run_mode"] = "fast" if state.get("run_mode") == "fast" else "realtime" if state["source_mode"] in LIVE_SOURCE_MODES: state["run_mode"] = "realtime" state["frame_mode"] = "pal" if state.get("frame_mode") == "pal" else "hd" state["save"] = bool(state.get("save", True)) state["archive_mode"] = str(state.get("archive_mode") or "fragments").lower() if state["archive_mode"] not in ARCHIVE_RECORD_MODES: state["archive_mode"] = "fragments" state["fragment_gap_sec"] = float(max(0.0, min(3600.0, float(state.get("fragment_gap_sec") or 15.0)))) state["error_output"] = bool(state.get("error_output", False)) state["error_protocol"] = str(state.get("error_protocol") or "guidance_v1").lower() if state["error_protocol"] not in ERROR_PROTOCOLS: state["error_protocol"] = "guidance_v1" state["error_units"] = str(state.get("error_units") or "px").lower() if state["error_units"] not in ERROR_UNITS: state["error_units"] = "px" state["error_host"] = str(state.get("error_host") or DEFAULT_ERROR_HOST).strip() or DEFAULT_ERROR_HOST state["error_port"] = int(max(1, min(65535, int(state.get("error_port") or 5010)))) state["error_object_id"] = int(max(1, min(255, int(state.get("error_object_id") or 1)))) state["error_hfov"] = float(max(1.0, min(179.0, float(state.get("error_hfov") or 90.0)))) state["error_vfov"] = float(max(1.0, min(179.0, float(state.get("error_vfov") or 60.0)))) state["error_range_m"] = float(max(0.0, float(state.get("error_range_m") or 0.0))) return state def process_running(): global CONTROL_PROCESS if CONTROL_PROCESS is None: return False if CONTROL_PROCESS.poll() is None: return True CONTROL_PROCESS = None return False def control_payload(): state = read_control_state() running = process_running() active_name = active_video_name(OUT_DIR) if running else None if not running: try: ACTIVE_VIDEO_PATH.unlink(missing_ok=True) except OSError: pass return { **state, "running": running, "pid": CONTROL_PROCESS.pid if CONTROL_PROCESS and CONTROL_PROCESS.poll() is None else None, "active_video": active_name or "", "qualities": QUALITIES, "fps_options": FPS_OPTIONS, "error_protocols": ERROR_PROTOCOLS, "error_units_options": ERROR_UNITS, "archive_record_modes": ARCHIVE_RECORD_MODES, "default_input": str(INPUT_DIR / "source.mp4"), "default_model": str(resolve_model_path(MODEL_PATH) or MODEL_PATH), } def stop_control_process(): global CONTROL_PROCESS, CONTROL_SOURCE if CONTROL_PROCESS is None or CONTROL_PROCESS.poll() is not None: CONTROL_PROCESS = None CONTROL_SOURCE = "" try: ACTIVE_VIDEO_PATH.unlink(missing_ok=True) except OSError: pass return False pid = CONTROL_PROCESS.pid if os.name == "nt": subprocess.run( ["taskkill", "/PID", str(pid), "/T", "/F"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) else: CONTROL_PROCESS.terminate() try: CONTROL_PROCESS.wait(timeout=5) except subprocess.TimeoutExpired: CONTROL_PROCESS.kill() CONTROL_PROCESS.wait(timeout=5) CONTROL_PROCESS = None CONTROL_SOURCE = "" try: ACTIVE_VIDEO_PATH.unlink(missing_ok=True) except OSError: pass return True def start_control_process(data): global CONTROL_PROCESS, CONTROL_SOURCE state = normalize_control(data) if state["source_mode"] in FILE_SOURCE_MODES and not Path(state["file_path"]).is_file(): raise FileNotFoundError(state["file_path"]) with CONTROL_LOCK: if UDP_PROBE_LOCK.locked(): raise RuntimeError("UDP probe is active") stop_control_process() DATA_DIR.mkdir(parents=True, exist_ok=True) INPUT_DIR.mkdir(parents=True, exist_ok=True) OUT_DIR.mkdir(parents=True, exist_ok=True) LOG_PATH.parent.mkdir(parents=True, exist_ok=True) FRAME_PATH.parent.mkdir(parents=True, exist_ok=True) GUIDANCE_PATH.parent.mkdir(parents=True, exist_ok=True) for path in (FRAME_PATH, GUIDANCE_PATH, ACTIVE_VIDEO_PATH): try: path.unlink(missing_ok=True) except OSError: pass width, height = parse_quality(state["quality"]) env = os.environ.copy() process_source_mode = state["source_mode"] if state["source_mode"] == "camera": bridge_url = os.environ.get("FPV_CAMERA_BRIDGE_URL", "").strip() if bridge_url: source = camera_bridge_source( bridge_url, state["camera_index"], width, height, state["fps"], ) process_source_mode = "camera_bridge" else: source = str(state["camera_index"]) elif state["source_mode"] in LIVE_SOURCE_MODES: source = f"udp://{state['input_host']}:{state['input_port']}" else: source = str(state["file_path"]) env.update({ "PYTHONUNBUFFERED": "1", "FPV_MODEL_PATH": state["model_path"], "FPV_DEVICE": str(state["device"]), "FPV_USE_HALF": "1" if state["use_half"] else "0", "FPV_CONF": str(state["conf"]), "FPV_IMG_SIZE_ROI": str(state["img_size_roi"]), "FPV_IMG_SIZE_FULL": str(state["img_size_full"]), "FPV_MAX_DET": str(state["max_det"]), "FPV_SOURCE": source, "FPV_SOURCE_MODE": process_source_mode, "FPV_UDP_INPUT_HOST": state["input_host"], "FPV_UDP_INPUT_PORT": str(state["input_port"]), "FPV_FRAME_SEPARATOR_BYTE": str(state["separator_byte"]), "FPV_FRAME_ENCODING": state["frame_encoding"], "FPV_UDP_PACKET_SCHEMA": json.dumps(state["packet_schema"], separators=(",", ":")), "FPV_CAP_BACKEND": os.environ.get("FPV_CAP_BACKEND", "dshow" if os.name == "nt" else "v4l2"), "FPV_CAMERA_WIDTH": str(width), "FPV_CAMERA_HEIGHT": str(height), "FPV_CAMERA_FPS": str(state["fps"]), "FPV_CAMERA_FOURCC": "MJPG", "FPV_TARGET_OUT_FPS": str(state["fps"]) if state["source_mode"] == "camera" else "0", "FPV_VIDEO_REALTIME": "0" if state["run_mode"] == "fast" else "1", "FPV_SHOW_OUTPUT": "0", "FPV_SAVE_INFER_VIDEO": "1" if state["save"] else "0", "FPV_OUT_VIDEO_PATH": str(OUT_DIR / "out_infer.mp4"), "FPV_ARCHIVE_RECORD_MODE": state["archive_mode"], "FPV_DETECTION_CLIP_MAX_GAP_SEC": str(state["fragment_gap_sec"]), "FPV_FORCE_EFFECTIVE_PAL": "1" if state["frame_mode"] == "pal" else "0", "FPV_EFFECTIVE_W": str(width), "FPV_EFFECTIVE_H": str(height), "FPV_UI_FRAME_EXPORT_ENABLE": "1", "FPV_UI_FRAME_EXPORT_PATH": str(FRAME_PATH), "FPV_UI_FRAME_EXPORT_EVERY": "1", "FPV_UI_FRAME_EXPORT_JPEG_QUALITY": os.environ.get("FPV_UI_FRAME_EXPORT_JPEG_QUALITY", "82"), "FPV_UI_FRAME_EXPORT_MAX_FPS": os.environ.get("FPV_UI_FRAME_EXPORT_MAX_FPS", "50"), "FPV_REALTIME_SKIP_STALE_FRAMES": os.environ.get("FPV_REALTIME_SKIP_STALE_FRAMES", "1"), "FPV_REALTIME_MAX_SKIP_FRAMES": os.environ.get("FPV_REALTIME_MAX_SKIP_FRAMES", "8"), "FPV_REALTIME_PREVIEW_SKIPPED_FRAMES": os.environ.get("FPV_REALTIME_PREVIEW_SKIPPED_FRAMES", "1"), "FPV_REALTIME_ANALYSIS_EVERY": os.environ.get("FPV_REALTIME_ANALYSIS_EVERY", "4"), "FPV_GUIDANCE_EXPORT_ENABLE": "1", "FPV_GUIDANCE_EXPORT_PATH": str(GUIDANCE_PATH), "FPV_ERROR_OUTPUT_ENABLE": "1" if state["error_output"] else "0", "FPV_ERROR_OUTPUT_PROTOCOL": state["error_protocol"], "FPV_ERROR_OUTPUT_HOST": state["error_host"], "FPV_ERROR_OUTPUT_PORT": str(state["error_port"]), "FPV_ERROR_OUTPUT_OBJECT_ID": str(state["error_object_id"]), "FPV_ERROR_OUTPUT_UNITS": state["error_units"], "FPV_ERROR_OUTPUT_HFOV_DEG": str(state["error_hfov"]), "FPV_ERROR_OUTPUT_VFOV_DEG": str(state["error_vfov"]), "FPV_ERROR_OUTPUT_RANGE_M": str(state["error_range_m"]), "FPV_AUTOPILOT_BACKEND": "json", "FPV_AUTOPILOT_JSON_PATH": str(DATA_DIR / "autopilot" / "autopilot_cmd.json"), }) stdout = LOG_PATH.open("w", encoding="utf-8", buffering=1) stderr = (LOG_PATH.parent / "main.err.log").open("w", encoding="utf-8", buffering=1) creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) try: CONTROL_PROCESS = subprocess.Popen( [PYTHON_EXE, "-u", str(MAIN_SCRIPT)], cwd=str(APP_DIR), env=env, stdout=stdout, stderr=stderr, creationflags=creationflags, ) if process_source_mode == "camera_bridge" or state["source_mode"] in LIVE_SOURCE_MODES: source_kind = "stream" else: source_kind = "camera" if state["source_mode"] == "camera" else "file" CONTROL_SOURCE = f"{source} ({source_kind})" finally: stdout.close() stderr.close() write_json_file(CONTROL_PATH, state) return control_payload() def parse_range_header(range_header, size): match = re.fullmatch(r"bytes=(\d*)-(\d*)", (range_header or "").strip()) if not match or size <= 0: return None start_text, end_text = match.groups() if not start_text and not end_text: return None if not start_text: length = int(end_text) if length <= 0: return None return max(0, size - length), size - 1 start = int(start_text) end = int(end_text) if end_text else size - 1 if start >= size or start > end: return None return start, min(end, size - 1) def parse_perf(log_text): match = None for line in log_text.splitlines(): if "[perf]" in line: match = line if not match: return {} perf = re.search( r"fps~([0-9.]+).*?iter p50=([0-9.]+) p95=([0-9.]+).*?" r"yolo p50=([0-9.]+) p95=([0-9.]+)(?: skip=([0-9]+))?" r"(?: pass=([0-9]+))?(?: analysisEvery=([0-9]+))?", match, ) if not perf: return {} return { "fps": perf.group(1), "p50": perf.group(2), "p95": perf.group(3), "yolo_p50": perf.group(4), "yolo_p95": perf.group(5), "skip": perf.group(6) or "0", "pass": perf.group(7) or "0", "analysis_every": perf.group(8) or "1", } def parse_source(log_text): for line in reversed(log_text.splitlines()): match = re.search(r"Opened source: (.+?) \(([^)]+)\)", line) if match: return f"{match.group(1)} ({match.group(2)})" for line in reversed(log_text.splitlines()): match = re.search(r"\[entrypoint\] source=([^ ]+)", line) if match: return match.group(1) return os.environ.get("FPV_SOURCE", "") def status_payload(): logs = tail_text(LOG_PATH) parse_logs = tail_text(LOG_PATH, lines=500) video = latest_file(OUT_DIR, "*.mp4") control = control_payload() active_name = control["active_video"] or None return { "logs": logs, "guidance": read_json_file(GUIDANCE_PATH), "perf": parse_perf(parse_logs), "source": CONTROL_SOURCE if CONTROL_SOURCE and control["running"] else parse_source(parse_logs), "frame_exists": FRAME_PATH.exists(), "video": active_name or (video.name if video else ""), "control": control, } def udp_probe_payload(data): if not isinstance(data, dict): raise ValueError("JSON object expected") state = normalize_control(data) width, height = parse_quality(state["quality"]) duration = max(0.2, min(15.0, float(data.get("duration", 3.0)))) result = run_udp_probe( state["input_host"], state["input_port"], UDP_PROBE_DIR, width=width, height=height, separator=state["separator_byte"], frame_encoding=state["frame_encoding"], duration=duration, ) capture = result.get("exact_capture") if capture: capture["dump_url"] = f"/udp-probe/{quote(capture['dump_name'])}" capture["report_url"] = f"/udp-probe/{quote(capture['report_name'])}" result["requested_host"] = state["input_host"] result["requested_port"] = state["input_port"] return result def copy_bytes(src, dst, remaining): while remaining > 0: try: chunk = src.read(min(1024 * 1024, remaining)) except OSError: return if not chunk: break try: dst.write(chunk) except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError): return remaining -= len(chunk) def copy_stream(src, dst): while True: try: chunk = src.read(1024 * 1024) except OSError: return if not chunk: return try: dst.write(chunk) except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError): return def copy_exact(src, dst, size, chunk_size=4 * 1024 * 1024): remaining = int(size) while remaining > 0: chunk = src.read(min(int(chunk_size), remaining)) if not chunk: raise EOFError("upload interrupted") dst.write(chunk) remaining -= len(chunk) return int(size) def input_upload_path(filename): name = str(filename or "").strip() if ( not name or name in {".", ".."} or "/" in name or "\\" in name or "\x00" in name or len(name) > 240 ): return None return INPUT_DIR / name def _path_under(path, roots): try: resolved = Path(path).resolve(strict=False) for root in roots: resolved.relative_to(Path(root).resolve()) return True except (OSError, ValueError): return False return False def resolve_model_path(value=None): raw = str(value or MODEL_PATH).strip().strip('"') path = Path(raw) if not path.is_absolute(): path = APP_DIR / path if path.suffix.lower() not in MODEL_EXTENSIONS: return None roots = (APP_DIR, MODEL_DIR) return path if _path_under(path, roots) else None def model_upload_path(filename): name = str(filename or "").strip() if ( not name or name in {".", ".."} or "/" in name or "\\" in name or "\x00" in name or len(name) > 240 or Path(name).suffix.lower() not in MODEL_EXTENSIONS ): return None path = MODEL_DIR / name return path if _path_under(path, (MODEL_DIR,)) else None def model_catalog(): rows = {} for root in (APP_DIR, MODEL_DIR): if not root.exists(): continue for path in root.glob("*.pt"): if not path.is_file() or not _path_under(path, (root,)): continue try: stat = path.stat() except OSError: continue rows[str(path.resolve())] = { "name": path.name, "path": str(path), "size": stat.st_size, "mtime": int(stat.st_mtime), } return sorted(rows.values(), key=lambda item: item["name"].lower()) def inspect_model(path): path = resolve_model_path(path) if path is None or not path.is_file(): raise FileNotFoundError(str(path or "model")) stat = path.stat() key = (str(path.resolve()), stat.st_mtime_ns, stat.st_size) with MODEL_INFO_LOCK: cached = MODEL_INFO_CACHE.get(key) if cached is not None: return cached import __main__ import cbam_register for name in ("ChannelAttentionDyn", "SpatialAttention", "CBAM"): setattr(__main__, name, getattr(cbam_register, name)) from ultralytics import YOLO loaded = YOLO(str(path)) network = getattr(loaded, "model", loaded) parameters = sum(parameter.numel() for parameter in network.parameters()) trainable = sum(parameter.numel() for parameter in network.parameters() if parameter.requires_grad) layers = [] for index, (name, module) in enumerate(network.named_modules()): if not name: continue layers.append({ "index": index, "name": name, "type": type(module).__name__, "params": sum(parameter.numel() for parameter in module.parameters(recurse=False)), }) if len(layers) >= 512: break names = getattr(loaded, "names", {}) if isinstance(names, dict): names = [names[key] for key in sorted(names)] else: names = list(names or []) info = { "name": path.name, "path": str(path), "size": stat.st_size, "task": str(getattr(loaded, "task", "detect")), "classes": names, "parameters": parameters, "trainable": trainable, "layers": layers, } MODEL_INFO_CACHE.clear() MODEL_INFO_CACHE[key] = info return info def model_payload(): state = read_control_state() selected = resolve_model_path(state.get("model_path")) or resolve_model_path(MODEL_PATH) return { "models": model_catalog(), "selected": str(selected) if selected else "", "default": str(resolve_model_path(MODEL_PATH) or MODEL_PATH), } def ensure_netron(path=None): model = resolve_model_path(path or model_payload()["selected"]) if model is None or not model.is_file(): raise FileNotFoundError(str(model or "model")) try: import netron except ImportError as exc: raise RuntimeError("Netron не установлен в Docker-образе") from exc model_key = str(model.resolve()) address = (NETRON_HOST, NETRON_PORT) with NETRON_LOCK: try: running = netron.status(address) except Exception: running = False if NETRON_STATE.get("path") != model_key or not running: netron.stop(address) netron.start(str(model), address=address, browse=False) NETRON_STATE.update({"path": model_key}) return {"url": "/netron/", "path": str(model), "port": NETRON_PORT} def select_model(data): if process_running(): raise RuntimeError("остановите инференс перед сменой модели") if not isinstance(data, dict): raise ValueError("JSON object expected") path = resolve_model_path(data.get("path")) if path is None or not path.is_file(): raise FileNotFoundError(str(data.get("path") or "model")) state = read_control_state() state["model_path"] = str(path) for key in ("device", "use_half", "conf", "img_size_roi", "img_size_full", "max_det"): if key in data: state[key] = data[key] state = normalize_control(state) write_json_file(CONTROL_PATH, state) return control_payload() def ffmpeg_executable(): exe = shutil.which("ffmpeg") if exe: return exe try: from imageio_ffmpeg import get_ffmpeg_exe except ImportError: return None return get_ffmpeg_exe() HTML = """ FPV Панель

FPV

подключение...
ожидание кадра...
""" class Handler(BaseHTTPRequestHandler): def do_GET(self): parsed = urlparse(self.path) path = parsed.path if path == "/": return self.send_bytes(HTML.encode("utf-8"), "text/html; charset=utf-8") if path == "/api/status": data = json.dumps(status_payload(), ensure_ascii=False).encode("utf-8") return self.send_bytes(data, "application/json; charset=utf-8") if path == "/api/control": return self.send_json(control_payload()) if path == "/api/models": return self.send_json(model_payload()) if path == "/api/model/netron": try: query = parse_qs(parsed.query) requested = (query.get("path") or [None])[0] return self.send_json({"ok": True, **ensure_netron(requested)}) except FileNotFoundError as exc: return self.send_json({"ok": False, "error": f"file not found: {exc}"}, status=404) except RuntimeError as exc: return self.send_json({"ok": False, "error": str(exc)}, status=503) except Exception as exc: return self.send_json({"ok": False, "error": str(exc)}, status=400) if path == "/api/model/architecture": if process_running(): return self.send_json({"ok": False, "error": "остановите инференс перед загрузкой архитектуры"}, status=409) try: query = parse_qs(parsed.query) requested = (query.get("path") or [None])[0] return self.send_json({"ok": True, "model": inspect_model(requested or model_payload()["selected"])}) except FileNotFoundError as exc: return self.send_json({"ok": False, "error": f"file not found: {exc}"}, status=404) except Exception as exc: return self.send_json({"ok": False, "error": str(exc)}, status=400) if path == "/netron": self.send_response(301) self.send_header("Location", "/netron/") self.end_headers() return if path.startswith("/netron/"): return self.proxy_netron(parsed) if path == "/api/input-files": return self.send_json(input_video_files()) if path == "/api/archive": active_name = control_payload()["active_video"] or None data = json.dumps(archive_files(OUT_DIR, active_name), ensure_ascii=False).encode("utf-8") return self.send_bytes(data, "application/json; charset=utf-8") if path == "/frame.jpg": return self.send_file(FRAME_PATH, "image/jpeg") if path == "/stream.mjpg": return self.send_mjpeg() if path.startswith("/udp-probe/"): probe = udp_probe_path(UDP_PROBE_DIR, path.removeprefix("/udp-probe/")) if probe and probe.is_file(): return self.send_file(probe, attachment=True, download_name=probe.name) if path == "/video": video = latest_file(OUT_DIR, "*.mp4") if video: return self.send_file(video) if path.startswith("/archive-play/"): archive = archive_path(OUT_DIR, path.removeprefix("/archive-play/")) active_name = control_payload()["active_video"] or None if not archive or not archive.exists(): self.send_error(404) return if active_name and archive.name == active_name: self.send_error(409, "active recording") return return self.send_transcoded_video(archive) if path.startswith("/archive/"): archive = archive_path(OUT_DIR, path.removeprefix("/archive/")) if archive and archive.exists(): download = parse_qs(parsed.query).get("download") == ["1"] if download: active_name = control_payload()["active_video"] or None if active_name and archive.name == active_name: self.send_error(409, "active recording") return return self.send_h264_download(archive) return self.send_file(archive) self.send_error(404) def do_POST(self): parsed = urlparse(self.path) path = parsed.path if path == "/api/control/start": try: return self.send_json(start_control_process(self.read_json_body())) except FileNotFoundError as exc: return self.send_json({"ok": False, "error": f"file not found: {exc}"}, status=400) except ValueError as exc: return self.send_json({"ok": False, "error": str(exc)}, status=400) except Exception as exc: return self.send_json({"ok": False, "error": str(exc)}, status=500) if path == "/api/control/stop": with CONTROL_LOCK: stopped = stop_control_process() return self.send_json({"ok": True, "stopped": stopped, "control": control_payload()}) if path == "/api/model/select": try: return self.send_json({"ok": True, "control": select_model(self.read_json_body())}) except FileNotFoundError as exc: return self.send_json({"ok": False, "error": f"file not found: {exc}"}, status=404) except ValueError as exc: return self.send_json({"ok": False, "error": str(exc)}, status=400) except RuntimeError as exc: return self.send_json({"ok": False, "error": str(exc)}, status=409) if path == "/api/model-upload": return self.handle_model_upload(parse_qs(parsed.query)) if path == "/api/udp-probe": if not UDP_PROBE_LOCK.acquire(blocking=False): return self.send_json({"ok": False, "error": "UDP probe is already running"}, status=409) try: with CONTROL_LOCK: if process_running(): return self.send_json( {"ok": False, "error": "stop the active stream before UDP probe"}, status=409, ) return self.send_json({"ok": True, **udp_probe_payload(self.read_json_body())}) except (OSError, ValueError) as exc: return self.send_json({"ok": False, "error": str(exc)}, status=400) finally: UDP_PROBE_LOCK.release() if path == "/api/upload": return self.handle_upload(parse_qs(parsed.query)) self.send_error(404) def do_DELETE(self): path = urlparse(self.path).path if path.startswith("/api/archive/"): archive = archive_path(OUT_DIR, path.removeprefix("/api/archive/")) active_name = control_payload()["active_video"] or None if not archive or not archive.exists(): self.send_error(404) return if active_name and archive.name == active_name: self.send_error(409, "active recording") return archive.unlink() delete_h264_cache(archive) self.send_bytes(b'{"ok": true}', "application/json") return self.send_error(404) def log_message(self, fmt, *args): return def proxy_netron(self, parsed): try: ensure_netron() target_path = parsed.path.removeprefix("/netron") or "/" target = target_path + (f"?{parsed.query}" if parsed.query else "") upstream_conn = http.client.HTTPConnection(NETRON_HOST, NETRON_PORT, timeout=10) upstream_conn.request("GET", target, headers={"Host": f"{NETRON_HOST}:{NETRON_PORT}"}) upstream = upstream_conn.getresponse() except (OSError, RuntimeError, FileNotFoundError) as exc: self.send_error(502, f"Netron unavailable: {exc}") return self.send_response(upstream.status, upstream.reason) hop_by_hop = {"connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade", "server", "date", "content-length"} for key, value in upstream.getheaders(): if key.lower() not in hop_by_hop: self.send_header(key, value) content_length = upstream.getheader("Content-Length") if content_length: self.send_header("Content-Length", content_length) self.end_headers() try: while True: chunk = upstream.read(1024 * 1024) if not chunk: break self.wfile.write(chunk) except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError): pass finally: upstream_conn.close() def read_json_body(self): try: length = int(self.headers.get("Content-Length", "0") or "0") except (TypeError, ValueError) as exc: raise ValueError("invalid Content-Length") from exc if length < 0 or length > MAX_JSON_BODY_BYTES: raise ValueError("JSON request body too large") if length <= 0: return {} data = self.rfile.read(length) return json.loads(data.decode("utf-8")) def send_json(self, data, status=200): body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() try: self.wfile.write(body) except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError): pass def handle_upload(self, params): filename = str((params.get("name") or [""])[0]) dst = input_upload_path(filename) if dst is None: return self.send_json({"ok": False, "error": "неподдерживаемое имя файла"}, status=400) try: size = int(self.headers.get("Content-Length", "0") or "0") except ValueError: return self.send_json({"ok": False, "error": "неверный размер файла"}, status=400) if size <= 0: return self.send_json({"ok": False, "error": "пустой файл"}, status=400) max_size = int(os.environ.get("FPV_UI_MAX_UPLOAD_BYTES", str(64 * 1024**3))) if size > max_size: return self.send_json({"ok": False, "error": "файл превышает лимит загрузки"}, status=413) temp = dst.with_name(f".{dst.name}.{threading.get_ident()}.upload") try: INPUT_DIR.mkdir(parents=True, exist_ok=True) with temp.open("wb", buffering=4 * 1024 * 1024) as out: copy_exact(self.rfile, out, size) temp.replace(dst) state = read_control_state() requested_mode = str((params.get("source_mode") or [""])[0]).lower() if requested_mode in FILE_SOURCE_MODES: state["source_mode"] = requested_mode else: state["source_mode"] = "udp_dump" if input_file_kind(dst) == "udp_dump" else "file" state["file_path"] = str(dst) write_json_file(CONTROL_PATH, state) return self.send_json({"ok": True, "path": str(dst), "size": size, "control": control_payload()}) except Exception as exc: return self.send_json({"ok": False, "error": str(exc)}, status=500) finally: try: temp.unlink() except OSError: pass def handle_model_upload(self, params): if process_running(): return self.send_json({"ok": False, "error": "остановите инференс перед загрузкой модели"}, status=409) filename = str((params.get("name") or [""])[0]) dst = model_upload_path(filename) if dst is None: return self.send_json({"ok": False, "error": "нужен файл модели .pt без пути"}, status=400) try: size = int(self.headers.get("Content-Length", "0") or "0") except ValueError: return self.send_json({"ok": False, "error": "неверный размер файла"}, status=400) if size <= 0: return self.send_json({"ok": False, "error": "пустой файл"}, status=400) max_size = int(os.environ.get("FPV_UI_MAX_MODEL_UPLOAD_BYTES", str(8 * 1024**3))) if size > max_size: return self.send_json({"ok": False, "error": "модель превышает лимит загрузки"}, status=413) temp = dst.with_name(f".{dst.name}.{threading.get_ident()}.upload") try: MODEL_DIR.mkdir(parents=True, exist_ok=True) with temp.open("wb", buffering=4 * 1024 * 1024) as out: copy_exact(self.rfile, out, size) temp.replace(dst) state = read_control_state() state["model_path"] = str(dst) write_json_file(CONTROL_PATH, state) return self.send_json({"ok": True, "path": str(dst), "size": size, "control": control_payload()}) except Exception as exc: return self.send_json({"ok": False, "error": str(exc)}, status=500) finally: try: temp.unlink() except OSError: pass def send_bytes(self, data, content_type): self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(data))) self.end_headers() try: self.wfile.write(data) except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError): pass def send_file(self, path, content_type=None, attachment=False, download_name=None): try: size = path.stat().st_size except FileNotFoundError: self.send_error(404) return ctype = content_type or mimetypes.guess_type(path.name)[0] or "application/octet-stream" range_header = self.headers.get("Range") if range_header: parsed = parse_range_header(range_header, size) if not parsed: self.send_response(416) self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Range", f"bytes */{size}") self.end_headers() return start, end = parsed self.send_response(206) self.send_header("Content-Type", ctype) self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Range", f"bytes {start}-{end}/{size}") self.send_header("Content-Length", str(end - start + 1)) if attachment: self.send_header("Content-Disposition", content_disposition(download_name or path.name)) self.end_headers() try: with path.open("rb") as f: f.seek(start) copy_bytes(f, self.wfile, end - start + 1) except FileNotFoundError: pass return self.send_response(200) self.send_header("Content-Type", ctype) self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Length", str(size)) if attachment: self.send_header("Content-Disposition", content_disposition(download_name or path.name)) self.end_headers() try: with path.open("rb") as f: copy_bytes(f, self.wfile, size) except FileNotFoundError: pass def send_h264_download(self, path): try: download_path = h264_download_file(path) except RuntimeError as exc: self.send_error(500, str(exc)[:160]) return return self.send_file( download_path, content_type="video/mp4", attachment=True, download_name=h264_download_name(path), ) def send_mjpeg(self): try: self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) except OSError: pass self.send_response(200) self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=frame") self.send_header("Cache-Control", "no-store, no-cache, must-revalidate") self.send_header("Pragma", "no-cache") self.end_headers() last_jpeg = None try: while True: try: jpeg = FRAME_PATH.read_bytes() if jpeg and jpeg != last_jpeg: self.wfile.write( b"--frame\r\n" b"Content-Type: image/jpeg\r\n" + f"Content-Length: {len(jpeg)}\r\n\r\n".encode("ascii") + jpeg + b"\r\n" ) self.wfile.flush() last_jpeg = jpeg except FileNotFoundError: pass time.sleep(0.01) except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError): return def send_transcoded_video(self, path): ffmpeg = ffmpeg_executable() if not ffmpeg: self.send_error(500, "ffmpeg unavailable") return command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-i", str(path), "-an", "-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency", "-crf", "18", "-pix_fmt", "yuv420p", "-movflags", "frag_keyframe+empty_moov+default_base_moof", "-f", "mp4", "pipe:1", ] try: proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) except OSError: self.send_error(500, "ffmpeg unavailable") return self.send_response(200) self.send_header("Content-Type", "video/mp4") self.send_header("Cache-Control", "no-store") self.end_headers() try: copy_stream(proc.stdout, self.wfile) finally: if proc.poll() is None: proc.kill() proc.wait() def main(): removed = cleanup_empty_recordings(OUT_DIR) if removed: print(f"Removed empty recordings: {removed}", flush=True) server = ThreadingHTTPServer((HOST, PORT), Handler) print(f"UI listening on http://{HOST}:{PORT}", flush=True) server.serve_forever() if __name__ == "__main__": main()