Inspectra- DB schemas/tables

[canelbirlik]

Implement relevant db tables to cover query needs for all pages.

• Take previous schemas,
• identify missing fields from page queries and whether we can cover them.
• Mimick production db for model related outputs.

Document instruction to guide Client team to follow the design on production project.

[nanwang-mindtrace]

Can will share DB schema table diagram or the actual tables when task is done.

[nanwang-mindtrace]

Can is confirming w/ Vineeth and then will record the finalised design & instruction

[canelbirlik]

Overall Design & Agreed initial Schema:

Collections(tables):

• organizations, plants, lines, parts, cameras are catalogs.
• part_scans : one per inspected item (immutable parent).
• location_scans : one per camera capture (immutable child).
• media : URIs to images/masks/derived artifacts.
• inferences — hot path; one row per semantic task output from a run.

Design points:

• Flat Inference collection: queryable main fields plus contract specific extra payloads
• Semantic contracts: contract_name: "visual_qlty_inspection" | "presence_check" | "laser_dimension" | etc. UI & backend read a registry to know how to render/expect fields.

Contract Management between UI&Backend.
A semantic name that both backend & UI understand. It describes how to interpret an Inference row’s canonical fields and its extras.

example:

.
We are to store it in the config collection, 
// Config.scope=org, key="contracts.registry"
-> 
  "visual_qlty_inspection": {
    "summaryFields": ["label", "severity", "confidence"],
    "expects": ["cls", "det", "severity"],
    "extrasSchemaVersion": 1,
    "ui": { "renderer": "VisualQualityCard", "icons": ["weld","crack"] }
  },
  "sfz": {
    "summaryFields": ["label", "presence"],
    "expects": ["presence", "count],
    "ui": { "renderer": "SpatterContractCard" }
  },
  "laser": {
    "summaryFields": ["value", "unit"],
    "expects": ["measure","det","segmentation"],
    "ui": { "renderer": "LaserContractCard" }
  },
  "generic": {
    "summaryFields": ["label","confidence","present","quantity","severity","value","unit"],
    "ui": { "renderer": "GenericInferenceCard" }
  }

in turn UI would implement VisualInspectionCard and know to render it.

from __future__ import annotations
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Union, Literal

from beanie import Link, PydanticObjectId
from pydantic import BaseModel, Field, EmailStr
from pymongo import IndexModel, ASCENDING, DESCENDING

UTC = timezone.utc

# ========= Enums / Literals =========
MediaKind   = Literal["image", "mask", "heatmap", "other"]
UserRole    = Literal["admin", "qc", "viewer"]
UserPersona = Literal["plant_manager", "CEO"]

# ========= Config (scoped key/value) =========
class Config(MindtraceDocument):
    scope: Literal["org", "plant", "line"]
    scope_ref: PydanticObjectId
    key: str
    value: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "configs"
        indexes = [
            IndexModel([("scope", 1), ("scope_ref", 1), ("key", 1)],
                       name="cfg_scope_ref_key_uq", unique=True)
        ]

# ========= Catalogs =========
class Organization(MindtraceDocument):
    name: str
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "organizations"
        indexes = [IndexModel([("name", 1)], name="org_name_uq", unique=True)]

class Plant(MindtraceDocument):
    org: Link[Organization]
    name: str
    location: Optional[Union[str, Dict[str, Any]]] = None
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "plants"
        indexes = [
            IndexModel([("org.$id", 1), ("name", 1)],
                       name="plant_org_name_uq", unique=True)
        ]

class Part(MindtraceDocument):
    org: Link[Organization]
    code: Union[int, str]
    name: Optional[str] = None
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "parts"
        indexes = [
            IndexModel([("org.$id", 1), ("code", 1)],
                       name="part_org_code_uq", unique=True)
        ]

class Line(MindtraceDocument):
    plant: Link[Plant]
    name: str
    active: bool = True
    supported_part_ids: List[PydanticObjectId] = Field(default_factory=list)
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "lines"
        indexes = [
            IndexModel([("plant.$id", 1), ("name", 1)],
                       name="line_plant_name_uq", unique=True)
        ]

class CameraConfig(BaseModel):
    exposure_ms: Optional[int] = None
    gain: Optional[float] = None
    white_balance: Optional[str] = None
    lens: Optional[str] = None

class Camera(MindtraceDocument):
    line: Link[Line]
    camera_id: str                   # device/location id
    name: str                        # e.g. "top_left"
    config: CameraConfig = Field(default_factory=CameraConfig)
    last_updated: datetime = Field(default_factory=lambda: datetime.now(UTC))
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "cameras"
        indexes = [
            IndexModel([("line.$id", 1), ("camera_id", 1)],
                       name="cam_line_cameraid_uq", unique=True),
            IndexModel([("line.$id", 1), ("name", 1)],
                       name="cam_line_name"),
        ]

class User(MindtraceDocument):
    email: EmailStr
    email_norm: str = ""
    role: UserRole = "qc"
    persona: Optional[UserPersona] = None
    name: str
    pw_hash: str
    org_ids: List[PydanticObjectId] = Field(default_factory=list)
    plant_ids: List[PydanticObjectId] = Field(default_factory=list)
    line_ids: List[PydanticObjectId] = Field(default_factory=list)
    status: Literal["active", "inactive"] = "active"
    meta: Dict[str, Any] = Field(default_factory=dict)

    async def before_save(self):
        self.email_norm = self.email.casefold()

    class Settings:
        name = "users"
        indexes = [
            IndexModel([("email_norm", 1)],
                       name="user_email_norm_uq", unique=True)
        ]

# ========= Scans & Media =========
class PartScan(MindtraceDocument):  # immutable parent
    line: Link[Line]
    plant: Link[Plant]
    org: Link[Organization]
    part: Link[Part]
    part_no: Union[int, str]
    part_name: Optional[str] = None
    serial: Optional[str] = None
    scanned_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    is_defective: Literal["Unknown", "Healthy", "Defective"] = "Unknown"
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "part_scans"
        indexes = [
            IndexModel([("line.$id", 1), ("scanned_at", -1)], name="ps_line_scanned"),
            IndexModel([("plant.$id", 1), ("scanned_at", -1)], name="ps_plant_scanned"),
            IndexModel([("org.$id", 1), ("scanned_at", -1)], name="ps_org_scanned"),
            IndexModel([("part_no", 1), ("scanned_at", -1)], name="ps_part_time"),
        ]

class Media(MindtraceDocument):
    kind: MediaKind
    uri: str  # relative/path or s3://...
    hash: Optional[str] = None
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "media"
        indexes = [
            IndexModel([("hash", 1)], name="media_hash_uq", unique=True, sparse=True),
            IndexModel([("created_at", -1)], name="media_created"),
        ]

class LocationScan(MindtraceDocument):  # per camera/location shot
    part_scan: Link[PartScan]
    line: Link[Line]
    plant: Link[Plant]
    org: Link[Organization]
    camera: Link[Camera]

    location_name: str  # optional extra alias for the location if you want

    # denorm hot filters
    line_id: PydanticObjectId
    line_name: str
    camera_id: PydanticObjectId
    camera_name: str
    part_no: Union[int, str]
    part_name: Optional[str] = None

    captured_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    input_media: List[Link[Media]] = Field(default_factory=list)
    meta: Dict[str, Any] = Field(default_factory=dict)
    is_defective: Literal["Unknown", "Healthy", "Defective"] = "Unknown"

    class Settings:
        name = "location_scans"
        indexes = [
            IndexModel([("part_scan.$id", 1), ("captured_at", 1)], name="ls_part_time"),
            IndexModel([("line_id", 1), ("camera_name", 1), ("captured_at", -1)], name="ls_line_cam_time"),
            IndexModel([("camera_id", 1), ("captured_at", -1)], name="ls_camera_time"),
        ]

# ========= Model registry =========
class ModelCapability(BaseModel):
    # Loose & future-proof; add keys as you standardize
    type: Literal["classification", "detection", "regression", "presence", "segmentor", "measurement"]
    labels: Optional[List[str]] = None
    metrics: Optional[List[str]] = None
    thresholds: Dict[str, Any] = Field(default_factory=dict)

class Model(MindtraceDocument):
    name: str
    version: str
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    capabilities: List[ModelCapability]
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "models"
        indexes = [
            IndexModel([("name", 1), ("version", 1)],
                       name="model_name_version_uq", unique=True)
        ]

# ========= Inference (single-result row; semantic contract + base fields + extras) =========
class Inference(MindtraceDocument):
    # links + denorm (for fast filters)
    location_scan: Link[LocationScan]
    location_scan_id: PydanticObjectId

    part_scan: Link[PartScan]
    part_no: Union[int, str]
    part_name: Optional[str] = None

    line: Link[Line]
    line_id: PydanticObjectId
    line_name: str

    plant: Link[Plant]
    plant_id: PydanticObjectId

    org: Link[Organization]
    org_id: PydanticObjectId

    camera: Link[Camera]
    camera_id: PydanticObjectId
    camera_name: str

    model: Link[Model]
    model_ver: str

    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    run_id: Optional[Union[str, PydanticObjectId]] = None

    # semantic grouping only (no enforcement right now)
    contract_name: str   # e.g., "mig66", "sfz", "laser", "paslin", "generic"

    # canonical queryable fields (all optional for now)
    # classification
    label: Optional[str] = None
    confidence: Optional[float] = None
    # presence
    present: Optional[bool] = None
    quantity: Optional[float] = None
    # severity / regression
    value: Optional[float] = None
    severity: Optional[float] = None
    unit: Optional[str] = None
    # detection / geometry
    bboxes: Optional[List[Dict[str, Any]]] = None
    # keypoints
    kpoints: Optional[List[Dict[str, Any]]] = None

    # non-indexed, contract-specific payload
    extras: Dict[str, Any] = Field(default_factory=dict)
    extras_ver: int = 1

    input_media: Optional[Link[Media]] = None
    artifacts: List[Link[Media]] = Field(default_factory=list)
    meta: Dict[str, Any] = Field(default_factory=dict)

    class Settings:
        name = "inferences"
        indexes = [
            # scope & time
            IndexModel([("org_id", 1), ("plant_id", 1), ("line_id", 1), ("created_at", -1)],
                       name="inf_scope_time"),
            IndexModel([("line_id", 1), ("created_at", -1)],
                       name="inf_line_time"),
            IndexModel([("line_name", 1), ("created_at", -1)],
                       name="inf_linename_time"),
            IndexModel([("camera_name", 1), ("created_at", -1)],
                       name="inf_camname_time"),
            IndexModel([("part_no", 1), ("created_at", -1)],
                       name="inf_part_time"),
            IndexModel([("contract_name", 1), ("created_at", -1)],
                       name="inf_contract_time"),
            # (optional) one row per (capture, model, version, contract, label)
            IndexModel([("location_scan.$id", 1), ("model.$id", 1), ("model_ver", 1),
                        ("contract_name", 1), ("label", 1)],
                       name="inf_unique", unique=True, sparse=True),
        ]