Add genlocke lineage tracking with aligned timeline view

Implement read-only lineage view that traces Pokemon across genlocke legs
via existing transfer records. Backend walks transfer chains to build
lineage entries; frontend renders them as cards with a column-aligned
timeline grid so leg dots line up vertically across all lineages.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Julian Tabel
2026-02-09 11:58:38 +01:00
parent 4e00e3cad8
commit d3b65e3c79
9 changed files with 541 additions and 8 deletions

View File

@@ -20,12 +20,15 @@ from app.schemas.genlocke import (
GenlockeDetailResponse,
GenlockeGraveyardResponse,
GenlockeLegDetailResponse,
GenlockeLineageResponse,
GenlockeListItem,
GenlockeResponse,
GenlockeStatsResponse,
GenlockeUpdate,
GraveyardEntryResponse,
GraveyardLegSummary,
LineageEntry,
LineageLegEntry,
RetiredPokemonResponse,
SurvivorResponse,
)
@@ -261,6 +264,171 @@ async def get_genlocke_graveyard(
)
@router.get(
"/{genlocke_id}/lineages",
response_model=GenlockeLineageResponse,
)
async def get_genlocke_lineages(
genlocke_id: int, session: AsyncSession = Depends(get_session)
):
# Load genlocke with legs + game
result = await session.execute(
select(Genlocke)
.where(Genlocke.id == genlocke_id)
.options(
selectinload(Genlocke.legs).selectinload(GenlockeLeg.game),
)
)
genlocke = result.scalar_one_or_none()
if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found")
# Query all transfers for this genlocke
transfer_result = await session.execute(
select(GenlockeTransfer).where(
GenlockeTransfer.genlocke_id == genlocke_id
)
)
transfers = transfer_result.scalars().all()
if not transfers:
return GenlockeLineageResponse(lineages=[], total_lineages=0)
# Build forward/backward maps
forward: dict[int, GenlockeTransfer] = {} # source_encounter_id -> transfer
backward: set[int] = set() # target_encounter_ids
for t in transfers:
forward[t.source_encounter_id] = t
backward.add(t.target_encounter_id)
# Find roots: sources that are NOT targets
roots = [t.source_encounter_id for t in transfers if t.source_encounter_id not in backward]
# Deduplicate while preserving order
seen_roots: set[int] = set()
unique_roots: list[int] = []
for r in roots:
if r not in seen_roots:
seen_roots.add(r)
unique_roots.append(r)
# Walk forward from each root to build chains
chains: list[list[int]] = []
for root in unique_roots:
chain = [root]
current = root
while current in forward:
target = forward[current].target_encounter_id
chain.append(target)
current = target
chains.append(chain)
# Batch-load all encounters in the chains
all_encounter_ids: set[int] = set()
for chain in chains:
all_encounter_ids.update(chain)
enc_result = await session.execute(
select(Encounter)
.where(Encounter.id.in_(all_encounter_ids))
.options(
selectinload(Encounter.pokemon),
selectinload(Encounter.current_pokemon),
selectinload(Encounter.route),
)
)
encounter_map: dict[int, Encounter] = {
enc.id: enc for enc in enc_result.scalars().all()
}
# Build run_id -> (leg_order, game_name) lookup
run_lookup: dict[int, tuple[int, str]] = {}
for leg in genlocke.legs:
if leg.run_id is not None:
run_lookup[leg.run_id] = (leg.leg_order, leg.game.name)
# Load HoF encounter IDs for all runs
run_ids = [leg.run_id for leg in genlocke.legs if leg.run_id is not None]
hof_encounter_ids: set[int] = set()
if run_ids:
run_result = await session.execute(
select(NuzlockeRun.id, NuzlockeRun.hof_encounter_ids).where(
NuzlockeRun.id.in_(run_ids)
)
)
for row in run_result:
if row.hof_encounter_ids:
hof_encounter_ids.update(row.hof_encounter_ids)
# Build lineage entries
lineages: list[LineageEntry] = []
for chain in chains:
legs: list[LineageLegEntry] = []
first_pokemon = None
for enc_id in chain:
enc = encounter_map.get(enc_id)
if enc is None:
continue
leg_order, game_name = run_lookup.get(enc.run_id, (0, "Unknown"))
is_alive = enc.faint_level is None and enc.status == "caught"
entered_hof = enc.id in hof_encounter_ids
was_transferred = enc.id in forward
pokemon_resp = PokemonResponse.model_validate(enc.pokemon)
if first_pokemon is None:
first_pokemon = pokemon_resp
current_pokemon_resp = (
PokemonResponse.model_validate(enc.current_pokemon)
if enc.current_pokemon
else None
)
legs.append(
LineageLegEntry(
leg_order=leg_order,
game_name=game_name,
encounter_id=enc.id,
pokemon=pokemon_resp,
current_pokemon=current_pokemon_resp,
nickname=enc.nickname,
catch_level=enc.catch_level,
faint_level=enc.faint_level,
death_cause=enc.death_cause,
is_shiny=enc.is_shiny,
route_name=enc.route.name if enc.route else "Unknown",
is_alive=is_alive,
entered_hof=entered_hof,
was_transferred=was_transferred,
)
)
if not legs or first_pokemon is None:
continue
# Status based on last encounter in the chain
last_leg = legs[-1]
status = "alive" if last_leg.is_alive else "dead"
lineages.append(
LineageEntry(
nickname=legs[0].nickname,
pokemon=first_pokemon,
legs=legs,
status=status,
)
)
# Sort by first leg order, then by encounter ID
lineages.sort(key=lambda l: (l.legs[0].leg_order, l.legs[0].encounter_id))
return GenlockeLineageResponse(
lineages=lineages,
total_lineages=len(lineages),
)
@router.post("", response_model=GenlockeResponse, status_code=201)
async def create_genlocke(
data: GenlockeCreate, session: AsyncSession = Depends(get_session)