| from __future__ import annotations |
|
|
| import json |
| import threading |
| import time |
| from collections import deque |
| from dataclasses import dataclass, field |
| from http import HTTPStatus |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| from typing import Any |
| from urllib.parse import parse_qs, urlparse |
|
|
| from src.core.utils import infer_modality |
|
|
|
|
| def _build_mock_pgm() -> bytes: |
| width = 64 |
| height = 64 |
| pixels = bytearray(width * height) |
| for y in range(height): |
| for x in range(width): |
| value = 220 if 12 <= x < 52 and 12 <= y < 52 and ((x // 4 + y // 4) % 2 == 0) else 20 |
| if x == y or x + y == width - 1: |
| value = 255 |
| pixels[y * width + x] = value |
| return f"P5\n{width} {height}\n255\n".encode("ascii") + bytes(pixels) |
|
|
|
|
| MOCK_IMAGE_BYTES = _build_mock_pgm() |
|
|
|
|
| @dataclass(slots=True) |
| class MockFrameSpec: |
| image_path: str |
| video_name: str |
|
|
|
|
| @dataclass(slots=True) |
| class MockTranslationSpec: |
| translation_x: float |
| translation_y: float |
| translation_z: float |
| health_status: str |
|
|
|
|
| @dataclass(slots=True) |
| class MockSessionData: |
| session_name: str |
| frames: list[MockFrameSpec] |
| translations: list[MockTranslationSpec] |
|
|
|
|
| def build_default_session(frame_count: int = 3, session_name: str = "mock_session") -> MockSessionData: |
| frames = [ |
| MockFrameSpec(image_path=f"/{session_name}/frame_{index:06d}.jpg", video_name=session_name) |
| for index in range(frame_count) |
| ] |
| translations = [ |
| MockTranslationSpec( |
| translation_x=float(index), |
| translation_y=float(index) * 0.5, |
| translation_z=10.0 + float(index), |
| health_status="1" if index < max(frame_count - 1, 1) else "0", |
| ) |
| for index in range(frame_count) |
| ] |
| return MockSessionData(session_name=session_name, frames=frames, translations=translations) |
|
|
|
|
| @dataclass(slots=True) |
| class _RateWindow: |
| limit_per_minute: int |
| events: deque[float] = field(default_factory=deque) |
|
|
| def allow(self) -> bool: |
| now = time.time() |
| while self.events and now - self.events[0] > 60.0: |
| self.events.popleft() |
| if len(self.events) >= self.limit_per_minute: |
| return False |
| self.events.append(now) |
| return True |
|
|
|
|
| @dataclass(slots=True) |
| class MockServerState: |
| username: str |
| password: str |
| session: MockSessionData |
| mode: str = "batch" |
| prediction_limit_per_minute: int = 80 |
| manifest_limit_per_minute: int = 5 |
| token: str = "mock-token" |
| object_limit: int = 16 |
| accepted_predictions: dict[str, dict[str, Any]] = field(default_factory=dict) |
| prediction_window: _RateWindow = field(init=False) |
| manifest_window: _RateWindow = field(init=False) |
| sequential_session_id: str | None = None |
| sequential_next_index: int = 0 |
| sequential_waiting_frame_url: str | None = None |
| sequential_open: bool = False |
| sequential_ready_at: float = 0.0 |
| sequential_warmup_delay_s: float = 0.0 |
| next_frame_delay_s: float = 0.0 |
| prediction_delay_s: float = 0.0 |
| next_frame_retry_failures_remaining: int = 0 |
| prediction_retry_failures_remaining: int = 0 |
| next_frame_timeout_failures_remaining: int = 0 |
| next_frame_timeout_delay_s: float = 0.0 |
|
|
| def __post_init__(self) -> None: |
| self.prediction_window = _RateWindow(limit_per_minute=self.prediction_limit_per_minute) |
| self.manifest_window = _RateWindow(limit_per_minute=self.manifest_limit_per_minute) |
|
|
| def reset_sequential(self) -> None: |
| self.sequential_session_id = None |
| self.sequential_next_index = 0 |
| self.sequential_waiting_frame_url = None |
| self.sequential_open = False |
| self.sequential_ready_at = 0.0 |
|
|
|
|
| class OfficialRepoMockServer: |
| def __init__( |
| self, |
| *, |
| host: str = "127.0.0.1", |
| port: int = 0, |
| username: str = "team", |
| password: str = "password", |
| session: MockSessionData | None = None, |
| mode: str = "batch", |
| prediction_limit_per_minute: int = 80, |
| manifest_limit_per_minute: int = 5, |
| sequential_warmup_delay_s: float = 0.0, |
| next_frame_delay_s: float = 0.0, |
| prediction_delay_s: float = 0.0, |
| next_frame_retry_failures: int = 0, |
| prediction_retry_failures: int = 0, |
| next_frame_timeout_failures: int = 0, |
| next_frame_timeout_delay_s: float = 0.0, |
| ) -> None: |
| self.state = MockServerState( |
| username=username, |
| password=password, |
| session=session or build_default_session(), |
| mode=mode, |
| prediction_limit_per_minute=prediction_limit_per_minute, |
| manifest_limit_per_minute=manifest_limit_per_minute, |
| sequential_warmup_delay_s=sequential_warmup_delay_s, |
| next_frame_delay_s=next_frame_delay_s, |
| prediction_delay_s=prediction_delay_s, |
| next_frame_retry_failures_remaining=next_frame_retry_failures, |
| prediction_retry_failures_remaining=prediction_retry_failures, |
| next_frame_timeout_failures_remaining=next_frame_timeout_failures, |
| next_frame_timeout_delay_s=next_frame_timeout_delay_s, |
| ) |
| self._server = ThreadingHTTPServer((host, port), self._build_handler()) |
| self._thread: threading.Thread | None = None |
|
|
| @property |
| def base_url(self) -> str: |
| host, port = self._server.server_address |
| return f"http://{host}:{port}/" |
|
|
| def start(self) -> None: |
| self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) |
| self._thread.start() |
|
|
| def stop(self) -> None: |
| self._server.shutdown() |
| self._server.server_close() |
| if self._thread is not None: |
| self._thread.join(timeout=2.0) |
|
|
| def _build_handler(self) -> type[BaseHTTPRequestHandler]: |
| state = self.state |
| server_ref = self |
|
|
| class Handler(BaseHTTPRequestHandler): |
| def log_message(self, format: str, *args: Any) -> None: |
| return |
|
|
| def _send_json(self, status_code: int, payload: dict[str, Any] | list[dict[str, Any]]) -> None: |
| body = json.dumps(payload).encode("utf-8") |
| self.send_response(status_code) |
| self.send_header("Content-Type", "application/json") |
| self.send_header("Content-Length", str(len(body))) |
| self.end_headers() |
| self.wfile.write(body) |
|
|
| def _send_bytes(self, status_code: int, payload: bytes, content_type: str = "application/octet-stream") -> None: |
| self.send_response(status_code) |
| self.send_header("Content-Type", content_type) |
| self.send_header("Content-Length", str(len(payload))) |
| self.end_headers() |
| self.wfile.write(payload) |
|
|
| def _send_empty(self, status_code: int) -> None: |
| self.send_response(status_code) |
| self.send_header("Content-Length", "0") |
| self.end_headers() |
|
|
| def _is_authorized(self) -> bool: |
| return self.headers.get("Authorization") == f"Token {state.token}" |
|
|
| def _frame_payload(self) -> list[dict[str, Any]]: |
| payload: list[dict[str, Any]] = [] |
| for index, frame in enumerate(state.session.frames, start=1): |
| payload.append( |
| { |
| "url": f"{server_ref.base_url}frames/{index}/", |
| "image_url": frame.image_path, |
| "video_name": frame.video_name, |
| } |
| ) |
| return payload |
|
|
| def _translation_payload(self) -> list[dict[str, Any]]: |
| payload: list[dict[str, Any]] = [] |
| for translation in state.session.translations: |
| payload.append( |
| { |
| "translation_x": translation.translation_x, |
| "translation_y": translation.translation_y, |
| "translation_z": translation.translation_z, |
| "health_status": translation.health_status, |
| } |
| ) |
| return payload |
|
|
| def _sequential_frame_payload(self, index: int) -> dict[str, Any]: |
| frame = state.session.frames[index] |
| translation = state.session.translations[index] |
| frame_url = f"{server_ref.base_url}frames/{index + 1}/" |
| return { |
| "session_id": state.sequential_session_id, |
| "frame_id": index + 1, |
| "frame_url": frame_url, |
| "image_url": frame.image_path, |
| "video_name": frame.video_name, |
| "translation_x": translation.translation_x, |
| "translation_y": translation.translation_y, |
| "translation_z": translation.translation_z, |
| "health_status": translation.health_status, |
| "camera_mode": infer_modality(frame.video_name), |
| "deadline_ms": 1000, |
| } |
|
|
| def do_GET(self) -> None: |
| parsed = urlparse(self.path) |
| if parsed.path == "/frames/": |
| if not self._is_authorized(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Unauthorized"}) |
| return |
| if not state.manifest_window.allow(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "You do not have permission to perform this action."}) |
| return |
| self._send_json(HTTPStatus.OK, self._frame_payload()) |
| return |
|
|
| if parsed.path == "/translation/": |
| if not self._is_authorized(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Unauthorized"}) |
| return |
| self._send_json(HTTPStatus.OK, self._translation_payload()) |
| return |
|
|
| if parsed.path == "/session/": |
| if not self._is_authorized(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Unauthorized"}) |
| return |
| self._send_json(HTTPStatus.OK, {"name": state.session.session_name}) |
| return |
|
|
| if parsed.path == "/session/next/": |
| if not self._is_authorized(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Unauthorized"}) |
| return |
| if state.mode != "sequential": |
| self._send_json(HTTPStatus.NOT_FOUND, {"detail": "Sequential mode disabled"}) |
| return |
| if not state.sequential_open or not state.sequential_session_id: |
| self._send_json(HTTPStatus.CONFLICT, {"detail": "Session is not open"}) |
| return |
| if time.time() < state.sequential_ready_at: |
| self._send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"detail": "Warm-up in progress"}) |
| return |
| if state.next_frame_retry_failures_remaining > 0: |
| state.next_frame_retry_failures_remaining -= 1 |
| self._send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"detail": "Retry next_frame"}) |
| return |
| if state.next_frame_timeout_failures_remaining > 0: |
| state.next_frame_timeout_failures_remaining -= 1 |
| time.sleep(state.next_frame_timeout_delay_s) |
| if state.next_frame_delay_s > 0: |
| time.sleep(state.next_frame_delay_s) |
| if state.sequential_waiting_frame_url is not None: |
| self._send_json(HTTPStatus.CONFLICT, {"detail": "Prediction required before next frame"}) |
| return |
| if state.sequential_next_index >= len(state.session.frames): |
| self._send_empty(HTTPStatus.NO_CONTENT) |
| return |
|
|
| payload = self._sequential_frame_payload(state.sequential_next_index) |
| state.sequential_waiting_frame_url = str(payload["frame_url"]) |
| self._send_json(HTTPStatus.OK, payload) |
| return |
|
|
| if parsed.path.startswith("/classes/"): |
| class_id = parsed.path.strip("/").split("/")[-1] |
| self._send_json(HTTPStatus.OK, {"id": class_id, "url": f"{server_ref.base_url}classes/{class_id}/"}) |
| return |
|
|
| if parsed.path.startswith("/media/"): |
| self._send_bytes(HTTPStatus.OK, MOCK_IMAGE_BYTES, content_type="image/x-portable-graymap") |
| return |
|
|
| self._send_json(HTTPStatus.NOT_FOUND, {"detail": "Not found"}) |
|
|
| def do_POST(self) -> None: |
| parsed = urlparse(self.path) |
| raw_body = self.rfile.read(int(self.headers.get("Content-Length", "0") or "0")) |
|
|
| if parsed.path == "/auth/": |
| form = parse_qs(raw_body.decode("utf-8")) |
| username = form.get("username", [""])[0] |
| password = form.get("password", [""])[0] |
| if username == state.username and password == state.password: |
| self._send_json(HTTPStatus.OK, {"token": state.token}) |
| else: |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Login failed"}) |
| return |
|
|
| if parsed.path == "/prediction/": |
| if not self._is_authorized(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Unauthorized"}) |
| return |
| if not state.prediction_window.allow(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "You do not have permission to perform this action."}) |
| return |
| payload = json.loads(raw_body.decode("utf-8") or "{}") |
| if state.mode == "batch": |
| required_keys = {"frame", "detected_objects", "detected_translations"} |
| payload_keys = set(payload.keys()) |
| if payload_keys != required_keys: |
| self._send_json( |
| HTTPStatus.BAD_REQUEST, |
| { |
| "detail": "Batch payload field mismatch", |
| "missing": sorted(required_keys - payload_keys), |
| "unexpected": sorted(payload_keys - required_keys), |
| }, |
| ) |
| return |
| for item in payload.get("detected_objects", []): |
| if "motion_status" in item: |
| self._send_json(HTTPStatus.BAD_REQUEST, {"detail": "motion_status not allowed in batch 2025 payload"}) |
| return |
| frame_url = payload.get("frame") |
| if frame_url in state.accepted_predictions: |
| self._send_json(HTTPStatus.NOT_ACCEPTABLE, {"detail": "Prediction already exists."}) |
| return |
| state.accepted_predictions[str(frame_url)] = payload |
| self._send_json(HTTPStatus.CREATED, payload) |
| return |
|
|
| if parsed.path == "/session/open/": |
| if not self._is_authorized(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Unauthorized"}) |
| return |
| if state.mode != "sequential": |
| self._send_json(HTTPStatus.NOT_FOUND, {"detail": "Sequential mode disabled"}) |
| return |
| state.reset_sequential() |
| state.sequential_session_id = f"seq-{int(time.time() * 1000)}" |
| state.sequential_open = True |
| state.sequential_ready_at = time.time() + max(float(state.sequential_warmup_delay_s), 0.0) |
| self._send_json( |
| HTTPStatus.OK, |
| { |
| "session_id": state.sequential_session_id, |
| "session_name": state.session.session_name, |
| "object_limit": state.object_limit, |
| "reference_manifest": [], |
| }, |
| ) |
| return |
|
|
| if parsed.path == "/session/prediction/": |
| if not self._is_authorized(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Unauthorized"}) |
| return |
| if state.mode != "sequential": |
| self._send_json(HTTPStatus.NOT_FOUND, {"detail": "Sequential mode disabled"}) |
| return |
| if state.prediction_retry_failures_remaining > 0: |
| state.prediction_retry_failures_remaining -= 1 |
| self._send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"detail": "Retry prediction"}) |
| return |
| if state.prediction_delay_s > 0: |
| time.sleep(state.prediction_delay_s) |
| payload = json.loads(raw_body.decode("utf-8") or "{}") |
| session_id = payload.get("session_id") |
| frame_url = payload.get("frame") |
| if not state.sequential_open or session_id != state.sequential_session_id: |
| self._send_json(HTTPStatus.CONFLICT, {"detail": "Session mismatch"}) |
| return |
| if state.sequential_waiting_frame_url is None: |
| self._send_json(HTTPStatus.CONFLICT, {"detail": "No frame pending prediction"}) |
| return |
| if frame_url != state.sequential_waiting_frame_url: |
| self._send_json(HTTPStatus.NOT_ACCEPTABLE, {"detail": "Unexpected frame order"}) |
| return |
| if frame_url in state.accepted_predictions: |
| self._send_json(HTTPStatus.NOT_ACCEPTABLE, {"detail": "Prediction already exists."}) |
| return |
| state.accepted_predictions[str(frame_url)] = payload |
| state.sequential_waiting_frame_url = None |
| state.sequential_next_index += 1 |
| self._send_json(HTTPStatus.CREATED, payload) |
| return |
|
|
| if parsed.path == "/session/close/": |
| if not self._is_authorized(): |
| self._send_json(HTTPStatus.FORBIDDEN, {"detail": "Unauthorized"}) |
| return |
| payload = json.loads(raw_body.decode("utf-8") or "{}") |
| closed = bool(state.sequential_open and payload.get("session_id") == state.sequential_session_id) |
| state.reset_sequential() |
| self._send_json(HTTPStatus.OK, {"closed": closed}) |
| return |
|
|
| self._send_json(HTTPStatus.NOT_FOUND, {"detail": "Not found"}) |
|
|
| return Handler |
|
|
|
|
| if __name__ == "__main__": |
| server = OfficialRepoMockServer() |
| server.start() |
| print(server.base_url) |
| try: |
| while True: |
| time.sleep(0.5) |
| except KeyboardInterrupt: |
| server.stop() |
|
|
