"""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""
f"{html.escape(tr(zh, en))}
", 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"))
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.",
))
@st.fragment(run_every="60s")
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"{value}
"
f"
{html.escape(label)}