| from __future__ import annotations
|
|
|
| from abc import ABC, abstractmethod
|
| from dataclasses import dataclass, field
|
| from time import perf_counter
|
| from time import sleep
|
| from typing import Any
|
|
|
| from src.config.settings import SequentialProtocolSettings
|
| from src.core.frame_state import FrameEnvelope, FrameResult
|
| from src.core.logger import StructuredLogger
|
| from src.server.client import HttpResponse, SimpleHttpClient
|
| from src.server.protocol import ProtocolError
|
|
|
|
|
| class SequentialProtocolAdapter(ABC):
|
| """Final yarisma icin kare sirali adapter arayuzu."""
|
|
|
| @abstractmethod
|
| def login(self) -> str:
|
| raise NotImplementedError
|
|
|
| @abstractmethod
|
| def open_session(self) -> dict[str, Any]:
|
| raise NotImplementedError
|
|
|
| @abstractmethod
|
| def fetch_next_frame(self) -> FrameEnvelope | None:
|
| raise NotImplementedError
|
|
|
| @abstractmethod
|
| def download_image(self, frame: FrameEnvelope) -> bytes:
|
| raise NotImplementedError
|
|
|
| @abstractmethod
|
| def build_wire_prediction(self, result: FrameResult) -> dict[str, Any]:
|
| raise NotImplementedError
|
|
|
| @abstractmethod
|
| def send_wire_prediction(self, payload: dict[str, Any]) -> HttpResponse:
|
| raise NotImplementedError
|
|
|
| @abstractmethod
|
| def close_session(self) -> None:
|
| raise NotImplementedError
|
|
|
|
|
| @dataclass(slots=True)
|
| class FinalSequentialAdapter(SequentialProtocolAdapter):
|
| """Config-driven sequential adapter; batch adapter'dan bagimsizdir."""
|
|
|
| settings: SequentialProtocolSettings
|
| client: SimpleHttpClient = field(default_factory=SimpleHttpClient)
|
| logger: StructuredLogger | None = None
|
| auth_token: str | None = None
|
| session_id: str | None = None
|
| session_name: str | None = None
|
| current_frame: FrameEnvelope | None = None
|
| warmup_completed: bool = False
|
| _warmup_wait_logged: bool = False
|
|
|
| def __post_init__(self) -> None:
|
| base_url = self.settings.normalized_base_url
|
| self.url_login = f"{base_url}{self.settings.resolve_path('auth')}"
|
| self.url_open_session = f"{base_url}{self.settings.resolve_path('open_session')}"
|
| self.url_next_frame = f"{base_url}{self.settings.resolve_path('next_frame')}"
|
| self.url_prediction = f"{base_url}{self.settings.resolve_path('prediction')}"
|
| self.url_close_session = f"{base_url}{self.settings.resolve_path('close_session')}"
|
| self.base_url = base_url
|
| self.logger = self.logger or StructuredLogger()
|
|
|
| def _auth_headers(self) -> dict[str, str]:
|
| if not self.auth_token:
|
| raise ProtocolError("Sequential adapter giris yapmadan kullanilamaz.")
|
| return {"Authorization": f"Token {self.auth_token}"}
|
|
|
| def login(self) -> str:
|
| t0 = perf_counter()
|
| response = self.client.post_form(
|
| self.url_login,
|
| {"username": self.settings.username, "password": self.settings.password},
|
| timeout=self.settings.effective_timeout("request_timeout_s", self.settings.request_timeout_s),
|
| )
|
| latency_ms = round((perf_counter() - t0) * 1000.0, 3)
|
| if response.status_code != 200:
|
| self.logger.log_error(
|
| event="sequential_login_failed",
|
| adapter=type(self).__name__,
|
| status_code=response.status_code,
|
| latency_ms=latency_ms,
|
| diagnostics={"response_text": response.text},
|
| )
|
| raise ProtocolError(f"Sequential login basarisiz: {response.status_code}")
|
| payload = response.json()
|
| token = payload.get("token")
|
| if not token:
|
| raise ProtocolError("Sequential login cevabinda token alani yok.")
|
| self.auth_token = str(token)
|
| self.logger.log_runtime(
|
| event="sequential_login_succeeded",
|
| adapter=type(self).__name__,
|
| status_code=response.status_code,
|
| latency_ms=latency_ms,
|
| )
|
| return self.auth_token
|
|
|
| def open_session(self) -> dict[str, Any]:
|
| if not self.auth_token:
|
| self.login()
|
| response = self.client.post_json(
|
| self.url_open_session,
|
| {},
|
| headers=self._auth_headers(),
|
| timeout=self.settings.effective_timeout("request_timeout_s", self.settings.request_timeout_s),
|
| )
|
| if response.status_code != 200:
|
| raise ProtocolError(f"Sequential session acilamadi: {response.status_code} {response.text}")
|
| payload = response.json()
|
| self.session_id = str(payload.get("session_id"))
|
| self.session_name = str(payload.get("session_name", self.session_id))
|
| self.warmup_completed = False
|
| self._warmup_wait_logged = False
|
| self.logger.log_runtime(
|
| event="sequential_session_opened",
|
| adapter=type(self).__name__,
|
| session_name=self.session_name,
|
| status_code=response.status_code,
|
| diagnostics={"object_limit": payload.get("object_limit"), "reference_manifest": payload.get("reference_manifest", [])},
|
| )
|
| return payload
|
|
|
| def fetch_next_frame(self) -> FrameEnvelope | None:
|
| if not self.auth_token:
|
| self.login()
|
| if not self.session_id:
|
| self.open_session()
|
| first_frame_wait = not self.warmup_completed
|
| if first_frame_wait and not self._warmup_wait_logged:
|
| self.logger.log_runtime(
|
| event="sequential_warmup_wait_started",
|
| adapter=type(self).__name__,
|
| session_name=self.session_name,
|
| diagnostics={"warmup_timeout_s": self.settings.warmup_timeout_s},
|
| )
|
| self._warmup_wait_logged = True
|
| response = self._retry_request(
|
| request_label="next_frame",
|
| request_fn=lambda timeout: self.client.get_json(
|
| self.url_next_frame,
|
| headers=self._auth_headers(),
|
| timeout=timeout,
|
| ),
|
| retryable_statuses=self.settings.next_frame_retryable_statuses,
|
| timeout=(
|
| self.settings.effective_timeout("first_frame_timeout_s", self.settings.first_frame_timeout_s)
|
| if first_frame_wait
|
| else self.settings.effective_timeout("request_timeout_s", self.settings.request_timeout_s)
|
| ),
|
| )
|
| if response.status_code == 204:
|
| self.current_frame = None
|
| return None
|
| if response.status_code != 200:
|
| raise ProtocolError(f"Sequential frame alinamadi: {response.status_code} {response.text}")
|
| payload = response.json()
|
| frame = FrameEnvelope(
|
| frame_url=str(payload["frame_url"]),
|
| image_url=str(payload["image_url"]),
|
| video_name=str(payload["video_name"]),
|
| translation_x=float(payload["translation_x"]),
|
| translation_y=float(payload["translation_y"]),
|
| translation_z=float(payload["translation_z"]),
|
| health_status=str(payload["health_status"]),
|
| metadata={
|
| "session_id": payload.get("session_id"),
|
| "frame_id": payload.get("frame_id"),
|
| "camera_mode": payload.get("camera_mode"),
|
| "deadline_ms": payload.get("deadline_ms"),
|
| },
|
| )
|
| self.current_frame = frame
|
| if not self.warmup_completed:
|
| self.warmup_completed = True
|
| self.logger.log_runtime(
|
| event="sequential_warmup_completed",
|
| adapter=type(self).__name__,
|
| session_name=self.session_name,
|
| frame_url=frame.frame_url,
|
| diagnostics={"frame_id": frame.metadata.get("frame_id")},
|
| )
|
| return frame
|
|
|
| def resolve_image_url(self, image_url: str) -> str:
|
| if image_url.startswith("http://") or image_url.startswith("https://"):
|
| return image_url
|
| normalized = image_url.lstrip("/")
|
| if normalized.startswith("media/"):
|
| return f"{self.base_url}{normalized}"
|
| return f"{self.base_url}media/{normalized}"
|
|
|
| def download_image(self, frame: FrameEnvelope) -> bytes:
|
| response = self.client.get_bytes(
|
| self.resolve_image_url(frame.image_url),
|
| timeout=self.settings.effective_timeout("image_timeout_s", self.settings.image_timeout_s),
|
| )
|
| if response.status_code != 200:
|
| raise ProtocolError(f"Sequential gorsel indirilemedi: {response.status_code}")
|
| return response.body
|
|
|
| def build_wire_prediction(self, result: FrameResult) -> dict[str, Any]:
|
| if not self.current_frame:
|
| raise ProtocolError("Prediction olusturmak icin aktif frame yok.")
|
|
|
| payload: dict[str, Any] = {
|
| "session_id": self.session_id,
|
| "frame_id": self.current_frame.metadata.get("frame_id"),
|
| "frame": result.frame_url,
|
| "detected_objects": [],
|
| "detected_translations": [],
|
| "detected_undefined_objects": [],
|
| }
|
| for obj in result.detected_objects:
|
| wire_obj: dict[str, Any] = {
|
| "cls": f"{self.base_url}classes/{int(obj.class_id) + 1}/",
|
| "landing_status": str(obj.landing_status),
|
| "top_left_x": str(obj.top_left_x),
|
| "top_left_y": str(obj.top_left_y),
|
| "bottom_right_x": str(obj.bottom_right_x),
|
| "bottom_right_y": str(obj.bottom_right_y),
|
| }
|
|
|
| if int(obj.class_id) == 0 and obj.motion_status in (0, 1):
|
| wire_obj["motion_status"] = str(obj.motion_status)
|
| payload["detected_objects"].append(wire_obj)
|
|
|
| for translation in result.detected_translations:
|
| payload["detected_translations"].append(
|
| {
|
| "translation_x": str(translation.translation_x),
|
| "translation_y": str(translation.translation_y),
|
| "translation_z": str(translation.translation_z),
|
| }
|
| )
|
|
|
|
|
| for uobj in result.detected_undefined_objects:
|
| payload["detected_undefined_objects"].append(
|
| {
|
| "object_id": str(uobj.object_id),
|
| "top_left_x": str(uobj.top_left_x),
|
| "top_left_y": str(uobj.top_left_y),
|
| "bottom_right_x": str(uobj.bottom_right_x),
|
| "bottom_right_y": str(uobj.bottom_right_y),
|
| }
|
| )
|
|
|
| return payload
|
|
|
| def send_wire_prediction(self, payload: dict[str, Any]) -> HttpResponse:
|
| if not self.auth_token:
|
| self.login()
|
| response = self._retry_request(
|
| request_label="prediction",
|
| request_fn=lambda timeout: self.client.post_json(
|
| self.url_prediction,
|
| payload,
|
| headers=self._auth_headers(),
|
| timeout=timeout,
|
| ),
|
| retryable_statuses=self.settings.prediction_retryable_statuses,
|
| timeout=self.settings.effective_timeout("request_timeout_s", self.settings.request_timeout_s),
|
| )
|
| if response.status_code != 201:
|
| raise ProtocolError(f"Sequential prediction gonderilemedi: {response.status_code} {response.text}")
|
| self.logger.log_runtime(
|
| event="sequential_prediction_sent",
|
| adapter=type(self).__name__,
|
| session_name=self.session_name,
|
| frame_url=str(payload.get("frame")),
|
| status_code=response.status_code,
|
| detected_objects=len(payload.get("detected_objects", [])),
|
| detected_translations=len(payload.get("detected_translations", [])),
|
| )
|
| self.current_frame = None
|
| return response
|
|
|
| def close_session(self) -> None:
|
| if not self.auth_token or not self.session_id:
|
| self.current_frame = None
|
| self.session_id = None
|
| return
|
| response = self.client.post_json(
|
| self.url_close_session,
|
| {"session_id": self.session_id},
|
| headers=self._auth_headers(),
|
| timeout=self.settings.effective_timeout("request_timeout_s", self.settings.request_timeout_s),
|
| )
|
| if response.status_code != 200:
|
| raise ProtocolError(f"Sequential session kapatilamadi: {response.status_code} {response.text}")
|
| self.logger.log_runtime(
|
| event="sequential_session_closed",
|
| adapter=type(self).__name__,
|
| session_name=self.session_name,
|
| status_code=response.status_code,
|
| )
|
| self.current_frame = None
|
| self.session_id = None
|
| self.session_name = None
|
| self.warmup_completed = False
|
| self._warmup_wait_logged = False
|
|
|
| def _retry_request(
|
| self,
|
| *,
|
| request_label: str,
|
| request_fn,
|
| retryable_statuses: list[int],
|
| timeout: float,
|
| ) -> HttpResponse:
|
| max_retries = int(self.settings.retry_policy.get("max_retries", 0))
|
| backoff_s = float(self.settings.retry_policy.get("backoff_s", 0.0))
|
| last_error: str | None = None
|
| for attempt in range(max_retries + 1):
|
| try:
|
| response = request_fn(timeout)
|
| except Exception as exc:
|
| last_error = str(exc)
|
| if attempt >= max_retries:
|
| break
|
| self.logger.log_runtime(
|
| event="sequential_request_retrying",
|
| adapter=type(self).__name__,
|
| session_name=self.session_name,
|
| fallback_mode=request_label,
|
| diagnostics={"attempt": attempt + 1, "error": last_error},
|
| )
|
| sleep(backoff_s)
|
| continue
|
|
|
| if response.status_code not in retryable_statuses or attempt >= max_retries:
|
| return response
|
| self.logger.log_runtime(
|
| event="sequential_request_retrying",
|
| adapter=type(self).__name__,
|
| session_name=self.session_name,
|
| fallback_mode=request_label,
|
| status_code=response.status_code,
|
| diagnostics={"attempt": attempt + 1, "response_text": response.text},
|
| )
|
| sleep(backoff_s)
|
|
|
| self.logger.log_error(
|
| event="sequential_request_retry_exhausted",
|
| adapter=type(self).__name__,
|
| session_name=self.session_name,
|
| fallback_mode=request_label,
|
| diagnostics={"error": last_error, "max_retries": max_retries},
|
| )
|
| raise ProtocolError(f"Sequential {request_label} retry limiti doldu: {last_error}")
|
|
|
