Spaces:
Running
Running
File size: 8,367 Bytes
c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 c22731c a2bad67 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | """Public-safe batch ingestion facts from active PostgreSQL membership only.
Video creation is database ingestion, not camera collection. These facts describe
currently active rows and cannot reconstruct a historical deletion audit.
"""
from __future__ import annotations
from collections import Counter
from dataclasses import dataclass
from datetime import date, datetime, timezone
import hashlib
import math
import re
from uuid import UUID
from zoneinfo import ZoneInfo
import psycopg
from annotation_statistics import validate_public_labels
from live_inventory import (
_boundary_path, _city_for_coordinates, _load_boundaries,
_MISSING_COORDINATES, _OUTSIDE_BOUNDARIES,
_valid_float,
)
_BEIJING = ZoneInfo('Asia/Shanghai')
_BATCH_SQL = """
SELECT b.batch_id, array_agg(v.collect_lon), array_agg(v.collect_lat),
array_agg(v.created_at), array_agg(v.updated_at),
bool_or(v.sha256 = b.representative_video_sha256) AS has_representative,
b.processing_info, b.validation_passed, b.review_status, b.labels,
array_agg(v.duration_seconds)
FROM aerial.videos v
LEFT JOIN aerial.annotation_batch_videos bv ON bv.video_sha256 = v.sha256
LEFT JOIN aerial.annotation_batches b ON b.batch_id = bv.batch_id
WHERE v.deleted_at IS NULL
GROUP BY b.batch_id
"""
@dataclass(frozen=True)
class BatchIngestionDay:
day: date
videos: int
known_seconds: float | None = None
known_videos: int | None = None
unknown_videos: int | None = None
@property
def duration_hours(self) -> float | None:
return self.known_seconds / 3600 if self.known_seconds is not None and self.known_videos else None
@dataclass(frozen=True)
class BatchUpdateRecord:
code: str
city: str
videos: int
first_ingested_at: datetime | None
last_ingested_at: datetime | None
source_updated_at: datetime | None
annotation_completed_at: datetime | None
annotations_available: bool
ingestion_days: tuple[BatchIngestionDay, ...]
known_seconds: float | None = None
known_videos: int | None = None
unknown_videos: int | None = None
@property
def duration_hours(self) -> float | None:
return self.known_seconds / 3600 if self.known_seconds is not None and self.known_videos else None
@dataclass(frozen=True)
class BatchUpdateCatalog:
records: tuple[BatchUpdateRecord, ...]
unlinked_videos: int = 0
unlinked_known_seconds: float | None = None
unlinked_known_videos: int | None = None
unlinked_unknown_videos: int | None = None
@property
def unlinked_duration_hours(self) -> float | None:
return (self.unlinked_known_seconds / 3600
if self.unlinked_known_seconds is not None and self.unlinked_known_videos else None)
@property
def includes_durations(self) -> bool:
"""An empty measured catalog is available; legacy count-only catalogs are not."""
return (self.unlinked_known_seconds is not None
and self.unlinked_known_videos is not None
and self.unlinked_unknown_videos is not None
and all(record.known_seconds is not None and record.known_videos is not None
and record.unknown_videos is not None
and all(day.known_seconds is not None and day.known_videos is not None
and day.unknown_videos is not None for day in record.ingestion_days)
for record in self.records))
def _public_code(batch_id: UUID) -> str:
identifier = batch_id if isinstance(batch_id, UUID) else UUID(str(batch_id))
return 'B-' + hashlib.sha256(b'nds-public-batch-v1\0' + identifier.bytes).hexdigest()[:16]
def _finished_at(value: object) -> datetime | None:
if not isinstance(value, str) or not re.fullmatch(
r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2})', value
):
return None
try:
return datetime.fromisoformat(value).astimezone(timezone.utc)
except (ValueError, OverflowError):
return None
def fetch_batch_updates(dsn: str) -> BatchUpdateCatalog:
"""Read every active membership once; private identifiers stay local.
A batch exists here if it has at least one active video. The source
membership constraint assigns each video to one batch. GPS votes prefer
known administrative cities, breaking ties lexically, then use explicit
fallbacks. Member arrays are aggregated before transfer so labels travel
only once per batch.
"""
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(_BATCH_SQL)
rows = cursor.fetchall()
groups = {}
unlinked = 0
unlinked_seconds, unlinked_known, unlinked_unknown = [], 0, 0
coordinate_cities = {}
for batch_id, lons, lats, created_times, updated_times, representative, processing, validated, review, labels, durations in rows:
if batch_id is None:
unlinked += len(created_times)
for seconds in durations:
duration = _valid_float(seconds)
if duration is None:
unlinked_unknown += 1
else:
unlinked_known += 1
unlinked_seconds.append(float(duration))
continue
group = dict(cities=Counter(), created=[], updated=[], days={}, seconds=[], known=0, unknown=0,
videos=len(created_times), representative=bool(representative), processing=processing,
validated=validated, review=review, labels=labels)
groups[batch_id] = group
for lon, lat, created, updated, seconds in zip(lons, lats, created_times, updated_times, durations):
duration = _valid_float(seconds)
if duration is None:
group['unknown'] += 1
else:
group['known'] += 1
group['seconds'].append(float(duration))
if (lon, lat) not in coordinate_cities:
coordinate_cities[(lon, lat)] = _city_for_coordinates(lon, lat, boundaries)
group['cities'][coordinate_cities[(lon, lat)]] += 1
if created is not None:
group['created'].append(created)
day = group['days'].setdefault(created.astimezone(_BEIJING).date(), [0, [], 0, 0])
day[0] += 1
if duration is None:
day[3] += 1
else:
day[1].append(float(duration))
day[2] += 1
if updated is not None:
group['updated'].append(updated)
records = []
for batch_id, group in groups.items():
cities = group['cities']
known = {city: count for city, count in cities.items()
if city not in {_MISSING_COORDINATES, _OUTSIDE_BOUNDARIES}}
city = min(known or cities, key=lambda name: (-(known or cities)[name], name))
processing = group['processing'] if isinstance(group['processing'], dict) else {}
completed = processing.get('status') == 'completed'
available = bool(completed and group['validated'] is True and group['review'] != 'rejected'
and group['representative'])
if available:
try:
validate_public_labels(group['labels'])
except ValueError:
available = False
records.append(BatchUpdateRecord(
_public_code(batch_id), city, group['videos'],
min(group['created'], default=None), max(group['created'], default=None),
max(group['updated'], default=None),
_finished_at(processing.get('finished_at')) if completed else None, available,
tuple(BatchIngestionDay(day, values[0], math.fsum(values[1]), values[2], values[3])
for day, values in sorted(group['days'].items())),
math.fsum(group['seconds']), group['known'], group['unknown'],
))
return BatchUpdateCatalog(tuple(sorted(records, key=lambda record: record.code)), unlinked,
math.fsum(unlinked_seconds), unlinked_known, unlinked_unknown)
|