Adriano
95d643d6d8
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NABedeMX43pyXZbDNUgA5B
74 lines
3.3 KiB
Python
74 lines
3.3 KiB
Python
"""One row per vision execution - not per quote."""
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import DateTime, Enum, ForeignKey, Integer, JSON, String, func
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from src.backend.database import Base
|
|
|
|
|
|
class VisionResult(Base):
|
|
__tablename__ = "vision_results"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
task_id: Mapped[int] = mapped_column(
|
|
Integer, ForeignKey("recipe_tasks.id", ondelete="CASCADE"),
|
|
nullable=False, index=True,
|
|
)
|
|
image_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
|
|
overlay_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
|
|
engine_version: Mapped[str] = mapped_column(String(64), nullable=False)
|
|
executed_on: Mapped[str] = mapped_column(
|
|
Enum("server", "station", name="vision_executed_on_enum"), nullable=False
|
|
)
|
|
station_id: Mapped[Optional[int]] = mapped_column(
|
|
Integer, ForeignKey("stations.id"), nullable=True
|
|
)
|
|
device_code: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
|
|
# URL del worker effettivamente chiamato (#45), copiato: la stazione puo' cambiarlo dopo
|
|
worker_url: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
|
|
# Copied, not referenced: a calibration redone six months from now must not
|
|
# rewrite what a measurement already taken means. Same for the graph.
|
|
calibration_snapshot: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
|
|
graph_snapshot: Mapped[dict] = mapped_column(JSON, nullable=False)
|
|
duration_ms: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
|
|
executed_at: Mapped[datetime] = mapped_column(
|
|
DateTime, nullable=False, server_default=func.now()
|
|
)
|
|
|
|
__table_args__ = ({"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"},)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<VisionResult task={self.task_id} engine={self.engine_version}>"
|
|
|
|
|
|
class VisionReferenceImage(Base):
|
|
"""An image kept so a graph can be composed, tried and re-tried without a
|
|
camera - which is exactly what VisionSuite's source abstraction is for."""
|
|
|
|
__tablename__ = "vision_reference_images"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
task_id: Mapped[int] = mapped_column(
|
|
Integer, ForeignKey("recipe_tasks.id", ondelete="CASCADE"),
|
|
nullable=False, index=True,
|
|
)
|
|
path: Mapped[str] = mapped_column(String(500), nullable=False)
|
|
station_id: Mapped[Optional[int]] = mapped_column(
|
|
Integer, ForeignKey("stations.id"), nullable=True
|
|
)
|
|
device_code: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
|
|
calibration_snapshot: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
|
|
engine_version: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
|
|
note: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
|
|
expected_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
|
|
acquired_at: Mapped[datetime] = mapped_column(
|
|
DateTime, nullable=False, server_default=func.now()
|
|
)
|
|
|
|
__table_args__ = ({"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"},)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<VisionReferenceImage task={self.task_id} path={self.path}>"
|