149 lines
5.4 KiB
Python
149 lines
5.4 KiB
Python
"""Assemble the scope-centric fit guide.
|
|
|
|
Builds the data for the top-level ``/scopes`` page: one row per augmented part
|
|
of part type "Scope", one column per non-Scope part type, where each cell lists
|
|
the parts of that type that fit the row's scope.
|
|
"""
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.crud import attachment_point as attachment_point_crud
|
|
from app.crud import augmented_part as augmented_part_crud
|
|
from app.crud import fit as fit_crud
|
|
from app.crud import part_type as part_type_crud
|
|
from app.crud import scope_info as scope_info_crud
|
|
from app.models.fit import FitLabel
|
|
|
|
SCOPE_TYPE_NAME = "Scope"
|
|
OBJECTIVE_ATTACHMENT_POINT_NAME = "Objective"
|
|
|
|
# "exact" outranks "possible" when the same part fits a scope via several fits.
|
|
_LABEL_RANK = {FitLabel.EXACT.value: 2, FitLabel.POSSIBLE.value: 1}
|
|
|
|
|
|
async def build(
|
|
db: AsyncSession,
|
|
*,
|
|
description: str = "",
|
|
manufacturer_id: int | None = None,
|
|
scope_model_id: int | None = None,
|
|
sub_part_type_id: int | None = None,
|
|
fits_part_type_id: int | None = None,
|
|
) -> dict:
|
|
"""Return the context for ``scopes.html``.
|
|
|
|
The scope rows are narrowed by the scope-attribute filters (``description``
|
|
matches the scope's description; the ``*_id`` filters match exactly).
|
|
``fits_part_type_id`` keeps only scopes that have at least one (non-``none``)
|
|
fit for that part type.
|
|
|
|
Shape::
|
|
|
|
{
|
|
"column_types": [PartType, ...], # non-Scope, ordered by name
|
|
"rows": [
|
|
{
|
|
"scope": AugmentedPart,
|
|
"info": ScopeInfo | None,
|
|
"objective_attachment_point": AttachmentPoint | None,
|
|
"cells": {part_type_id: [{"part": AugmentedPart, "label": str}]},
|
|
},
|
|
...,
|
|
],
|
|
}
|
|
"""
|
|
part_types = await part_type_crud.list_all(db, limit=1000)
|
|
scope_type = next((pt for pt in part_types if pt.name == SCOPE_TYPE_NAME), None)
|
|
column_types = [
|
|
pt for pt in part_types if scope_type is None or pt.id != scope_type.id
|
|
]
|
|
|
|
if scope_type is None:
|
|
return {"column_types": column_types, "rows": []}
|
|
|
|
scopes = await augmented_part_crud.list_filtered(
|
|
db,
|
|
description=description,
|
|
manufacturer_id=manufacturer_id,
|
|
scope_model_id=scope_model_id,
|
|
sub_part_type_id=sub_part_type_id,
|
|
part_type_id=scope_type.id,
|
|
limit=1000,
|
|
)
|
|
scope_ids = {s.id for s in scopes}
|
|
if not scope_ids:
|
|
return {"column_types": column_types, "rows": []}
|
|
|
|
fits = await fit_crud.list_for_parts(
|
|
db, list(scope_ids), exclude_label=FitLabel.NONE
|
|
)
|
|
|
|
# Resolve the "other" (non-scope) part on each fit in a single batched query.
|
|
other_ids: set[int] = set()
|
|
for fit in fits:
|
|
scope_id, other_id = _scope_and_other(fit, scope_ids)
|
|
if scope_id is not None and other_id not in scope_ids:
|
|
other_ids.add(other_id)
|
|
other_parts = {
|
|
p.id: p for p in await augmented_part_crud.get_many(db, list(other_ids))
|
|
}
|
|
|
|
objective_aps = {
|
|
ap.augmented_part_id: ap
|
|
for ap in await attachment_point_crud.list_by_augmented_parts_and_name(
|
|
db, list(scope_ids), OBJECTIVE_ATTACHMENT_POINT_NAME
|
|
)
|
|
}
|
|
|
|
infos = {
|
|
info.augmented_part_id: info
|
|
for info in await scope_info_crud.list_by_augmented_parts(db, list(scope_ids))
|
|
}
|
|
|
|
# cell_map[scope_id][type_id][other_part_id] -> {"part", "label"}, dedup'd so a
|
|
# part appears once per cell, keeping its strongest fit label.
|
|
cell_map: dict[int, dict[int, dict[int, dict]]] = {sid: {} for sid in scope_ids}
|
|
for fit in fits:
|
|
scope_id, other_id = _scope_and_other(fit, scope_ids)
|
|
if scope_id is None:
|
|
continue
|
|
other = other_parts.get(other_id)
|
|
if other is None or other.part_type_id == scope_type.id:
|
|
continue # scope-to-scope fits have no column
|
|
by_type = cell_map[scope_id].setdefault(other.part_type_id, {})
|
|
existing = by_type.get(other.id)
|
|
label = fit.label.value
|
|
if existing is None or _LABEL_RANK[label] > _LABEL_RANK[existing["label"]]:
|
|
by_type[other.id] = {"part": other, "label": label}
|
|
|
|
rows = []
|
|
for scope in scopes:
|
|
cells = {
|
|
type_id: sorted(
|
|
entries.values(), key=lambda e: e["part"].description.lower()
|
|
)
|
|
for type_id, entries in cell_map[scope.id].items()
|
|
}
|
|
if fits_part_type_id is not None and fits_part_type_id not in cells:
|
|
continue # scope has no fit for the required accessory type
|
|
rows.append(
|
|
{
|
|
"scope": scope,
|
|
"info": infos.get(scope.id),
|
|
"objective_attachment_point": objective_aps.get(scope.id),
|
|
"cells": cells,
|
|
}
|
|
)
|
|
|
|
return {"column_types": column_types, "rows": rows}
|
|
|
|
|
|
def _scope_and_other(fit, scope_ids: set[int]) -> tuple[int | None, int]:
|
|
"""Return (scope_part_id, other_part_id) for a fit, or (None, _) if neither
|
|
side is a scope. When both sides are scopes, side A is treated as the scope."""
|
|
if fit.augmented_part_a_id in scope_ids:
|
|
return fit.augmented_part_a_id, fit.augmented_part_b_id
|
|
if fit.augmented_part_b_id in scope_ids:
|
|
return fit.augmented_part_b_id, fit.augmented_part_a_id
|
|
return None, fit.augmented_part_b_id
|