103 lines
4.1 KiB
Python
103 lines
4.1 KiB
Python
"""Application service for the consolidated Repair Center read model."""
|
|
|
|
from app.repairs.center_repository import RepairCenterRepository
|
|
from app.repairs.models import (
|
|
ActiveRepairSummary,
|
|
RepairCenterAccess,
|
|
RepairCenterCapabilities,
|
|
RepairCenterFilters,
|
|
RepairCenterItem,
|
|
RepairCenterList,
|
|
RepairCenterLocationOptions,
|
|
RepairCenterMeta,
|
|
RepairCenterResponse,
|
|
RepairCenterSummary,
|
|
)
|
|
|
|
|
|
class RepairCenterService:
|
|
"""Build typed, permission-aware Repair Center projections."""
|
|
|
|
def __init__(self, repository: RepairCenterRepository):
|
|
self.repository = repository
|
|
|
|
@staticmethod
|
|
def _capabilities(
|
|
item: RepairCenterItem,
|
|
access: RepairCenterAccess,
|
|
) -> RepairCenterCapabilities:
|
|
modular = item.source == "modular" and item.repair_id is not None
|
|
legacy = item.source == "legacy"
|
|
placed = bool(item.site_id and item.room_id and item.position_id)
|
|
return RepairCenterCapabilities(
|
|
can_open_miner=access.can_view_miners,
|
|
can_open_map=access.can_view_locations and placed,
|
|
can_read_modular_timeline=modular and access.can_read_repairs,
|
|
can_read_modular_parts=modular and access.can_read_repairs,
|
|
can_manage_modular_parts=(
|
|
modular and item.active and access.can_read_repairs
|
|
and access.can_manage_parts
|
|
),
|
|
can_complete_modular_repair=(
|
|
modular and item.active and access.can_complete_repairs
|
|
),
|
|
can_read_legacy_history=legacy and access.can_read_repairs,
|
|
can_manage_legacy=legacy and access.can_manage_legacy,
|
|
can_mark_legacy_repaired=legacy and access.can_manage_legacy,
|
|
can_decommission_legacy=legacy and access.can_manage_legacy,
|
|
)
|
|
|
|
def list_repair_center(
|
|
self,
|
|
filters: RepairCenterFilters,
|
|
access: RepairCenterAccess,
|
|
) -> RepairCenterResponse:
|
|
query = filters.query.strip() if filters.query else None
|
|
model = filters.model.strip() if filters.model else None
|
|
status = filters.status.strip() if filters.status else None
|
|
repository_filters = filters.model_dump()
|
|
repository_filters.update(query=query or None, model=model or None,
|
|
status=status or None)
|
|
projection = self.repository.list_repair_center(
|
|
repository_filters,
|
|
include_miner_details=access.can_view_miners,
|
|
include_location_details=access.can_view_locations,
|
|
)
|
|
items: list[RepairCenterItem] = []
|
|
for row in projection["items"]:
|
|
row = {
|
|
**row,
|
|
"miner_status": row.get("runtime_status"),
|
|
"latest_event_status": row.get("legacy_event_status"),
|
|
"latest_event_description": row.get("latest_context"),
|
|
"latest_event_at": row.get("latest_context_at"),
|
|
"asset_state": row.get("legacy_asset_state"),
|
|
"state_notes": row.get("latest_context"),
|
|
"reported_error": row.get("reported_problem"),
|
|
}
|
|
item = RepairCenterItem.model_validate(
|
|
{**row, "capabilities": RepairCenterCapabilities()}
|
|
)
|
|
item.capabilities = self._capabilities(item, access)
|
|
items.append(item)
|
|
|
|
summary = RepairCenterSummary.model_validate(projection["summary"])
|
|
location_options = RepairCenterLocationOptions.model_validate(
|
|
projection.get("location_options") or {}
|
|
)
|
|
return RepairCenterResponse(
|
|
data=RepairCenterList(items=items, summary=summary),
|
|
meta=RepairCenterMeta(
|
|
limit=filters.limit,
|
|
offset=filters.offset,
|
|
returned=len(items),
|
|
total=summary.total,
|
|
location_options=location_options,
|
|
),
|
|
)
|
|
|
|
def summarize_active_repairs(self) -> ActiveRepairSummary:
|
|
return ActiveRepairSummary.model_validate(
|
|
self.repository.summarize_active_repairs()
|
|
)
|