feat(unit-swap): show benched candidates and clean stale modem pairings

`available-units` and `available-modems` now accept `include_benched=true`
to also return units/modems with `deployed=False`.  Default is False so
the existing location-detail swap modal is unchanged.  Each row carries
a `deployed` boolean for badge rendering.  The Unit Swap wizard fetches
with the flag enabled — exactly the candidates a field tech pulls off
the shelf.

The /swap endpoint now flips the incoming unit (and modem) back to
`deployed=True` when they came in benched, keeping the legacy roster
flag consistent with the active-assignment signal.

Adds the symmetric half of the orphan-pairing fix: when a newly-paired
modem still claims a different seismograph (whose
`deployed_with_modem_id` was never cleared in a past swap), break that
stale back-reference before re-pairing.

`locations-with-assignments` includes `modem.deployed` so the wizard
can badge the current modem in the location card, the "Keep current
modem" choice, the picker rows, and the review screen.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-18 04:46:23 +00:00
parent 44ab4d8427
commit 6d37bd759e
2 changed files with 93 additions and 32 deletions
+52 -16
View File
@@ -318,6 +318,7 @@ async def get_locations_with_assignments(
"hardware_model": modem.hardware_model,
"ip_address": modem.ip_address,
"phone_number": modem.phone_number,
"deployed": bool(modem.deployed),
}
results.append({
@@ -1306,12 +1307,28 @@ async def swap_unit_on_location(
modem = db.query(RosterUnit).filter_by(id=modem_id, device_type="modem").first()
if not modem:
raise HTTPException(status_code=404, detail=f"Modem '{modem_id}' not found")
# Symmetric cleanup: if this modem still claims a previous partner
# (a different seismograph whose deployed_with_modem_id never got
# cleared in a past swap), break that stale link before re-pairing.
if modem.deployed_with_unit_id and modem.deployed_with_unit_id != unit_id:
prev_partner = db.query(RosterUnit).filter_by(id=modem.deployed_with_unit_id).first()
if prev_partner and prev_partner.deployed_with_modem_id == modem_id:
prev_partner.deployed_with_modem_id = None
unit.deployed_with_modem_id = modem_id
modem.deployed_with_unit_id = unit_id
# If the modem was on the bench, swapping it into the field puts it
# back in rotation.
if not modem.deployed:
modem.deployed = True
else:
# Clear modem pairing if not provided
unit.deployed_with_modem_id = None
# If the incoming unit was benched, putting it in the field flips it
# back to deployed (so polling / dashboards see it as in rotation).
if not unit.deployed:
unit.deployed = True
db.commit()
return JSONResponse({
@@ -1329,23 +1346,32 @@ async def swap_unit_on_location(
async def get_available_modems(
project_id: str,
db: Session = Depends(get_db),
include_benched: bool = Query(False),
):
"""
Get all deployed, non-retired modems for the modem assignment dropdown.
Get all non-retired modems for the modem assignment dropdown.
By default only deployed (in-rotation) modems are returned, preserving
the existing behavior for callers like the location-detail swap modal.
Pass ``include_benched=true`` to also include benched modems
(``RosterUnit.deployed == False``) — useful when picking a modem to
pull off the bench for a field swap. Each row's ``deployed`` flag is
returned so the UI can badge benched candidates.
"""
modems = db.query(RosterUnit).filter(
and_(
RosterUnit.device_type == "modem",
RosterUnit.deployed == True,
RosterUnit.retired == False,
)
).order_by(RosterUnit.id).all()
filters = [
RosterUnit.device_type == "modem",
RosterUnit.retired == False,
]
if not include_benched:
filters.append(RosterUnit.deployed == True)
modems = db.query(RosterUnit).filter(and_(*filters)).order_by(RosterUnit.id).all()
return [
{
"id": m.id,
"hardware_model": m.hardware_model,
"ip_address": m.ip_address,
"deployed": bool(m.deployed),
}
for m in modems
]
@@ -1356,22 +1382,31 @@ async def get_available_units(
project_id: str,
location_type: str = Query(...),
db: Session = Depends(get_db),
include_benched: bool = Query(False),
):
"""
Get list of available units for assignment to a location.
Filters by device type matching the location type.
By default only deployed (in-rotation) units are returned, preserving
the existing location-detail swap-modal behavior. Pass
``include_benched=true`` to also include benched units
(``RosterUnit.deployed == False``) — exactly the candidates you'd
pull off the bench for a field swap. Each row carries a ``deployed``
flag so the UI can badge benched picks.
"""
# Determine required device type
required_device_type = "slm" if location_type == "sound" else "seismograph"
# Get all units of the required type that are deployed and not retired
all_units = db.query(RosterUnit).filter(
and_(
RosterUnit.device_type == required_device_type,
RosterUnit.deployed == True,
RosterUnit.retired == False,
)
).all()
# Get all units of the required type that aren't retired (and optionally
# exclude benched units).
filters = [
RosterUnit.device_type == required_device_type,
RosterUnit.retired == False,
]
if not include_benched:
filters.append(RosterUnit.deployed == True)
all_units = db.query(RosterUnit).filter(and_(*filters)).all()
# Filter out units that already have active assignments (active = assigned_until IS NULL)
assigned_unit_ids = db.query(UnitAssignment.unit_id).filter(
@@ -1385,6 +1420,7 @@ async def get_available_units(
"device_type": unit.device_type,
"location": unit.address or unit.location,
"model": unit.slm_model if unit.device_type == "slm" else unit.unit_type,
"deployed": bool(unit.deployed),
}
for unit in all_units
if unit.id not in assigned_unit_ids