serversdown
56bd3041cf
feat: Location no longer assigned directly to unit, locations and coords are assigned to location only, unit only is deployed or benched.
126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""
|
|
Active-assignment location resolution for roster units.
|
|
|
|
`RosterUnit.location`, `.address`, `.coordinates` are legacy per-unit fields.
|
|
The current source of truth for "where is this unit deployed right now" is the
|
|
active `UnitAssignment` (assigned_until IS NULL) pointing at a
|
|
`MonitoringLocation`, which carries the canonical address/coordinates/name.
|
|
|
|
Modems don't get their own `UnitAssignment` — they're paired with a
|
|
seismograph or SLM via `deployed_with_unit_id`. A deployed modem inherits the
|
|
location of its paired device's active assignment.
|
|
|
|
Returned dict shape (or None if no active assignment resolvable):
|
|
{
|
|
"location_id": "uuid",
|
|
"project_id": "uuid",
|
|
"name": "NRL-001",
|
|
"address": "123 Main St" | None,
|
|
"coordinates": "34.0522,-118.2437" | None,
|
|
"via_paired_unit_id": "BE1234" | None, # set only for modems
|
|
}
|
|
"""
|
|
|
|
from typing import Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.models import MonitoringLocation, RosterUnit, UnitAssignment
|
|
|
|
|
|
def _serialize(loc: MonitoringLocation, via_paired_unit_id: Optional[str] = None) -> dict:
|
|
return {
|
|
"location_id": loc.id,
|
|
"project_id": loc.project_id,
|
|
"name": loc.name,
|
|
"address": loc.address or None,
|
|
"coordinates": loc.coordinates or None,
|
|
"via_paired_unit_id": via_paired_unit_id,
|
|
}
|
|
|
|
|
|
def _active_location_for_unit_id(db: Session, unit_id: str) -> Optional[MonitoringLocation]:
|
|
"""Return the MonitoringLocation tied to this unit's active assignment, if any."""
|
|
row = (
|
|
db.query(MonitoringLocation)
|
|
.join(UnitAssignment, UnitAssignment.location_id == MonitoringLocation.id)
|
|
.filter(
|
|
UnitAssignment.unit_id == unit_id,
|
|
UnitAssignment.assigned_until == None, # noqa: E711
|
|
)
|
|
.order_by(UnitAssignment.assigned_at.desc())
|
|
.first()
|
|
)
|
|
return row
|
|
|
|
|
|
def get_active_location(db: Session, unit_id: str) -> Optional[dict]:
|
|
"""
|
|
Resolve the active deployment location for a unit.
|
|
|
|
Seismographs / SLMs: their own active UnitAssignment.
|
|
Modems: follow `deployed_with_unit_id` to the paired device's active
|
|
assignment (modems don't carry their own assignment).
|
|
"""
|
|
unit = db.query(RosterUnit).filter_by(id=unit_id).first()
|
|
if unit is None:
|
|
return None
|
|
|
|
if (unit.device_type or "seismograph") == "modem":
|
|
paired_id = unit.deployed_with_unit_id
|
|
if not paired_id:
|
|
return None
|
|
loc = _active_location_for_unit_id(db, paired_id)
|
|
return _serialize(loc, via_paired_unit_id=paired_id) if loc else None
|
|
|
|
loc = _active_location_for_unit_id(db, unit_id)
|
|
return _serialize(loc) if loc else None
|
|
|
|
|
|
def bulk_active_locations(db: Session, units: list[RosterUnit]) -> dict[str, dict]:
|
|
"""
|
|
Resolve active locations for many units in two queries. Use this from
|
|
snapshot-style loops to avoid N+1 lookups.
|
|
|
|
Returns {unit_id: <serialized location dict>} — only populated for units
|
|
that resolve to an active assignment. Modems are resolved by walking
|
|
`deployed_with_unit_id` to the paired device's entry in the same map.
|
|
"""
|
|
if not units:
|
|
return {}
|
|
|
|
direct_unit_ids = [
|
|
u.id for u in units
|
|
if (u.device_type or "seismograph") != "modem"
|
|
]
|
|
|
|
direct: dict[str, MonitoringLocation] = {}
|
|
if direct_unit_ids:
|
|
rows = (
|
|
db.query(UnitAssignment.unit_id, MonitoringLocation)
|
|
.join(MonitoringLocation, MonitoringLocation.id == UnitAssignment.location_id)
|
|
.filter(
|
|
UnitAssignment.unit_id.in_(direct_unit_ids),
|
|
UnitAssignment.assigned_until == None, # noqa: E711
|
|
)
|
|
.order_by(UnitAssignment.assigned_at.desc())
|
|
.all()
|
|
)
|
|
# First row wins per unit_id (most recent assigned_at).
|
|
for unit_id, loc in rows:
|
|
direct.setdefault(unit_id, loc)
|
|
|
|
out: dict[str, dict] = {
|
|
uid: _serialize(loc) for uid, loc in direct.items()
|
|
}
|
|
|
|
# Modems inherit from paired device.
|
|
for u in units:
|
|
if (u.device_type or "seismograph") != "modem":
|
|
continue
|
|
paired_id = u.deployed_with_unit_id
|
|
if paired_id and paired_id in direct:
|
|
out[u.id] = _serialize(direct[paired_id], via_paired_unit_id=paired_id)
|
|
|
|
return out
|