Spaces:
Sleeping
Sleeping
| """Read-only collection-point grouping with private representative GPS. | |
| Only city labels and opaque batch/location membership leave this adapter. A | |
| group is a derived 100m complete-linkage cluster, not a reviewed place identity. | |
| """ | |
| from __future__ import annotations | |
| from collections.abc import Iterable | |
| from dataclasses import dataclass | |
| import hashlib | |
| import math | |
| from uuid import UUID | |
| import psycopg | |
| from annotation_statistics import validate_public_labels | |
| from live_inventory import ( | |
| _boundary_path, _city_for_coordinates, _load_boundaries, | |
| _MISSING_COORDINATES, _OUTSIDE_BOUNDARIES, | |
| ) | |
| _MAX_DISTANCE_METERS = 100 | |
| _EARTH_RADIUS_METERS = 6_371_008.8 | |
| # Match annotation_statistics eligibility exactly. Only the representative's | |
| # own GPS is read; EXISTS neither multiplies batches nor borrows member GPS. | |
| _LOCATION_SQL = """ | |
| SELECT b.batch_id, b.created_at, v.collect_lon, v.collect_lat, b.labels | |
| FROM aerial.annotation_batches b | |
| JOIN aerial.videos v ON v.sha256 = b.representative_video_sha256 | |
| WHERE b.processing_info->>'status' = 'completed' | |
| AND b.validation_passed IS TRUE | |
| AND b.review_status IS DISTINCT FROM 'rejected' | |
| AND v.deleted_at IS NULL | |
| AND EXISTS ( | |
| SELECT 1 FROM aerial.annotation_batch_videos bv | |
| WHERE bv.batch_id = b.batch_id AND bv.video_sha256 = v.sha256 | |
| ) | |
| ORDER BY b.created_at ASC NULLS LAST, b.batch_id ASC | |
| """ | |
| class LocationGroup: | |
| code: str | |
| city: str | |
| batch_codes: tuple[str, ...] | |
| class LocationCatalog: | |
| groups: tuple[LocationGroup, ...] | |
| unresolved_batch_codes: tuple[str, ...] | |
| max_distance_meters: int = _MAX_DISTANCE_METERS | |
| def _identifier(value: object) -> UUID: | |
| return value if isinstance(value, UUID) else UUID(str(value)) | |
| def _location_code(anchor: UUID) -> str: | |
| return 'L-' + hashlib.sha256(b'nds-public-location-v1\0' + anchor.bytes).hexdigest()[:16] | |
| def _coordinates(lon: object, lat: object) -> tuple[float, float] | None: | |
| if lon is None or lat is None or isinstance(lon, bool) or isinstance(lat, bool): | |
| return None | |
| try: | |
| longitude, latitude = float(lon), float(lat) | |
| except (TypeError, ValueError, OverflowError): | |
| return None | |
| if (not math.isfinite(longitude) or not math.isfinite(latitude) | |
| or not -180 <= longitude <= 180 or not -90 <= latitude <= 90): | |
| return None | |
| return longitude, latitude | |
| def _haversine_meters(first: tuple[float, float], second: tuple[float, float]) -> float: | |
| lon1, lat1 = map(math.radians, first) | |
| lon2, lat2 = map(math.radians, second) | |
| value = (math.sin((lat2 - lat1) / 2) ** 2 | |
| + math.cos(lat1) * math.cos(lat2) * math.sin((lon2 - lon1) / 2) ** 2) | |
| value = min(1.0, max(0.0, value)) | |
| return 2 * _EARTH_RADIUS_METERS * math.atan2(math.sqrt(value), math.sqrt(1 - value)) | |
| def _cluster_city(city: str, members: list[tuple[UUID, str, tuple[float, float]]]) -> list[LocationGroup]: | |
| if len(members) == 1: | |
| assignments = (1,) | |
| else: | |
| # Public snapshot readers need only the immutable types, so the source | |
| # clustering dependency is loaded exclusively on this private path. | |
| from scipy.cluster.hierarchy import fcluster, linkage | |
| distances = [ | |
| _haversine_meters(members[first][2], members[second][2]) | |
| for first in range(len(members) - 1) | |
| for second in range(first + 1, len(members)) | |
| ] | |
| assignments = fcluster( | |
| linkage(distances, method='complete'), | |
| t=_MAX_DISTANCE_METERS, criterion='distance', | |
| ) | |
| clusters: dict[int, list[tuple[UUID, str, tuple[float, float]]]] = {} | |
| for assignment, member in zip(assignments, members): | |
| clusters.setdefault(int(assignment), []).append(member) | |
| # Members retain deterministic creation-time/UUID order, making the first | |
| # member the earliest-created public identity anchor even for equal GPS. | |
| return [ | |
| LocationGroup(_location_code(cluster[0][0]), city, tuple(sorted(member[1] for member in cluster))) | |
| for cluster in clusters.values() | |
| ] | |
| def fetch_location_catalog(dsn: str) -> LocationCatalog: | |
| """Group eligible complete-label representatives without exporting GPS. | |
| Missing, invalid, outside-boundary or ambiguous GPS remains individually | |
| unresolved. No source names, directories, images or member GPS are used. | |
| Connection/query errors propagate so publication can retry safely. | |
| """ | |
| from batch_updates import _public_code | |
| path = _boundary_path() | |
| boundaries = _load_boundaries(path, path.stat().st_mtime_ns) | |
| with psycopg.connect(dsn, options='-c default_transaction_read_only=on') as connection: | |
| with connection.cursor() as cursor: | |
| cursor.execute(_LOCATION_SQL) | |
| rows = cursor.fetchall() | |
| # Do not rely on a DB/fake cursor preserving query order. Stable input order | |
| # also resolves equal-distance complete-linkage ties reproducibly. | |
| rows = sorted(rows, key=lambda row: (row[1] is None, row[1], _identifier(row[0]).bytes)) | |
| by_city: dict[str, list[tuple[UUID, str, tuple[float, float]]]] = {} | |
| unresolved = [] | |
| for batch_id, created_at, lon, lat, labels in rows: | |
| try: | |
| validate_public_labels(labels) | |
| except ValueError: | |
| continue | |
| identifier = _identifier(batch_id) | |
| code = _public_code(identifier) | |
| coordinates = _coordinates(lon, lat) | |
| if coordinates is None: | |
| unresolved.append(code) | |
| continue | |
| city = _city_for_coordinates(*coordinates, boundaries) | |
| if city in {_MISSING_COORDINATES, _OUTSIDE_BOUNDARIES}: | |
| unresolved.append(code) | |
| continue | |
| by_city.setdefault(city, []).append((identifier, code, coordinates)) | |
| groups = [group for city, members in by_city.items() for group in _cluster_city(city, members)] | |
| return LocationCatalog(tuple(sorted(groups, key=lambda group: (group.city, group.code))), | |
| tuple(sorted(unresolved))) | |
| def align_location_catalog(catalog: LocationCatalog, batch_codes: Iterable[str]) -> LocationCatalog: | |
| """Restrict a separately queried catalog to the exact exported population. | |
| Surviving groups retain their derived identities. Exported batches absent | |
| from the source catalog become unresolved; extra source memberships and | |
| empty groups disappear. Overlap is rejected instead of silently repaired. | |
| """ | |
| seen: set[str] = set() | |
| for codes in (*[group.batch_codes for group in catalog.groups], catalog.unresolved_batch_codes): | |
| for code in codes: | |
| if code in seen: | |
| raise ValueError('Overlapping location membership') | |
| seen.add(code) | |
| population = set(batch_codes) | |
| groups = [] | |
| grouped: set[str] = set() | |
| for group in catalog.groups: | |
| members = tuple(sorted(population.intersection(group.batch_codes))) | |
| if members: | |
| groups.append(LocationGroup(group.code, group.city, members)) | |
| grouped.update(members) | |
| return LocationCatalog(tuple(sorted(groups, key=lambda group: (group.city, group.code))), | |
| tuple(sorted(population - grouped)), catalog.max_distance_meters) | |