Spaces:
Running
Running
| """All active collection batches and database ingestion updates.""" | |
| from __future__ import annotations | |
| from collections import Counter | |
| from datetime import date, datetime, timedelta | |
| import html | |
| import math | |
| from typing import Callable, Iterable | |
| from zoneinfo import ZoneInfo | |
| import pandas as pd | |
| import plotly.graph_objects as go | |
| import streamlit as st | |
| from batch_updates import BatchUpdateCatalog, BatchUpdateRecord | |
| from city_labels import city_display_name | |
| from dashboard_charts import DATA_BLUE, style_figure | |
| _BEIJING = ZoneInfo('Asia/Shanghai') | |
| def ingestion_totals(catalog: BatchUpdateCatalog, today: date) -> tuple[int, int, int]: | |
| """Today and inclusive last 7/30 Beijing days, excluding future entries.""" | |
| counts = Counter() | |
| for record in catalog.records: | |
| for item in record.ingestion_days: | |
| counts[item.day] += item.videos | |
| return tuple(sum(count for day, count in counts.items() | |
| if today - timedelta(days=window-1) <= day <= today) | |
| for window in (1, 7, 30)) | |
| def ingestion_timeline(catalog: BatchUpdateCatalog, today: date, | |
| window: int | None = 30) -> list[tuple[date, int]]: | |
| """Fill Beijing calendar days through today; future rows never add activity.""" | |
| counts = Counter() | |
| for record in catalog.records: | |
| for item in record.ingestion_days: | |
| if item.day <= today: | |
| counts[item.day] += item.videos | |
| if window is None: | |
| if not counts: | |
| return [] | |
| start = min(counts) | |
| else: | |
| start = today - timedelta(days=window-1) | |
| return [(start + timedelta(days=offset), counts[start + timedelta(days=offset)]) | |
| for offset in range((today-start).days+1)] | |
| def _hour_days(catalog: BatchUpdateCatalog, today: date) -> dict[date, tuple[float, int, int]] | None: | |
| if not catalog.includes_durations: | |
| return None | |
| buckets = {} | |
| for record in catalog.records: | |
| for item in record.ingestion_days: | |
| if item.day <= today: | |
| bucket = buckets.setdefault(item.day, [[], 0, 0]) | |
| bucket[0].append(item.known_seconds) | |
| bucket[1] += item.known_videos | |
| bucket[2] += item.unknown_videos | |
| return {day: (math.fsum(values[0]), values[1], values[2]) for day, values in buckets.items()} | |
| def ingestion_hours(catalog: BatchUpdateCatalog, today: date) -> tuple[float | None, ...]: | |
| """Known active-video hours received in Beijing day windows, never frame estimates.""" | |
| days = _hour_days(catalog, today) | |
| if days is None: | |
| return None, None, None | |
| totals = [] | |
| for window in (1, 7, 30): | |
| selected = [values for day, values in days.items() | |
| if today - timedelta(days=window-1) <= day <= today] | |
| known, unknown = sum(row[1] for row in selected), sum(row[2] for row in selected) | |
| totals.append(math.fsum(row[0] for row in selected) / 3600 if known or not unknown else None) | |
| return tuple(totals) | |
| def ingestion_hour_timeline(catalog: BatchUpdateCatalog, today: date, | |
| window: int | None = 30) -> list[tuple[date, float | None]] | None: | |
| """No arrivals is measured zero; all-unknown arrivals leave an explicit gap.""" | |
| days = _hour_days(catalog, today) | |
| if days is None: | |
| return None | |
| if window is None and not days: | |
| return [] | |
| start = min(days) if window is None else today - timedelta(days=window-1) | |
| rows = [] | |
| for offset in range((today-start).days+1): | |
| day = start + timedelta(days=offset) | |
| seconds, known, unknown = days.get(day, (0, 0, 0)) | |
| rows.append((day, seconds / 3600 if known or not unknown else None)) | |
| return rows | |
| def batch_table(records: Iterable[BatchUpdateRecord], translate: Callable) -> pd.DataFrame: | |
| """Export only curated batch fields, with public timestamps in Beijing time.""" | |
| def stamp(value): | |
| return value.astimezone(_BEIJING).strftime('%Y-%m-%d %H:%M:%S') if value else translate('未知') | |
| records = sorted(records, key=lambda record: (record.last_ingested_at is None, | |
| -record.last_ingested_at.timestamp() if record.last_ingested_at else 0, record.code)) | |
| columns = tuple(map(translate, ('批次编号','城市','已知时长 (h)','首次入库','最近入库', | |
| '源数据更新','标注完成','可用标签'))) | |
| return pd.DataFrame([ | |
| (record.code, city_display_name(record.city), record.duration_hours, stamp(record.first_ingested_at), | |
| stamp(record.last_ingested_at), stamp(record.source_updated_at), | |
| stamp(record.annotation_completed_at), translate('有' if record.annotations_available else '暂无')) | |
| for record in records | |
| ], columns=columns) | |
| def render_batch_updates(load_updates: Callable, translate: Callable) -> None: | |
| def tr(zh, en): | |
| translated = translate(zh) | |
| return en if st.session_state.get('lang') == 'en' and translated == zh else translated | |
| names = {'批次编号':'Batch code','城市':'City','已知时长 (h)':'Known duration (h)', | |
| '首次入库':'First ingested','最近入库':'Last ingested','源数据更新':'Source updated', | |
| '标注完成':'Annotation completed','可用标签':'Usable labels','有':'Available', | |
| '暂无':'Unavailable','未知':'Unknown'} | |
| def table_translate(value): | |
| return tr(value, names.get(value, value)) | |
| st.markdown("<div id='nav-updates' class='dash-section'><h2 class='dash-section-title'>" | |
| + html.escape(tr('数据更新', 'Dataset updates')) + '</h2></div>', unsafe_allow_html=True) | |
| st.caption(tr( | |
| '这里展示全部活跃采集批次，独立于城市、采集日期与 ODD 筛选。入库时间是数据库接收视频的时间，' | |
| '不等于相机采集时间；新增量按当前活跃视频的首次入库日计算，不是包含删除记录的历史审计。' | |
| '时间统一为北京时间；标注完成取实际处理结束时间，源数据更新取视频修改时间。', | |
| 'All active collection batches are shown independently of city, collection-date and ODD filters. ' | |
| 'Ingestion means database receipt, not camera collection. Additions use the creation dates of currently ' | |
| 'active videos; this is not a deletion audit. Times use Beijing time. Annotation completion uses the ' | |
| 'actual processing finish time; source updates use video modification time.')) | |
| catalog, status = load_updates() | |
| if status != 'ok' or catalog is None: | |
| st.warning(tr('数据更新暂不可用，请稍后刷新。', 'Dataset updates are unavailable. Please refresh later.')) | |
| return | |
| today = datetime.now(_BEIJING).date() | |
| today_hours, seven_hours, month_hours = ingestion_hours(catalog, today) | |
| with st.container(key='batch_updates_kpis'): | |
| for column, label, english, value in zip(st.columns(4), | |
| ('活跃采集批次','今日新增 (h)','近 7 日新增 (h)','近 30 日新增 (h)'), | |
| ('Active batches','Added today (h)','Added in 7 days (h)','Added in 30 days (h)'), | |
| (len(catalog.records), today_hours, seven_hours, month_hours)): | |
| column.metric(tr(label, english), (f'{value:,}' if label == '活跃采集批次' | |
| else f'{value:,.3f}' if value is not None else '—')) | |
| if not catalog.includes_durations: | |
| st.info(tr('批次及每日入库时长尚未发布；暂不显示时长趋势，不以视频数量换算。', | |
| 'Batch and daily ingestion durations have not been published. The duration trend is unavailable.')) | |
| elif any(record.unknown_videos for record in catalog.records): | |
| st.caption(tr('含未知时长数据，仅合计实际已知时长；全未知显示空值，测得零时长保留为 0 h。', | |
| 'Some durations are unknown. Only measured duration is summed; all-unknown values are blank, measured zero remains 0 h.')) | |
| if any(sum(day.videos for day in record.ingestion_days) < record.videos for record in catalog.records): | |
| st.caption(tr('存在入库日期缺失的数据；每日趋势和近期新增指标仅统计有入库日期的已知时长。', | |
| 'Some ingestion dates are missing. Daily and recent additions include only measured duration with an ingestion date.')) | |
| if catalog.unlinked_videos: | |
| hours = catalog.unlinked_duration_hours | |
| value = f'{hours:,.3f} h' if hours is not None else tr('时长未知或尚未发布', 'duration unknown or unpublished') | |
| st.info(tr(f'另有未关联采集批次的数据（已知部分：{value}），未计入批次新增量与下表。', | |
| f'Unlinked data (known portion: {value}) is excluded from batch additions and the table.')) | |
| if not catalog.records: | |
| st.info(tr('暂无活跃采集批次。', 'No active collection batches are available.')) | |
| return | |
| ranges = (tr('近 7 日', 'Last 7 days'), tr('近 30 日', 'Last 30 days'), tr('全部', 'All time')) | |
| selected_range = st.selectbox(tr('入库时间范围', 'Ingestion time range'), ranges, index=1, | |
| key='batch_updates_range') | |
| window = {ranges[0]:7, ranges[1]:30, ranges[2]:None}[selected_range] | |
| timeline = ingestion_hour_timeline(catalog, today, window) | |
| if timeline: | |
| figure = go.Figure(go.Bar(x=[day.isoformat() for day, _ in timeline], | |
| y=[count for _, count in timeline], marker_color=DATA_BLUE, | |
| hovertemplate='%{x}: %{y:.3f} h<extra></extra>')) | |
| style_figure(figure, title=tr('每日新增时长 (h) · 当前活跃数据', 'Daily ingested duration (h) · currently active data'), | |
| height=300) | |
| figure.update_xaxes(title=tr('首次入库日（北京时间）', 'First ingestion day (Beijing)'), type='date') | |
| figure.update_yaxes(title='h', rangemode='tozero', tickformat='.3~f') | |
| st.plotly_chart(figure, width='stretch', key='batch_updates_timeline') | |
| elif timeline is not None: | |
| st.info(tr('当前批次尚无可用入库日期。', 'No ingestion dates are available for these batches.')) | |
| left, right = st.columns(2) | |
| query = left.text_input(tr('搜索批次编号或城市', 'Search batch code or city'), key='batch_updates_search').strip().casefold() | |
| cities = right.multiselect(tr('更新表城市', 'Batch table cities'), sorted({r.city for r in catalog.records}), | |
| key='batch_updates_cities', format_func=city_display_name) | |
| selected = tuple(record for record in catalog.records | |
| if (not cities or record.city in cities) | |
| and (not query or query in record.code.casefold() or query in record.city.casefold() | |
| or query in city_display_name(record.city).casefold())) | |
| st.caption(tr(f'显示 {len(selected):,} / {len(catalog.records):,} 个批次；CSV 始终包含全部活跃批次。', | |
| f'Showing {len(selected):,} / {len(catalog.records):,} batches; CSV always includes all active batches.')) | |
| st.dataframe(batch_table(selected, table_translate), hide_index=True, width='stretch') | |
| st.download_button(tr('下载全部批次 CSV', 'Download all batches CSV'), | |
| batch_table(catalog.records, table_translate).to_csv(index=False).encode('utf-8-sig'), | |
| file_name='aerial-batch-updates.csv', mime='text/csv', key='batch_updates_csv') | |