"""One exact matched representative per automatically grouped collection point.
Safe batch membership is joined only after the dashboard's metadata and ODD
filters. All details describe the selected current frame, never merged labels.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable, Mapping
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import date
from functools import lru_cache
import hashlib
from http.client import HTTPException
import html
import math
from pathlib import Path
import re
from urllib.request import Request
import streamlit as st
from annotation_statistics import AnnotationRecord, definitions
from classic_statistics import selection_label
from city_labels import city_display_name
_PAGE_SIZE = 12
_LOCATION_CODE = re.compile(r'L-[0-9a-f]{16}\Z')
_BATCH_CODE = re.compile(r'B-[0-9a-f]{16}\Z')
_FRAME_KEY = re.compile(r'[0-9a-f]{64}\.jpg\Z')
_PAGE_KEY = 'representative_gallery_page'
_CITY_KEY = 'representative_gallery_city'
_SELECTION_KEY = '_representative_gallery_selection'
_POPULATION_KEY = '_representative_gallery_population'
_MAX_FRAME_BYTES = 4 * 1024 * 1024
_STYLE = """
"""
@dataclass(frozen=True)
class LocationExample:
"""Filtered members of one safe location, in representative priority order."""
code: str
city: str
records: tuple[AnnotationRecord, ...]
total_batches: int
@property
def record(self) -> AnnotationRecord:
return self.records[0]
@property
def matching_batches(self) -> int:
return len(self.records)
@property
def collected_on(self) -> date | None:
return self.record.collected_on
@property
def batch_code(self) -> str:
return self.record.batch_code
def location_examples(records: Iterable[AnnotationRecord], locations
) -> tuple[tuple[LocationExample, ...], int]:
"""Join exact safe batch codes after filtering; unmapped batches stay pending."""
records = tuple(records)
matched = {record.batch_code: record for record in records
if isinstance(getattr(record, 'batch_code', None), str)
and _BATCH_CODE.fullmatch(record.batch_code)}
grouped = set()
examples = []
for group in locations.groups:
if not isinstance(group.code, str) or not _LOCATION_CODE.fullmatch(group.code):
continue
members = tuple(dict.fromkeys(code for code in group.batch_codes
if isinstance(code, str) and _BATCH_CODE.fullmatch(code)))
candidates = tuple(sorted((matched[code] for code in members
if code in matched and code not in grouped
and matched[code].city == group.city), key=lambda record: (
not (isinstance(getattr(record, 'frame_key', None), str)
and _FRAME_KEY.fullmatch(record.frame_key)),
-(record.collected_on or date.min).toordinal(), record.batch_code)))
if candidates:
examples.append(LocationExample(group.code, group.city, candidates, len(members)))
grouped.update(record.batch_code for record in candidates)
return tuple(examples), sum(getattr(record, 'batch_code', None) not in grouped for record in records)
def _reset_gallery_page() -> None:
st.session_state[_PAGE_KEY] = 1
def gallery_page(records: Iterable[AnnotationRecord], page: int = 1
) -> tuple[tuple[AnnotationRecord, ...], int, int]:
"""Return a clamped 12-record page, newest collection date then safe code."""
ordered = sorted(records, key=lambda record: (
-(record.collected_on or date.min).toordinal(),
getattr(record, 'batch_code', None) or '',
))
pages = (len(ordered) + _PAGE_SIZE - 1) // _PAGE_SIZE
selected = min(max(1, page), max(1, pages))
start = (selected - 1) * _PAGE_SIZE
return tuple(ordered[start:start + _PAGE_SIZE]), selected, pages
@lru_cache(maxsize=256)
def _public_gallery_image(key: str) -> bytes:
"""Fetch immutable verified thumbnails on the server; failures stay uncached."""
from public_snapshot import FETCH_TIMEOUT_SECONDS, _open_public_url
from representative_frames import public_frame_url
request = Request(public_frame_url(key), headers={'Accept': 'image/jpeg', 'User-Agent': 'NDS-statistics/1'})
with _open_public_url(request, timeout=FETCH_TIMEOUT_SECONDS) as response:
length = response.headers.get('Content-Length')
if length is not None and not 0 <= int(length) <= _MAX_FRAME_BYTES:
raise ValueError('Representative frame transfer is too large')
content = response.read(_MAX_FRAME_BYTES + 1)
if (len(content) > _MAX_FRAME_BYTES or not content.startswith(b'\xff\xd8\xff')
or not content.endswith(b'\xff\xd9') or hashlib.sha256(content).hexdigest() + '.jpg' != key):
raise ValueError('Representative frame integrity check failed')
return content
def gallery_image(key: str | None, *, cache_root: Path | None = None) -> bytes | None:
"""Serve sanitized local or public JPEG bytes from the dashboard's domain."""
if not isinstance(key, str) or not _FRAME_KEY.fullmatch(key):
return None
from representative_frames import frame_file
cached = frame_file(key, cache_root=cache_root)
if cached is not None:
try:
return cached.read_bytes()
except OSError:
pass
try:
return _public_gallery_image(key)
except (OSError, HTTPException, ValueError, TypeError, OverflowError):
return None
def _page_images(records: Iterable[AnnotationRecord], image_loader: Callable) -> dict:
"""Load only the visible page's unique valid keys, with four requests at once."""
keys = tuple(dict.fromkeys(key for record in records
if isinstance(key := getattr(record, 'frame_key', None), str)
and _FRAME_KEY.fullmatch(key)))
if not keys:
return {}
def load(key):
try:
return image_loader(key)
except (OSError, HTTPException, ValueError, TypeError, OverflowError):
return None
with ThreadPoolExecutor(max_workers=4) as executor:
return dict(zip(keys, executor.map(load, keys)))
def render_representative_gallery(
records: Iterable[AnnotationRecord], translate: Callable[[str], str], *,
locations=None, location_status: str = 'not_published',
language: str = 'zh', field_label: Callable[[str], str] | None = None,
image_loader: Callable[[str], bytes | str | None] | None = None,
batch_hours: Mapping[str, float | None] | None = None,
core: bool | None = None,
) -> None:
"""Show one matched frame per location, twelve points within one city."""
records = tuple(records)
core = any(label.field == 'scene.type' for record in records for label in record.labels) if core is None else core
if core:
from core_annotations import definitions_v2
fields = definitions_v2()
else:
fields = definitions()
def tr(zh: str, en: str) -> str:
translated = translate(zh)
return en if language == 'en' and translated == zh else translated
def label_value(label) -> str:
if label is None:
return tr('未标注', 'Unlabeled')
if label.state != 'observed':
return translate(selection_label('@' + label.state, language))
return '、'.join(translate(value) for value in label.values)
st.markdown(_STYLE, unsafe_allow_html=True)
st.markdown(""
+ html.escape(tr('筛选结果 · 场景示例', 'Filter results · Scene examples')) + '
', unsafe_allow_html=True)
st.caption(tr(
'自动归组采集点 · 同城组内最大距离 100 米 · 每点一张匹配代表帧',
'Automatically grouped collection points · Maximum within-group distance: 100 metres · One matching frame per point'))
st.markdown(""
+ html.escape(tr('归组与时长说明', 'Grouping and duration details'))
+ '
' + html.escape(tr(
'采集点不是经人工确认的路口身份。示例随上方城市、日期、编码、时段及 ODD 筛选更新；'
'缺少定位信息或待归组的批次不列入地点示例。上方匹配时长及标签分布按关联视频实际已知时长（h）统计，不按地点去重。'
'使用图片右上角控件可放大。',
'Collection points are not reviewed intersection identities. Examples follow the city, date, codec, period '
'and ODD filters above; batches without location data or awaiting grouping are omitted. Matching hours '
'and label distributions sum actual known associated video duration before point deduplication. Use the image control to enlarge.'))
+ '
"
+ html.escape(tr(f'显示 {first:,}–{first + len(cards) - 1:,} / {len(city_examples):,} 个采集点',
f'Showing {first:,}–{first + len(cards) - 1:,} / {len(city_examples):,} collection points'))
+ '
', unsafe_allow_html=True)
st.caption(pending_summary)
load_image = image_loader or gallery_image
images = _page_images((example.record for example in cards), load_image)
# Up to four peer images share the canvas. Balance the actual page count over
# its rows (5 -> 3 + 2, 9 -> 3 + 3 + 3) rather than leave one orphan image.
# A single point stays at a readable thumbnail width, aligned to the canvas.
row_count = (len(cards) + 3) // 4
row_size, extra = divmod(len(cards), row_count)
with st.container(key='representative_gallery_grid', width=640 if len(cards) == 1 else 'stretch'):
start = 0
for row_index in range(row_count):
row = cards[start:start + row_size + (row_index < extra)]
for offset, (column, example) in enumerate(zip(st.columns(len(row)), row)):
with column, st.container(border=True, key=f'representative_example_{first + start + offset}'):
record = example.record
key = getattr(record, 'frame_key', None)
image = images.get(key) if isinstance(key, str) else None
if image is not None:
st.image(image, width='stretch')
else:
st.markdown(""
+ html.escape(tr('代表帧暂不可用', 'Representative frame unavailable'))
+ '
', unsafe_allow_html=True)
code = getattr(record, 'batch_code', None)
duration = batch_hours.get(code) if batch_hours is not None else None
valid_duration = (not isinstance(duration, bool) and isinstance(duration, (int, float))
and math.isfinite(duration) and duration >= 0)
duration_text = f'{duration:,.2f} h' if valid_duration else tr('时长未知', 'Duration unknown')
code = code if isinstance(code, str) and _BATCH_CODE.fullmatch(code) else tr('批次编号暂不可用', 'Batch code unavailable')
collected = record.collected_on.isoformat() if record.collected_on else tr('采集日期未知', 'Collection date unknown')
labels = {label.field: label for label in record.labels}
road = (label_value(labels.get('scene.type')) if core else
' · '.join(label_value(labels.get(field)) for field in ('top_road_category', 'top_road_subcategory')))
weather = label_value(labels.get('environment.light.intensity' if core else 'environment.weather'))
source = getattr(record, 'schema_version', 'aerial_annotation_v1')
if source == 'aerial_annotation_v2':
provenance = tr('来源：V2 核心标注', 'Source: V2 core annotation')
elif core:
provenance = tr('来源：旧版 V1 只读映射；不是 V2 重新标注',
'Source: read-only legacy V1 mapping; not V2 reannotation')
else:
provenance = tr('来源：旧版 V1 原始标签', 'Source: original legacy V1 labels')
st.markdown(
"',
unsafe_allow_html=True)
with st.expander(tr(f'查看全部 {len(fields)} 项标签', f'View all {len(fields)} labels')):
for definition in fields:
field = definition['code']
name = field_label(field) if field_label else translate(definition['name'])
st.markdown(f""
f"{html.escape(name)}{html.escape(label_value(labels.get(field)))}
",
unsafe_allow_html=True)
start += len(row)