Spaces:
Running
Running
| """Public statistics from a single, safely filtered aerial.videos catalog.""" | |
| from __future__ import annotations | |
| from datetime import date | |
| import html | |
| import math | |
| from typing import Callable, Mapping, Sequence | |
| import pandas as pd | |
| import plotly.graph_objects as go | |
| import streamlit as st | |
| from live_inventory import VideoCatalog, VideoRecord, filter_records, summarize_records | |
| from city_labels import city_display_name, is_city_label | |
| from dashboard_charts import DATA_BLUE, style_figure | |
| _UNKNOWN = "未知" | |
| _FILTER_KEYS = ("pg_cities", "pg_date_range", "pg_include_unknown_dates", | |
| "pg_codecs", "pg_periods") | |
| _EXPORT_COLUMNS = ( | |
| "city", "collection_month", "period", "codec", "resolution", "fps", | |
| "duration_range", "videos", "duration_hours", "size_tb", | |
| "duration_known_videos", "size_known_videos", | |
| ) | |
| def selected_records(catalog: VideoCatalog, state: Mapping) -> tuple[VideoRecord, ...]: | |
| """Use the same inclusive filters for KPIs, charts, maps and export.""" | |
| # Other pages and unavailable renders have no filter widgets. Their shared | |
| # facts must still use the last selection, while current widgets override it. | |
| state = {**state.get("_pg_filter_snapshot", {}), | |
| **{key: state[key] for key in _FILTER_KEYS if key in state}} | |
| dates = state.get("pg_date_range", ()) | |
| if isinstance(dates, date): | |
| dates = (dates, dates) | |
| start = dates[0] if dates else None | |
| end = dates[-1] if dates else None | |
| return filter_records( | |
| catalog.records, | |
| cities=state.get("pg_cities", ()), | |
| start_date=start, end_date=end, | |
| include_unknown_dates=state.get("pg_include_unknown_dates", True), | |
| codecs=state.get("pg_codecs", ()), | |
| periods=state.get("pg_periods", ()), | |
| ) | |
| def _resolution(record: VideoRecord) -> str: | |
| return f"{record.width} × {record.height}" if record.width and record.height else _UNKNOWN | |
| def _fps(record: VideoRecord) -> str: | |
| return f"{float(record.fps):g}" if record.fps is not None else _UNKNOWN | |
| def _duration_range(record: VideoRecord) -> str: | |
| value = record.duration_seconds | |
| if value is None: | |
| return _UNKNOWN | |
| if value < 60: | |
| return "< 1 min" | |
| if value < 300: | |
| return "1–5 min" | |
| if value < 900: | |
| return "5–15 min" | |
| if value < 1800: | |
| return "15–30 min" | |
| if value < 3600: | |
| return "30–60 min" | |
| return "≥ 60 min" | |
| def aggregate_rows(records: Sequence[VideoRecord]) -> pd.DataFrame: | |
| """Only coarse, aggregate metadata can be downloaded from the dashboard.""" | |
| groups: dict[tuple, list] = {} | |
| for record in records: | |
| key = (record.city, record.collected_on.strftime("%Y-%m") if record.collected_on else _UNKNOWN, | |
| record.period, record.codec or _UNKNOWN, _resolution(record), | |
| _fps(record), _duration_range(record)) | |
| totals = groups.setdefault(key, [0, 0.0, 0.0, 0, 0]) | |
| totals[0] += 1 | |
| if record.duration_seconds is not None: | |
| totals[1] += float(record.duration_seconds) / 3600 | |
| totals[3] += 1 | |
| if record.size_bytes is not None: | |
| totals[2] += float(record.size_bytes) / 1_000_000_000_000 | |
| totals[4] += 1 | |
| return pd.DataFrame( | |
| [(city_display_name(key[0]), *key[1:], totals[0], totals[1] if totals[3] else None, | |
| totals[2] if totals[4] else None, totals[3], totals[4]) | |
| for key, totals in sorted(groups.items())], | |
| columns=_EXPORT_COLUMNS, | |
| ) | |
| def duration_hours_by(records: Sequence[VideoRecord], key: Callable[[VideoRecord], str]) -> dict[str, float | None]: | |
| """Sum actual known durations per category; unknown-only is not measured zero.""" | |
| totals = {} | |
| for record in records: | |
| bucket = totals.setdefault(key(record), [0.0, False]) | |
| seconds = record.duration_seconds | |
| if seconds is not None and not isinstance(seconds, bool) and math.isfinite(float(seconds)) and seconds >= 0: | |
| bucket[0] += float(seconds) / 3600 | |
| bucket[1] = True | |
| return {value: hours if known else None for value, (hours, known) in totals.items()} | |
| def public_aggregate_rows(records: Sequence[VideoRecord]) -> pd.DataFrame: | |
| """Keep internal count fields for validation, expose hours and quality states.""" | |
| rows = aggregate_rows(records) | |
| for kind in ('duration', 'size'): | |
| known = rows[f'{kind}_known_videos'] | |
| rows[f'{kind}_status'] = ['unknown' if n == 0 else 'partial' if n < total else 'known' | |
| for n, total in zip(known, rows.videos)] | |
| return rows.drop(columns=['videos', 'duration_known_videos', 'size_known_videos']) | |
| def render_dashboard(load_catalog: Callable, clear_catalog: Callable, | |
| translate: Callable[[str], str], *, render_annotations: Callable | None = None) -> None: | |
| """Refresh the whole dashboard together, never a separate population.""" | |
| def tr(zh: str, en: str | None = None) -> str: | |
| translated = translate(zh) | |
| if st.session_state.get("lang") == "en" and translated == zh and en: | |
| return en | |
| return translated | |
| def category(value: str) -> str: | |
| if is_city_label(value): | |
| return city_display_name(value) | |
| labels = { | |
| _UNKNOWN: "Unknown", "缺少坐标": "Missing coordinates", | |
| "边界外或坐标异常": "Unassigned city", | |
| "已归属城市": "Assigned city", "早高峰 (07-09)": "Morning peak (07-09)", | |
| "日常规": "Daytime", "晚高峰 (17-19)": "Evening peak (17-19)", | |
| "夜间 (19-07)": "Night (19-07)", | |
| "未知采集时间": "Unknown collection date", "known": "Known", "partial": "Partly unknown", | |
| "unknown": "Unknown", "< 1 min": "< 0.017 h", "1–5 min": "0.017–0.083 h", | |
| "5–15 min": "0.083–0.25 h", "15–30 min": "0.25–0.5 h", "30–60 min": "0.5–1 h", "≥ 60 min": "≥ 1 h", | |
| } | |
| if value in {"< 1 min", "1–5 min", "5–15 min", "15–30 min", "30–60 min", "≥ 60 min"}: | |
| return labels[value] | |
| if value in {'known', 'partial', 'unknown'}: | |
| return tr({'known':'已知','partial':'部分未知','unknown':'未知'}[value], labels[value]) | |
| return tr(value, labels.get(value)) | |
| def heading(anchor: str, zh: str, en: str, description: str = "", english: str = "") -> None: | |
| st.markdown( | |
| f"<div id='{anchor}' class='dash-section'><h2 class='dash-section-title'>" | |
| f"{html.escape(tr(zh, en))}</h2></div>", unsafe_allow_html=True, | |
| ) | |
| if description: | |
| st.caption(tr(description, english)) | |
| def bar(counts: Mapping[str, float | None], title: str, english: str, | |
| *, order: Sequence[str] | None = None) -> None: | |
| keys = [key for key in order if key in counts] if order else sorted(counts, key=lambda key: (counts[key] is None, -(counts[key] or 0), key)) | |
| if order: | |
| keys.extend(key for key in counts if key not in keys) | |
| figure = go.Figure(go.Bar(x=[category(key) for key in keys], | |
| y=[counts[key] for key in keys], | |
| marker_color=DATA_BLUE, | |
| hovertemplate="%{x}: %{y:.3f} h<extra></extra>")) | |
| style_figure(figure, title=tr(title, english), height=320) | |
| figure.update_yaxes(title='h', rangemode="tozero", tickformat=",.2f") | |
| st.plotly_chart(figure, width="stretch") | |
| def unavailable_annotations() -> None: | |
| if render_annotations is not None: | |
| render_annotations() | |
| return | |
| st.info(tr( | |
| "当前视频表没有道路、天气、光照、动态目标、审核关键帧或经审定地点标签，相关标注分布与覆盖率暂不可统计。", | |
| "The video table does not contain road, weather, lighting, dynamic-object, reviewed-keyframe or approved-location labels. Their distributions and annotation coverage are unavailable.", | |
| )) | |
| def render_current_catalog() -> None: | |
| heading("nav-scenario", "实时视频统计", "Live video statistics") | |
| if st.button(tr("刷新数据", "Refresh data"), key="pg_refresh"): | |
| clear_catalog() | |
| catalog, status = load_catalog() | |
| if status == "not_configured": | |
| st.info(tr("尚未配置 PostgreSQL 数据源。", "PostgreSQL data source is not configured.")) | |
| return | |
| if status != "ok" or catalog is None: | |
| st.warning(tr("PostgreSQL 数据暂不可用，请稍后刷新。", "PostgreSQL data is unavailable. Please refresh later.")) | |
| return | |
| # Streamlit removes widget state when an unavailable page does not | |
| # render its controls. Keep a nonwidget snapshot so recovery preserves | |
| # the user's filters instead of silently showing the entire inventory. | |
| saved_filters = st.session_state.get("_pg_filter_snapshot", {}) | |
| for key, value in saved_filters.items(): | |
| if key not in st.session_state: | |
| st.session_state[key] = list(value) if isinstance(value, list) else value | |
| updated = str(catalog.latest_updated_at)[:19] if catalog.latest_updated_at else "—" | |
| st.caption(f"{tr('数据源', 'Source')}: aerial_db · aerial.videos | " | |
| f"{tr('库内最近更新', 'Latest database update')}: {updated} | " | |
| f"{tr('每 60 秒自动刷新', 'Refreshes every 60 seconds')}") | |
| if not catalog.records: | |
| st.info(tr("当前数据库没有有效视频记录。", "The database has no active video records.")) | |
| with st.expander(tr("筛选条件", "Filters"), expanded=True): | |
| # Retain removed selections as available options. Clearing them would | |
| # turn a deliberately narrow filter into the entire new population. | |
| city_options = sorted({record.city for record in catalog.records} | | |
| set(st.session_state.get("pg_cities", ()))) | |
| codec_options = sorted({record.codec or _UNKNOWN for record in catalog.records} | | |
| set(st.session_state.get("pg_codecs", ()))) | |
| period_options = sorted({record.period for record in catalog.records} | | |
| set(st.session_state.get("pg_periods", ()))) | |
| left, right = st.columns(2) | |
| with left: | |
| st.multiselect(tr("城市", "City"), city_options, key="pg_cities", format_func=category) | |
| dates = [record.collected_on for record in catalog.records if record.collected_on] | |
| if dates: | |
| bounds = (min(dates), max(dates)) | |
| previous_bounds = st.session_state.get("_pg_date_bounds") | |
| selected_dates = st.session_state.get("pg_date_range") | |
| if selected_dates is None or tuple(selected_dates) == previous_bounds: | |
| st.session_state["pg_date_range"] = bounds | |
| st.session_state["_pg_date_bounds"] = bounds | |
| if dates or "pg_date_range" in st.session_state: | |
| # The default 'today' reads an existing range from state; | |
| # this avoids competing widget-default and state values. | |
| st.date_input(tr("采集日期范围", "Collection date range"), | |
| key="pg_date_range") | |
| else: | |
| st.caption(tr("采集日期均未知。", "All collection dates are unknown.")) | |
| if "pg_include_unknown_dates" not in st.session_state: | |
| st.session_state["pg_include_unknown_dates"] = True | |
| st.checkbox(tr("包含未知采集日期", "Include unknown collection dates"), | |
| key="pg_include_unknown_dates") | |
| with right: | |
| st.multiselect(tr("编码格式", "Codec"), codec_options, key="pg_codecs", format_func=category) | |
| st.multiselect(tr("采集时段", "Collection period"), period_options, | |
| key="pg_periods", format_func=category) | |
| st.caption(tr("时段按北京时间分组；每个图表与导出均应用以上筛选。", | |
| "Periods use China Standard Time. These filters apply to every chart and export.")) | |
| st.session_state["_pg_filter_snapshot"] = { | |
| key: list(st.session_state[key]) if isinstance(st.session_state[key], list) | |
| else st.session_state[key] | |
| for key in _FILTER_KEYS if key in st.session_state | |
| } | |
| records = selected_records(catalog, st.session_state) | |
| snapshot = summarize_records(records, catalog.latest_updated_at) | |
| city_count = len({record.city for record in records if record.gps_status == "已归属城市"}) | |
| duration_known = sum(record.duration_seconds is not None for record in records) | |
| size_known = sum(record.size_bytes is not None for record in records) | |
| duration_display = f"{snapshot.duration_hours:,.1f}" if not records or duration_known else "—" | |
| size_display = f"{snapshot.size_tb:,.2f}" if not records or size_known else "—" | |
| gps_records = tuple(record for record in records if record.gps_status != "缺少坐标") | |
| gps_hours = sum(record.duration_seconds / 3600 for record in gps_records if record.duration_seconds is not None) | |
| gps_display = f"{gps_hours:,.2f}" if not gps_records or any(record.duration_seconds is not None for record in gps_records) else '—' | |
| metrics = ( | |
| (duration_display, tr("已知时长合计（h）", "Known duration total (h)")), | |
| (size_display, tr("已知容量合计（TB）", "Known size total (TB)")), | |
| (str(city_count), tr("覆盖城市", "Cities")), | |
| (gps_display, tr("含坐标视频时长（h）", "Geolocated video duration (h)")), | |
| ) | |
| for index, (column, (value, label)) in enumerate(zip(st.columns(4), metrics)): | |
| with column: | |
| style = ("kpi-blue", "kpi-indigo", "kpi-teal", "kpi-purple")[index % 4] | |
| st.markdown(f"<div class='kpi-card {style}'><div class='kpi-num'>{value}</div>" | |
| f"<div class='kpi-lbl'>{html.escape(label)}</div></div>", unsafe_allow_html=True) | |
| if not records: | |
| st.info(tr("当前筛选没有匹配的视频，请调整条件。", "No videos match these filters. Adjust the selections.")) | |
| heading("nav-odd", "元数据与标注状态", "Metadata and annotation status") | |
| unavailable_annotations() | |
| return | |
| heading("nav-city", "城市与采集分布", "Cities and collection distribution", | |
| "城市按采集坐标与行政区边界归属；地图点为城市行政区代表点，不表示视频采集位置。", | |
| "Cities are assigned from collection GPS and administrative boundaries. Map markers represent city administrative areas, not video collection positions.") | |
| city_counts = duration_hours_by(records, lambda record: record.city) | |
| centers = {center.city: center for center in catalog.city_centers} | |
| mapped_cities = sorted(set(city_counts) & set(centers)) | |
| left, right = st.columns(2) | |
| with left: | |
| if mapped_cities: | |
| figure = go.Figure(go.Scattermap( | |
| lon=[centers[city].longitude for city in mapped_cities], | |
| lat=[centers[city].latitude for city in mapped_cities], | |
| text=[f"{category(city)} · " + (f"{city_counts[city]:.3f} h" if city_counts[city] is not None else tr("时长未知", "Duration unknown")) for city in mapped_cities], | |
| customdata=[city_counts[city] for city in mapped_cities], | |
| mode="markers", marker=dict(size=14, color=DATA_BLUE), | |
| hovertemplate="%{text}<extra></extra>", | |
| )) | |
| style_figure(figure, height=340) | |
| figure.update_layout(map=dict(style="carto-positron", center=dict(lat=36, lon=113), zoom=3), | |
| margin=dict(l=0, r=0, t=0, b=0), height=340) | |
| st.plotly_chart(figure, width="stretch") | |
| else: | |
| st.info(tr("当前筛选没有可归属城市的记录。", "No selected videos can be assigned to a city.")) | |
| with right: | |
| bar(city_counts, "城市视频时长（h）", "Video duration by city (h)") | |
| left, right = st.columns(2) | |
| with left: | |
| months = duration_hours_by(records, lambda record: record.collected_on.strftime("%Y-%m") if record.collected_on else _UNKNOWN) | |
| bar(months, "采集月份时长（h）", "Duration by collection month (h)", order=sorted(months)) | |
| with right: | |
| bar(duration_hours_by(records, lambda record: record.period), "采集时段分布（h）", "Collection period (h)", | |
| order=("早高峰 (07-09)", "日常规", "晚高峰 (17-19)", "夜间 (19-07)", _UNKNOWN)) | |
| heading("nav-params", "视频参数分布", "Video parameters") | |
| for left_spec, right_spec in ( | |
| ((lambda record: record.codec or _UNKNOWN, "编码格式分布", "Codec distribution"), | |
| (_resolution, "分辨率分布", "Resolution distribution")), | |
| ((_fps, "帧率分布", "Frame rate distribution"), | |
| (_duration_range, "时长区间分布", "Duration distribution")), | |
| ): | |
| for column, (key, zh, en) in zip(st.columns(2), (left_spec, right_spec)): | |
| with column: | |
| order = ("< 1 min", "1–5 min", "5–15 min", "15–30 min", "30–60 min", "≥ 60 min", _UNKNOWN) if key is _duration_range else None | |
| bar(duration_hours_by(records, key), zh + "（h）", en + " (h)", order=order) | |
| heading("nav-odd", "元数据与标注状态", "Metadata and annotation status", | |
| "所有分布按已知视频时长（h）汇总；缺失字段独立显示，未知时长不推算，空值与测得零分开。", | |
| "Distributions sum known video duration in hours. Missing metadata remains separate; unknown durations are not estimated and blank is distinct from measured zero.") | |
| quality_specs = ( | |
| ("采集日期", "Collection date", lambda record: record.collected_on is not None), | |
| ("城市归属", "City attribution", lambda record: record.gps_status == "已归属城市"), | |
| ("编码", "Codec", lambda record: record.codec not in (None, "", _UNKNOWN)), | |
| ("分辨率", "Resolution", lambda record: bool(record.width and record.height)), | |
| ("帧率", "Frame rate", lambda record: record.fps is not None), | |
| ("时长", "Duration", lambda record: record.duration_seconds is not None), | |
| ("容量", "Size", lambda record: record.size_bytes is not None), | |
| ("入库日期", "Ingestion date", lambda record: record.added_on is not None), | |
| ) | |
| labels, present, absent = [], [], [] | |
| for zh, en, predicate in quality_specs: | |
| labels.append(tr(zh, en)) | |
| hours = duration_hours_by(records, lambda record: 'known' if predicate(record) else 'unknown') | |
| present.append(hours.get('known', 0.0)) | |
| absent.append(hours.get('unknown', 0.0)) | |
| figure = go.Figure(( | |
| go.Bar(name=tr("已知 / 有效", "Known / valid"), x=labels, y=present, marker_color=DATA_BLUE), | |
| go.Bar(name=tr("未知 / 无效", "Unknown / invalid"), x=labels, | |
| y=absent, marker_color="#dce5ec"), | |
| )) | |
| style_figure(figure, height=320) | |
| figure.update_layout(barmode="stack", margin=dict(t=50)) | |
| figure.update_yaxes(title='h', rangemode="tozero", tickformat=",.2f") | |
| figure.update_traces(hovertemplate='%{x}: %{y:.3f} h<extra>%{fullData.name}</extra>') | |
| st.plotly_chart(figure, width="stretch") | |
| left, right = st.columns(2) | |
| with left: | |
| bar(duration_hours_by(records, lambda record: record.gps_status), "坐标归属时长（h）", "GPS attribution duration (h)") | |
| with right: | |
| additions = duration_hours_by(records, lambda record: record.added_on.strftime("%Y-%m") if record.added_on else _UNKNOWN) | |
| bar(additions, "入库月份时长（h）", "Duration by ingestion month (h)", order=sorted(additions)) | |
| unavailable_annotations() | |
| st.subheader(tr("当前筛选汇总", "Filtered aggregate data")) | |
| st.caption(tr("时长单位 h；空值表示该组时长未知，部分未知状态表示仅合计已知部分；测得零保留为零。", | |
| "Duration is in h. Blank means unknown; partial means the known portion only. A measured zero remains zero.")) | |
| exported = public_aggregate_rows(records) | |
| exported['duration_range'] = exported.duration_range.map(category) | |
| exported['duration_status'] = exported.duration_status.map(category) | |
| exported['size_status'] = exported.size_status.map(category) | |
| st.dataframe(exported, hide_index=True, width="stretch", column_config={ | |
| "city": tr("城市", "City"), "collection_month": tr("采集月份", "Collection month"), | |
| "period": tr("时段", "Period"), "codec": tr("编码", "Codec"), | |
| "resolution": tr("分辨率", "Resolution"), "fps": tr("帧率", "FPS"), | |
| "duration_range": tr("时长区间", "Duration range"), | |
| "duration_hours": tr("时长（h）", "Duration (h)"), "size_tb": tr("容量（TB）", "Size (TB)"), | |
| "duration_status": tr("时长完整度", "Duration completeness"), | |
| "size_status": tr("容量完整度", "Size completeness"), | |
| }) | |
| st.download_button(tr("下载汇总 CSV", "Download aggregate CSV"), | |
| data=exported.to_csv(index=False).encode("utf-8-sig"), | |
| file_name="aerial-videos-statistics.csv", mime="text/csv", key="pg_download") | |
| render_current_catalog() | |