289 lines
7.9 KiB
Python
289 lines
7.9 KiB
Python
"""Typed contracts for the Repairs domain."""
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from typing import Any, Literal
|
|
from uuid import UUID
|
|
|
|
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, model_validator
|
|
from app.components.models import ComponentType
|
|
|
|
|
|
RepairReason = Literal[
|
|
"HARDWARE_FAILURE",
|
|
"LOW_HASHRATE",
|
|
"CONTROL_BOARD",
|
|
"HASHBOARD",
|
|
"PSU",
|
|
"FAN",
|
|
"OVERHEATING",
|
|
"NETWORKING",
|
|
"PREVENTIVE_MAINTENANCE",
|
|
"INSPECTION",
|
|
"OTHER",
|
|
]
|
|
RepairStatus = Literal[
|
|
"OPEN",
|
|
"IN_PROGRESS",
|
|
"WAITING_FOR_PARTS",
|
|
"READY_FOR_TEST",
|
|
"COMPLETED",
|
|
"CANCELLED",
|
|
]
|
|
RepairEventType = Literal[
|
|
"REPAIR_STARTED",
|
|
"INITIAL_NOTE",
|
|
"REPAIR_COMPLETED",
|
|
"CONTROL_BOARD_REPLACED",
|
|
"REPAIR_PART_RECORDED",
|
|
"REPAIR_PART_VOIDED",
|
|
]
|
|
RepairEligibleLifecycle = Literal["active", "spare"]
|
|
RepairPostLifecycle = Literal["active", "spare"]
|
|
RepairMinerLifecycle = Literal["active", "spare", "retired", "in_repair"]
|
|
|
|
|
|
class BeginRepairRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
miner_id: UUID
|
|
reason: RepairReason
|
|
notes: str | None = Field(default=None, max_length=4000)
|
|
|
|
|
|
class RepairCompleteRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
target_lifecycle: RepairPostLifecycle
|
|
completion_summary: str | None = Field(default=None, max_length=4000)
|
|
|
|
|
|
ControlBoardReplacementReason = Literal[
|
|
"failed",
|
|
"upgrade",
|
|
"diagnostic_swap",
|
|
"preventive_maintenance",
|
|
"temporary_donor",
|
|
"unknown",
|
|
]
|
|
RemovedBoardDisposition = Literal[
|
|
"repair_inventory",
|
|
"sent_for_repair",
|
|
"scrapped",
|
|
"retained_for_diagnosis",
|
|
"returned_to_vendor",
|
|
"unknown",
|
|
]
|
|
|
|
|
|
class ControlBoardReplacementRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
new_mac_address: str = Field(min_length=1, max_length=64)
|
|
new_control_board_serial: str | None = Field(..., max_length=255)
|
|
board_model: str | None = Field(default=None, max_length=255)
|
|
board_revision: str | None = Field(default=None, max_length=255)
|
|
firmware_version: str | None = Field(default=None, max_length=255)
|
|
removed_board_disposition: RemovedBoardDisposition
|
|
replacement_reason: ControlBoardReplacementReason
|
|
notes: str | None = Field(default=None, max_length=4000)
|
|
|
|
|
|
class RepairStartedPayload(BaseModel):
|
|
reason: RepairReason
|
|
from_lifecycle: RepairEligibleLifecycle
|
|
to_lifecycle: Literal["in_repair"] = "in_repair"
|
|
|
|
|
|
class InitialRepairNotePayload(BaseModel):
|
|
note: str = Field(min_length=1, max_length=4000)
|
|
|
|
|
|
class RepairCompletedPayload(BaseModel):
|
|
previous_miner_lifecycle: Literal["in_repair"] = "in_repair"
|
|
resulting_miner_lifecycle: RepairPostLifecycle
|
|
pre_repair_lifecycle: RepairEligibleLifecycle | None = None
|
|
completion_summary: str | None = Field(default=None, max_length=4000)
|
|
|
|
|
|
class ControlBoardReplacedPayload(BaseModel):
|
|
old_mac_address: str | None
|
|
new_mac_address: str
|
|
old_control_board_serial: str | None
|
|
new_control_board_serial: str | None
|
|
old_board_model: str | None = None
|
|
new_board_model: str | None = None
|
|
old_board_revision: str | None = None
|
|
new_board_revision: str | None = None
|
|
new_firmware_version: str | None = None
|
|
removed_board_disposition: RemovedBoardDisposition
|
|
replacement_reason: ControlBoardReplacementReason
|
|
notes: str | None = Field(default=None, max_length=4000)
|
|
|
|
|
|
class ControlBoardReplacementResult(BaseModel):
|
|
miner_id: UUID
|
|
asset_id: str | None = None
|
|
repair_id: UUID
|
|
installation_id: UUID
|
|
event_id: UUID
|
|
event_sequence: int = Field(ge=1)
|
|
current_mac: str
|
|
control_board_serial: str | None
|
|
replaced_at: AwareDatetime
|
|
replaced_by: UUID
|
|
|
|
|
|
class ControlBoardReplacementResponse(BaseModel):
|
|
data: ControlBoardReplacementResult
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
class BeginRepairResult(BaseModel):
|
|
repair_id: UUID
|
|
miner_id: UUID
|
|
status: Literal["OPEN"]
|
|
reason: RepairReason
|
|
opened_at: datetime
|
|
opened_by: UUID
|
|
|
|
|
|
class BeginRepairResponse(BaseModel):
|
|
data: BeginRepairResult
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
class RepairActor(BaseModel):
|
|
user_id: UUID
|
|
display_name: str
|
|
|
|
|
|
class RepairDetail(BaseModel):
|
|
repair_id: UUID
|
|
miner_id: UUID
|
|
miner_asset_id: str | None = None
|
|
status: RepairStatus
|
|
reason: RepairReason
|
|
opened_at: AwareDatetime
|
|
opened_by: RepairActor
|
|
closed_at: AwareDatetime | None = None
|
|
closed_by: RepairActor | None = None
|
|
miner_lifecycle: RepairMinerLifecycle
|
|
latest_event_sequence: int | None = Field(default=None, ge=1)
|
|
|
|
|
|
class RepairTimelineEvent(BaseModel):
|
|
event_id: UUID
|
|
sequence: int = Field(ge=1)
|
|
event_type: RepairEventType
|
|
actor: RepairActor
|
|
occurred_at: AwareDatetime
|
|
payload: dict[str, Any]
|
|
|
|
|
|
class RepairTimeline(BaseModel):
|
|
repair_id: UUID
|
|
events: list[RepairTimelineEvent] = Field(default_factory=list)
|
|
|
|
|
|
class RepairDetailResponse(BaseModel):
|
|
data: RepairDetail
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
class RepairTimelineResponse(BaseModel):
|
|
data: RepairTimeline
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
ConsumableUnit = Literal["each", "gram", "millilitre", "metre", "centimetre"]
|
|
RepairPartEntryKind = Literal["consumable", "tracked_component"]
|
|
|
|
|
|
class ConsumableInput(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
|
|
description: str = Field(min_length=1, max_length=255)
|
|
quantity: Decimal = Field(gt=0, max_digits=12, decimal_places=3)
|
|
unit: ConsumableUnit
|
|
|
|
@model_validator(mode="after")
|
|
def validate_each_quantity(self) -> "ConsumableInput":
|
|
if self.unit == "each" and self.quantity != self.quantity.to_integral_value():
|
|
raise ValueError("quantity for unit 'each' must be a whole number")
|
|
return self
|
|
|
|
|
|
class RepairPartRecordRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
|
|
consumable: ConsumableInput | None = None
|
|
tracked_component_uuid: UUID | None = None
|
|
notes: str | None = Field(default=None, max_length=4000)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_entry_shape(self) -> "RepairPartRecordRequest":
|
|
if (self.consumable is None) == (self.tracked_component_uuid is None):
|
|
raise ValueError(
|
|
"exactly one of consumable and tracked_component_uuid is required"
|
|
)
|
|
return self
|
|
|
|
|
|
class RepairPartVoidRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
|
|
reason: str = Field(min_length=1, max_length=1000)
|
|
|
|
|
|
class RepairPartConsumable(BaseModel):
|
|
description: str
|
|
quantity: Decimal
|
|
unit: ConsumableUnit
|
|
|
|
|
|
class RepairPartComponentSnapshot(BaseModel):
|
|
component_type: ComponentType
|
|
asset_id: str | None = None
|
|
manufacturer: str | None = None
|
|
model: str | None = None
|
|
manufacturer_serial: str | None = None
|
|
|
|
|
|
class RepairPartDetail(BaseModel):
|
|
repair_part_id: UUID
|
|
repair_id: UUID
|
|
entry_kind: RepairPartEntryKind
|
|
consumable: RepairPartConsumable | None = None
|
|
component_id: UUID | None = None
|
|
component_snapshot: RepairPartComponentSnapshot | None = None
|
|
notes: str | None = None
|
|
recorded_by: UUID
|
|
recorded_at: AwareDatetime
|
|
recorded_event_id: UUID
|
|
voided_by: UUID | None = None
|
|
voided_at: AwareDatetime | None = None
|
|
void_reason: str | None = None
|
|
void_event_id: UUID | None = None
|
|
|
|
|
|
class RepairPartList(BaseModel):
|
|
repair_id: UUID
|
|
entries: list[RepairPartDetail] = Field(default_factory=list)
|
|
|
|
|
|
class RepairPartResponse(BaseModel):
|
|
data: RepairPartDetail
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
class RepairPartListResponse(BaseModel):
|
|
data: RepairPartList
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|