from datetime import datetime
from typing import Dict, Optional
from enum import Enum
import json
import base64
from pydantic import BaseModel, Field, field_validator, model_validator, model_serializer
__all__ = [
"DetectionVerificationStatus",
"TriageEventReviewStatus",
"TaskType",
"WatchboxModel",
"GeoJSON",
"TriageEvent",
"WatchBox",
"Cluster",
"CreateWatchboxRequest",
"UpdateWatchboxRequest",
"WatchboxStatistics",
"WatchboxDetection",
"ProviderOption",
"ImageryProvider",
"ChariotRequest",
]
[docs]
class DetectionVerificationStatus(str, Enum):
NotReviewed = "not_reviewed"
NeedsReview = "needs_review"
Verified = "verified"
Rejected = "rejected"
[docs]
class TriageEventReviewStatus(str, Enum):
Reviewed = "reviewed"
NotReviewed = "not_reviewed"
[docs]
class TaskType(str, Enum):
ObjectDetection = "Object Detection"
OrientedObjectDetection = "Oriented Object Detection"
class BaseModelWithValidators(BaseModel):
@model_validator(mode="before")
@classmethod
def empty_string_to_none(cls, data):
if isinstance(data, dict):
return {k: (None if v == "" else v) for k, v in data.items()}
return data
[docs]
class WatchboxModel(BaseModelWithValidators):
id: str
score_threshold: Optional[float] = None
[docs]
class GeoJSON(BaseModelWithValidators):
"""
GeoJSON object, either a Polygon or a Point.
"""
# Note: could not use tuple[float, float] because OpenAI's
# API rejects it for use in MCP. So instead used list[float]
coordinates: list[list[list[float]]] | list[float]
type: str
[docs]
class ChariotRequest(BaseModelWithValidators):
id: str
date_created: datetime | None = None
date_processing_completed: datetime | None = None
date_updated: datetime | None = None
execution_count: int | None = None
imagery: str | None = None
model_id: str | None = None
name: str | None = None
result: str | None = None
status: str | None = None # type?
triage_event_id: str | None = None
[docs]
class TriageEvent(BaseModelWithValidators):
id: str
chariot_request: list[ChariotRequest] | None = None
date_created: datetime | None = None
date_updated: datetime | None = None
imagery_request: dict | None = None
review_status: TriageEventReviewStatus | None = None
task_type: TaskType | None = None
watchbox_id: str | None = Field(None, validation_alias="watch_point")
area: GeoJSON | None = None
[docs]
class WatchBox(BaseModelWithValidators):
id: str
active: bool
area: GeoJSON | None = None
date_created: datetime | None = None
date_updated: datetime | None = None
description: str | None = None
imagery_providers: list[str] | None = None
last_checked: datetime | None = None
models: list[WatchboxModel] | None = None
name: str | None = None
project_id: str | None = None
provider_filters: list[dict] | None = None
triage_events: list[TriageEvent] | None = None
[docs]
class Cluster(BaseModelWithValidators):
id: str
count: int | None = None
centroid: GeoJSON | None = None
[docs]
class CreateWatchboxRequest(BaseModelWithValidators):
project_id: str
name: str | None = None
description: str | None = None
area: GeoJSON | None = None
imagery_providers: list[str] | None = None
models: list[WatchboxModel] | None = None
provider_filters: list[dict] | None = None
[docs]
class UpdateWatchboxRequest(BaseModelWithValidators):
active: bool | None = None
name: str | None = None
description: str | None = None
imagery_providers: list[str] | None = None
models: list[WatchboxModel] | None = None
provider_filters: list[dict] | None = None
[docs]
class WatchboxStatistics(BaseModelWithValidators):
triage_event_id: str | None = None
var_date: datetime | None = None
counts: Dict[str, int] | None = None
[docs]
class WatchboxDetection(BaseModelWithValidators):
id: str | None = None
area: GeoJSON | None = None
bbox: str | None = None
chariot_request: str | None = None
chip: str | None = None
classification: str | None = None
created_by: str | None = None
dataset: str | None = None
date_created: datetime | None = None
date_updated: datetime | None = None
deleted_at: datetime | None = None
detection_metadata: str | dict | None = None
inference_id: str | None = None
lat: float | int | None = None
lon: float | int | None = None
reviewed_by: str | None = None
score: float | int | None = None
source_detection: str | None = None
time_of_emission: datetime | None = None
triage_event: str | None = None
user_created: bool | None = None
verification_status: str | None = None
inference_store_detection_id: str | None = None
model_id: str | None = None
[docs]
@field_validator("detection_metadata", mode="before")
@classmethod
def convert_to_dict(cls, v):
"""
Watchbox api appears to return base64 encoded json strings for
the detection_metadata. This makes them readable.
"""
if v is None or isinstance(v, dict):
return v
if isinstance(v, str):
try:
return json.loads(v)
except json.JSONDecodeError:
try:
decoded = base64.b64decode(v)
return json.loads(decoded)
except Exception:
return v
return v
[docs]
class ProviderOption(BaseModelWithValidators):
default: str | None = None
description: str | None = None
label: str | None = None
name: str | None = None
options: list[str] | None = None
required: bool | None = None
type: str | None = None
[docs]
class ImageryProvider(BaseModelWithValidators):
id: str
name: str | None = None
handler_id: str | None = None
date_created: datetime | None = None
date_updated: datetime | None = None