202 lines
7.0 KiB
Python
202 lines
7.0 KiB
Python
"""Detect "Possible" fits between AugmentedParts.
|
|
|
|
A part fits another part if:
|
|
- They are different PartTypes
|
|
- They have two attachment points that:
|
|
- Have different labels (AttachmentPointName)
|
|
- either:
|
|
- Are an ISO or UNC thread that exactly match in type and dimensions
|
|
- Are a Static Connection paired with a Clamp Connection where the
|
|
static dimension falls within the range of the clamp connection
|
|
|
|
Matches found here are recorded as Fit rows labeled Possible. Exact (manually
|
|
entered, or promoted from Possible) and None (explicitly marked impossible)
|
|
labels are user-asserted and are never created or removed by this algorithm.
|
|
"""
|
|
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.crud import fit as fit_crud
|
|
from app.models.attachment_point import AttachmentPoint
|
|
from app.models.augmented_part import AugmentedPart
|
|
from app.models.clamp_connection import ClampConnection
|
|
from app.models.fit import FitLabel
|
|
from app.models.static_connection import StaticConnection
|
|
|
|
MM_PER_INCH = 25.4
|
|
|
|
|
|
def _iso_match(ap_a: AttachmentPoint, ap_b: AttachmentPoint) -> bool:
|
|
a, b = ap_a.iso_thread, ap_b.iso_thread
|
|
if a is None or b is None:
|
|
return False
|
|
return a.diameter_mm == b.diameter_mm and a.pitch_mm == b.pitch_mm
|
|
|
|
|
|
def _unc_match(ap_a: AttachmentPoint, ap_b: AttachmentPoint) -> bool:
|
|
a, b = ap_a.unc_thread, ap_b.unc_thread
|
|
if a is None or b is None:
|
|
return False
|
|
return (
|
|
a.diameter_fraction == b.diameter_fraction
|
|
and a.tpi == b.tpi
|
|
and a.series == b.series
|
|
)
|
|
|
|
|
|
def _clamp_range_mm(clamp: ClampConnection) -> tuple[float | None, float | None]:
|
|
if clamp.min_mm is not None or clamp.max_mm is not None:
|
|
return clamp.min_mm, clamp.max_mm
|
|
min_mm = clamp.min_in * MM_PER_INCH if clamp.min_in is not None else None
|
|
max_mm = clamp.max_in * MM_PER_INCH if clamp.max_in is not None else None
|
|
return min_mm, max_mm
|
|
|
|
|
|
def _static_in_clamp_range(static: StaticConnection, clamp: ClampConnection) -> bool:
|
|
min_mm, max_mm = _clamp_range_mm(clamp)
|
|
if min_mm is None and max_mm is None:
|
|
return False
|
|
if min_mm is not None and static.od_mm < min_mm:
|
|
return False
|
|
if max_mm is not None and static.od_mm > max_mm:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _static_clamp_match(ap_a: AttachmentPoint, ap_b: AttachmentPoint) -> bool:
|
|
if ap_a.static_connection is not None and ap_b.clamp_connection is not None:
|
|
return _static_in_clamp_range(ap_a.static_connection, ap_b.clamp_connection)
|
|
if ap_b.static_connection is not None and ap_a.clamp_connection is not None:
|
|
return _static_in_clamp_range(ap_b.static_connection, ap_a.clamp_connection)
|
|
return False
|
|
|
|
|
|
def attachment_points_match(ap_a: AttachmentPoint, ap_b: AttachmentPoint) -> bool:
|
|
"""True if these two attachment points satisfy the fit-matching algorithm."""
|
|
if ap_a.attachment_point_name_id == ap_b.attachment_point_name_id:
|
|
return False
|
|
return (
|
|
_iso_match(ap_a, ap_b)
|
|
or _unc_match(ap_a, ap_b)
|
|
or _static_clamp_match(ap_a, ap_b)
|
|
)
|
|
|
|
|
|
def parts_could_fit(part_a: AugmentedPart, part_b: AugmentedPart) -> bool:
|
|
return part_a.id != part_b.id and part_a.part_type_id != part_b.part_type_id
|
|
|
|
|
|
@dataclass
|
|
class RecomputeResult:
|
|
created: int = 0
|
|
removed: int = 0
|
|
|
|
def __iadd__(self, other: "RecomputeResult") -> "RecomputeResult":
|
|
self.created += other.created
|
|
self.removed += other.removed
|
|
return self
|
|
|
|
|
|
async def _all_parts_by_id(db: AsyncSession) -> dict[int, AugmentedPart]:
|
|
result = await db.execute(select(AugmentedPart))
|
|
return {p.id: p for p in result.scalars().all()}
|
|
|
|
|
|
async def _attachment_points_by_part(
|
|
db: AsyncSession,
|
|
) -> dict[int, list[AttachmentPoint]]:
|
|
result = await db.execute(select(AttachmentPoint))
|
|
by_part: dict[int, list[AttachmentPoint]] = defaultdict(list)
|
|
for ap in result.scalars().all():
|
|
by_part[ap.augmented_part_id].append(ap)
|
|
return by_part
|
|
|
|
|
|
def _canonical_pair(a_id: int, b_id: int) -> tuple[int, int]:
|
|
return (a_id, b_id) if a_id < b_id else (b_id, a_id)
|
|
|
|
|
|
async def _reconcile(
|
|
db: AsyncSession,
|
|
*,
|
|
scoped_part_ids: set[int],
|
|
parts_by_id: dict[int, AugmentedPart],
|
|
points_by_part: dict[int, list[AttachmentPoint]],
|
|
) -> RecomputeResult:
|
|
"""Reconcile Possible fits for the given parts against every other part."""
|
|
candidate_pairs: set[tuple[int, int]] = set()
|
|
for part_id in scoped_part_ids:
|
|
part = parts_by_id.get(part_id)
|
|
if part is None:
|
|
continue
|
|
own_points = points_by_part.get(part_id, [])
|
|
if not own_points:
|
|
continue
|
|
for other_id, other in parts_by_id.items():
|
|
if not parts_could_fit(part, other):
|
|
continue
|
|
for own_ap in own_points:
|
|
for other_ap in points_by_part.get(other_id, []):
|
|
if attachment_points_match(own_ap, other_ap):
|
|
candidate_pairs.add(
|
|
_canonical_pair(own_ap.id, other_ap.id)
|
|
)
|
|
|
|
result = RecomputeResult()
|
|
|
|
existing_fits = []
|
|
for part_id in scoped_part_ids:
|
|
existing_fits.extend(await fit_crud.list_for_part(db, part_id))
|
|
existing_by_pair = {
|
|
(f.attachment_point_a_id, f.attachment_point_b_id): f for f in existing_fits
|
|
}
|
|
|
|
for pair, existing in existing_by_pair.items():
|
|
if existing.label == FitLabel.POSSIBLE and pair not in candidate_pairs:
|
|
await fit_crud.delete(db, existing)
|
|
result.removed += 1
|
|
|
|
for ap_a_id, ap_b_id in candidate_pairs:
|
|
if (ap_a_id, ap_b_id) in existing_by_pair:
|
|
continue
|
|
ap_a = await db.get(AttachmentPoint, ap_a_id)
|
|
ap_b = await db.get(AttachmentPoint, ap_b_id)
|
|
await fit_crud.create(db, ap_a=ap_a, ap_b=ap_b, label=FitLabel.POSSIBLE)
|
|
result.created += 1
|
|
|
|
return result
|
|
|
|
|
|
async def recompute_for_part(db: AsyncSession, augmented_part_id: int) -> RecomputeResult:
|
|
"""Reconcile Possible fits involving this part against the current algorithm.
|
|
|
|
Creates Possible fits for newly-matching attachment point pairs, and removes
|
|
previously-algorithmic Possible fits that no longer match. Exact/None labels
|
|
are user-asserted and are never touched.
|
|
"""
|
|
parts_by_id = await _all_parts_by_id(db)
|
|
if augmented_part_id not in parts_by_id:
|
|
return RecomputeResult()
|
|
points_by_part = await _attachment_points_by_part(db)
|
|
return await _reconcile(
|
|
db,
|
|
scoped_part_ids={augmented_part_id},
|
|
parts_by_id=parts_by_id,
|
|
points_by_part=points_by_part,
|
|
)
|
|
|
|
|
|
async def recompute_all(db: AsyncSession) -> RecomputeResult:
|
|
parts_by_id = await _all_parts_by_id(db)
|
|
points_by_part = await _attachment_points_by_part(db)
|
|
return await _reconcile(
|
|
db,
|
|
scoped_part_ids=set(parts_by_id),
|
|
parts_by_id=parts_by_id,
|
|
points_by_part=points_by_part,
|
|
)
|