823 lines
31 KiB
Python
Executable File
823 lines
31 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import argparse
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Any, Callable, Optional
|
|
|
|
try:
|
|
from .get_clip_by_eventid import get_associated_clip_ids
|
|
from .pdcl_clip_export_utils import (
|
|
build_clip_tasks_from_clip_ids,
|
|
run_clip_tasks_inference_exported,
|
|
)
|
|
except ImportError:
|
|
from get_clip_by_eventid import get_associated_clip_ids
|
|
from pdcl_clip_export_utils import (
|
|
build_clip_tasks_from_clip_ids,
|
|
run_clip_tasks_inference_exported,
|
|
)
|
|
|
|
DEFAULT_EVENT_CACHE_FILE = Path(__file__).resolve().parents[1] / ".cache" / "event_clip_cache.json"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResolvedEventRecord:
|
|
scene: str
|
|
record_index: int
|
|
event_id: str
|
|
event_id_field_used: str
|
|
source_record: dict[str, Any]
|
|
clip_ids: list[str]
|
|
clip_source: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventResolutionStats:
|
|
total_events: int
|
|
cache_hits: int
|
|
cache_misses: int
|
|
request_workers: int
|
|
cache_file: str
|
|
direct_clip_records: int = 0
|
|
event_lookup_records: int = 0
|
|
|
|
|
|
def _dedupe_preserve_order(values: list[str]) -> list[str]:
|
|
ordered: list[str] = []
|
|
seen: set[str] = set()
|
|
for value in values:
|
|
token = str(value).strip()
|
|
if not token or token in seen:
|
|
continue
|
|
seen.add(token)
|
|
ordered.append(token)
|
|
return ordered
|
|
|
|
|
|
def _sanitize_identifier_for_path(identifier: str, prefix: str = "event_id") -> str:
|
|
token = re.sub(r'[\\/:*?"<>|\s]+', "_", str(identifier or "").strip())
|
|
token = token.strip("._")
|
|
token = token or "unknown_id"
|
|
normalized_prefix = re.sub(r"[^0-9A-Za-z]+", "_", str(prefix or "").strip())
|
|
normalized_prefix = normalized_prefix.strip("._") or "id"
|
|
return f"{normalized_prefix}_{token}"
|
|
|
|
|
|
def _build_resolution_stats_payload(resolution_stats: "EventResolutionStats") -> dict[str, Any]:
|
|
return {
|
|
"total_events": resolution_stats.total_events,
|
|
"cache_hits": resolution_stats.cache_hits,
|
|
"cache_misses": resolution_stats.cache_misses,
|
|
"request_workers": resolution_stats.request_workers,
|
|
"cache_file": resolution_stats.cache_file,
|
|
"direct_clip_records": resolution_stats.direct_clip_records,
|
|
"event_lookup_records": resolution_stats.event_lookup_records,
|
|
}
|
|
|
|
|
|
def _record_key(record: "ResolvedEventRecord") -> tuple[str, int, str]:
|
|
return (record.scene, int(record.record_index), str(record.event_id))
|
|
|
|
|
|
def _normalize_direct_clip_ids(raw_value: Any) -> list[str]:
|
|
if isinstance(raw_value, list):
|
|
return _dedupe_preserve_order([str(item).strip() for item in raw_value if str(item).strip()])
|
|
|
|
if isinstance(raw_value, str):
|
|
text = raw_value.strip()
|
|
if not text:
|
|
return []
|
|
if text.startswith("["):
|
|
try:
|
|
parsed = json.loads(text)
|
|
except Exception:
|
|
parsed = None
|
|
if isinstance(parsed, list):
|
|
return _dedupe_preserve_order([str(item).strip() for item in parsed if str(item).strip()])
|
|
return _dedupe_preserve_order([token for token in re.split(r"[\s,]+", text) if token])
|
|
|
|
if raw_value is None:
|
|
return []
|
|
|
|
token = str(raw_value).strip()
|
|
if not token:
|
|
return []
|
|
return [token]
|
|
|
|
|
|
def _extract_direct_clip_ids(record: dict[str, Any], clip_ids_field: str) -> tuple[bool, list[str]]:
|
|
field_name = str(clip_ids_field or "").strip()
|
|
if not field_name or field_name not in record:
|
|
return False, []
|
|
return True, _normalize_direct_clip_ids(record.get(field_name))
|
|
|
|
|
|
def _resolve_record_identifier(
|
|
record: dict[str, Any],
|
|
preferred_field: str,
|
|
*,
|
|
allow_direct_clip_fallback: bool = False,
|
|
) -> tuple[str, str]:
|
|
candidate_fields: list[str] = []
|
|
preferred_token = str(preferred_field or "").strip()
|
|
if preferred_token:
|
|
candidate_fields.append(preferred_token)
|
|
|
|
if allow_direct_clip_fallback:
|
|
for field_name in ("rawid", "event_id", "data_path"):
|
|
if field_name not in candidate_fields:
|
|
candidate_fields.append(field_name)
|
|
|
|
for field_name in candidate_fields:
|
|
identifier = str(record.get(field_name, "")).strip()
|
|
if identifier:
|
|
return identifier, field_name
|
|
|
|
return "", preferred_token
|
|
|
|
|
|
def _extract_condition_values(source_record: dict[str, Any], condition_fields: list[str]) -> dict[str, str]:
|
|
return {
|
|
str(field): str(source_record.get(field, "")).strip()
|
|
for field in condition_fields
|
|
}
|
|
|
|
|
|
def _select_event_records_by_condition(
|
|
records: list["ResolvedEventRecord"],
|
|
*,
|
|
condition_fields: list[str],
|
|
max_records_per_condition: int,
|
|
selection_strategy: str,
|
|
) -> tuple[list["ResolvedEventRecord"], dict[str, Any]]:
|
|
normalized_fields = [str(field).strip() for field in condition_fields if str(field).strip()]
|
|
limit = max(0, int(max_records_per_condition))
|
|
selection_enabled = bool(normalized_fields and limit > 0)
|
|
summary: dict[str, Any] = {
|
|
"enabled": selection_enabled,
|
|
"condition_fields": normalized_fields,
|
|
"max_records_per_condition": limit,
|
|
"selection_strategy": selection_strategy,
|
|
"records_before_selection": len(records),
|
|
"records_after_selection": len(records),
|
|
"group_count": 0,
|
|
"groups": [],
|
|
}
|
|
if not selection_enabled:
|
|
return list(records), summary
|
|
|
|
if selection_strategy != "first":
|
|
raise ValueError(f"Unsupported condition selection strategy: {selection_strategy!r}")
|
|
|
|
selected_records: list[ResolvedEventRecord] = []
|
|
selected_counts: dict[tuple[str, tuple[str, ...]], int] = {}
|
|
group_summaries: dict[tuple[str, tuple[str, ...]], dict[str, Any]] = {}
|
|
|
|
for record in records:
|
|
condition_values = _extract_condition_values(record.source_record, normalized_fields)
|
|
condition_tuple = tuple(condition_values[field] for field in normalized_fields)
|
|
group_key = (record.scene, condition_tuple)
|
|
group_summary = group_summaries.setdefault(
|
|
group_key,
|
|
{
|
|
"scene": record.scene,
|
|
"condition_values": condition_values,
|
|
"record_count": 0,
|
|
"selected_count": 0,
|
|
"skipped_count": 0,
|
|
"selected_record_ids": [],
|
|
"selected_record_indices": [],
|
|
"skipped_record_ids": [],
|
|
"skipped_record_indices": [],
|
|
},
|
|
)
|
|
group_summary["record_count"] += 1
|
|
current_selected_count = selected_counts.get(group_key, 0)
|
|
|
|
if current_selected_count < limit:
|
|
selected_records.append(record)
|
|
selected_counts[group_key] = current_selected_count + 1
|
|
group_summary["selected_count"] += 1
|
|
group_summary["selected_record_ids"].append(record.event_id)
|
|
group_summary["selected_record_indices"].append(record.record_index)
|
|
continue
|
|
|
|
group_summary["skipped_count"] += 1
|
|
group_summary["skipped_record_ids"].append(record.event_id)
|
|
group_summary["skipped_record_indices"].append(record.record_index)
|
|
|
|
summary["records_after_selection"] = len(selected_records)
|
|
summary["group_count"] = len(group_summaries)
|
|
summary["groups"] = list(group_summaries.values())
|
|
return selected_records, summary
|
|
|
|
|
|
def _filter_selection_summary_for_scene(selection_summary: dict[str, Any], scene: str) -> dict[str, Any]:
|
|
if not selection_summary:
|
|
return {}
|
|
|
|
scene_groups = [
|
|
dict(group)
|
|
for group in selection_summary.get("groups", [])
|
|
if str(group.get("scene", "")) == str(scene)
|
|
]
|
|
filtered = dict(selection_summary)
|
|
filtered["groups"] = scene_groups
|
|
filtered["group_count"] = len(scene_groups)
|
|
if scene_groups:
|
|
filtered["records_before_selection"] = sum(int(group.get("record_count", 0)) for group in scene_groups)
|
|
filtered["records_after_selection"] = sum(int(group.get("selected_count", 0)) for group in scene_groups)
|
|
else:
|
|
filtered["records_before_selection"] = 0
|
|
filtered["records_after_selection"] = 0
|
|
return filtered
|
|
|
|
|
|
def _format_condition_summary(source_record: dict[str, Any], condition_fields: list[str]) -> str:
|
|
normalized_fields = [str(field).strip() for field in condition_fields if str(field).strip()]
|
|
if not normalized_fields:
|
|
return ""
|
|
values = _extract_condition_values(source_record, normalized_fields)
|
|
return ", ".join(f"{field}={values.get(field, '')}" for field in normalized_fields)
|
|
|
|
|
|
def _print_event_json_summary(
|
|
*,
|
|
args: argparse.Namespace,
|
|
manifest_path: Path,
|
|
resolved_records: list["ResolvedEventRecord"],
|
|
selected_records: list["ResolvedEventRecord"],
|
|
selection_summary: dict[str, Any],
|
|
scene_results: dict[str, dict[str, Any]],
|
|
) -> None:
|
|
print(f"Event JSON file: {args.event_json_file}")
|
|
if args.scene:
|
|
print(f"Scene filter: {args.scene}")
|
|
print(
|
|
"Resolved records: "
|
|
f"{len(resolved_records)}, selected records: {len(selected_records)}"
|
|
)
|
|
print(f"Manifest: {manifest_path}")
|
|
|
|
if not selected_records:
|
|
print("No records selected.")
|
|
return
|
|
|
|
condition_fields = [str(field).strip() for field in getattr(args, "condition_fields", []) if str(field).strip()]
|
|
if args.selection_only:
|
|
print("Selection-only mode: no clip export or inference was run.")
|
|
elif selection_summary.get("enabled"):
|
|
print(
|
|
"Condition selection: "
|
|
f"{selection_summary.get('group_count', 0)} groups, "
|
|
f"strategy={selection_summary.get('selection_strategy', '')}, "
|
|
f"max_records_per_condition={selection_summary.get('max_records_per_condition', 0)}"
|
|
)
|
|
|
|
for scene in sorted(scene_results):
|
|
scene_result = scene_results[scene]
|
|
print(
|
|
f"Scene {scene}: selected_record_count="
|
|
f"{scene_result.get('selected_record_count', len([record for record in selected_records if record.scene == scene]))}"
|
|
)
|
|
scene_manifest_path = scene_result.get("scene_manifest_path")
|
|
if scene_manifest_path:
|
|
print(f" Scene manifest: {scene_manifest_path}")
|
|
if scene_result.get("scene_output_dir"):
|
|
print(f" Output dir: {scene_result['scene_output_dir']}")
|
|
|
|
scene_records = [record for record in selected_records if record.scene == scene]
|
|
for record in scene_records:
|
|
condition_summary = _format_condition_summary(record.source_record, condition_fields)
|
|
summary_parts = []
|
|
if condition_summary:
|
|
summary_parts.append(condition_summary)
|
|
summary_parts.append(f"record_id={record.event_id}")
|
|
summary_parts.append(f"clips={len(record.clip_ids)}")
|
|
print(f" - {'; '.join(summary_parts)}")
|
|
|
|
|
|
def _build_scene_manifest_payload(
|
|
*,
|
|
args: argparse.Namespace,
|
|
scene: str,
|
|
scene_records: list["ResolvedEventRecord"],
|
|
resolution_stats: "EventResolutionStats",
|
|
selection_summary: Optional[dict[str, Any]] = None,
|
|
) -> dict[str, Any]:
|
|
payload = {
|
|
"event_json_file": args.event_json_file,
|
|
"scene": scene,
|
|
"event_id_field": args.event_id_field,
|
|
"event_clip_ids_field": args.event_clip_ids_field,
|
|
"resolution_stats": _build_resolution_stats_payload(resolution_stats),
|
|
"event_record_count": len(scene_records),
|
|
"records": [
|
|
{
|
|
"record_index": record.record_index,
|
|
"event_id": record.event_id,
|
|
"event_id_field_used": record.event_id_field_used,
|
|
"clip_ids": record.clip_ids,
|
|
"clip_source": record.clip_source,
|
|
"source_record": record.source_record,
|
|
}
|
|
for record in scene_records
|
|
],
|
|
}
|
|
if selection_summary is not None:
|
|
payload["selection"] = selection_summary
|
|
return payload
|
|
|
|
|
|
def _build_event_manifest_payload(
|
|
*,
|
|
args: argparse.Namespace,
|
|
scene: str,
|
|
event_id: str,
|
|
event_records: list["ResolvedEventRecord"],
|
|
clip_ids: list[str],
|
|
resolution_stats: "EventResolutionStats",
|
|
) -> dict[str, Any]:
|
|
condition_fields = [str(field).strip() for field in getattr(args, "condition_fields", []) if str(field).strip()]
|
|
payload = {
|
|
"event_json_file": args.event_json_file,
|
|
"scene": scene,
|
|
"event_id_field": args.event_id_field,
|
|
"event_clip_ids_field": args.event_clip_ids_field,
|
|
"event_id": event_id,
|
|
"resolution_stats": _build_resolution_stats_payload(resolution_stats),
|
|
"event_record_count": len(event_records),
|
|
"clip_ids": clip_ids,
|
|
"clip_count": len(clip_ids),
|
|
"records": [
|
|
{
|
|
"record_index": record.record_index,
|
|
"event_id": record.event_id,
|
|
"event_id_field_used": record.event_id_field_used,
|
|
"clip_ids": record.clip_ids,
|
|
"clip_source": record.clip_source,
|
|
"source_record": record.source_record,
|
|
}
|
|
for record in event_records
|
|
],
|
|
}
|
|
if event_records:
|
|
payload["event_id_field_used"] = event_records[0].event_id_field_used
|
|
payload["clip_source"] = sorted({record.clip_source for record in event_records})
|
|
if condition_fields and event_records:
|
|
payload["condition_values"] = _extract_condition_values(event_records[0].source_record, condition_fields)
|
|
return payload
|
|
|
|
|
|
def load_event_scene_json(json_file: str) -> dict[str, list[dict[str, Any]]]:
|
|
path = Path(json_file)
|
|
with path.open("r", encoding="utf-8") as file:
|
|
payload = json.load(file)
|
|
if not isinstance(payload, dict):
|
|
raise ValueError(f"Expected top-level dict in {path}, got {type(payload).__name__}")
|
|
|
|
normalized: dict[str, list[dict[str, Any]]] = {}
|
|
for scene, records in payload.items():
|
|
if not isinstance(records, list):
|
|
raise ValueError(f"Scene {scene!r} should map to a list, got {type(records).__name__}")
|
|
normalized[str(scene)] = [dict(item) if isinstance(item, dict) else {"value": item} for item in records]
|
|
return normalized
|
|
|
|
|
|
def load_event_clip_cache(cache_file: str | Path) -> dict[str, list[str]]:
|
|
cache_path = Path(cache_file)
|
|
if not cache_path.exists():
|
|
return {}
|
|
|
|
try:
|
|
payload = json.loads(cache_path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {}
|
|
|
|
if not isinstance(payload, dict):
|
|
return {}
|
|
|
|
cache: dict[str, list[str]] = {}
|
|
for event_id, clip_ids in payload.items():
|
|
if isinstance(clip_ids, list):
|
|
cache[str(event_id)] = [str(item).strip() for item in clip_ids if str(item).strip()]
|
|
return cache
|
|
|
|
|
|
def save_event_clip_cache(cache_file: str | Path, cache_payload: dict[str, list[str]]) -> Path:
|
|
cache_path = Path(cache_file)
|
|
cache_path.parent.mkdir(parents=True, exist_ok=True)
|
|
normalized = {
|
|
str(event_id): [str(item).strip() for item in clip_ids if str(item).strip()]
|
|
for event_id, clip_ids in cache_payload.items()
|
|
}
|
|
with cache_path.open("w", encoding="utf-8") as file:
|
|
json.dump(normalized, file, indent=2, ensure_ascii=False)
|
|
return cache_path
|
|
|
|
|
|
def resolve_event_clip_ids(
|
|
event_ids: list[str],
|
|
*,
|
|
timeout: float = 60.0,
|
|
cache_file: str | Path = DEFAULT_EVENT_CACHE_FILE,
|
|
workers: int = 4,
|
|
max_retries: int = 3,
|
|
retry_backoff_sec: float = 2.0,
|
|
) -> tuple[dict[str, list[str]], EventResolutionStats]:
|
|
ordered_event_ids: list[str] = []
|
|
seen: set[str] = set()
|
|
for event_id in event_ids:
|
|
normalized = str(event_id).strip()
|
|
if not normalized or normalized in seen:
|
|
continue
|
|
seen.add(normalized)
|
|
ordered_event_ids.append(normalized)
|
|
|
|
cache_payload = load_event_clip_cache(cache_file)
|
|
resolved: dict[str, list[str]] = {}
|
|
unresolved_event_ids: list[str] = []
|
|
cache_hits = 0
|
|
cache_misses = 0
|
|
|
|
for event_id in ordered_event_ids:
|
|
if event_id in cache_payload:
|
|
resolved[event_id] = list(cache_payload[event_id])
|
|
cache_hits += 1
|
|
else:
|
|
unresolved_event_ids.append(event_id)
|
|
cache_misses += 1
|
|
|
|
if unresolved_event_ids:
|
|
max_workers = max(1, min(int(workers), len(unresolved_event_ids)))
|
|
if max_workers == 1:
|
|
for event_id in unresolved_event_ids:
|
|
clip_ids = get_associated_clip_ids(
|
|
event_id,
|
|
timeout=timeout,
|
|
max_retries=max_retries,
|
|
retry_backoff_sec=retry_backoff_sec,
|
|
)
|
|
resolved[event_id] = clip_ids
|
|
cache_payload[event_id] = clip_ids
|
|
else:
|
|
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
future_to_event_id = {
|
|
executor.submit(
|
|
get_associated_clip_ids,
|
|
event_id,
|
|
timeout,
|
|
max_retries,
|
|
retry_backoff_sec,
|
|
): event_id
|
|
for event_id in unresolved_event_ids
|
|
}
|
|
for future in as_completed(future_to_event_id):
|
|
event_id = future_to_event_id[future]
|
|
clip_ids = future.result()
|
|
resolved[event_id] = clip_ids
|
|
cache_payload[event_id] = clip_ids
|
|
save_event_clip_cache(cache_file, cache_payload)
|
|
worker_count = max(1, min(int(workers), len(unresolved_event_ids)))
|
|
else:
|
|
worker_count = 0
|
|
|
|
stats = EventResolutionStats(
|
|
total_events=len(ordered_event_ids),
|
|
cache_hits=cache_hits,
|
|
cache_misses=cache_misses,
|
|
request_workers=worker_count,
|
|
cache_file=str(Path(cache_file).resolve()),
|
|
)
|
|
return resolved, stats
|
|
|
|
|
|
def resolve_event_records(
|
|
json_file: str,
|
|
scene_filter: Optional[str] = None,
|
|
event_id_field: str = "data_path",
|
|
clip_ids_field: str = "clips",
|
|
max_events: int = 0,
|
|
timeout: float = 60.0,
|
|
cache_file: str | Path = DEFAULT_EVENT_CACHE_FILE,
|
|
workers: int = 4,
|
|
max_retries: int = 3,
|
|
retry_backoff_sec: float = 2.0,
|
|
) -> tuple[list[ResolvedEventRecord], EventResolutionStats]:
|
|
scene_payload = load_event_scene_json(json_file)
|
|
scene_names = [scene_filter] if scene_filter else list(scene_payload)
|
|
pending_records: list[tuple[str, int, dict[str, Any], str, str, list[str], str]] = []
|
|
lookup_event_ids: list[str] = []
|
|
processed = 0
|
|
direct_clip_records = 0
|
|
event_lookup_records = 0
|
|
|
|
for scene in scene_names:
|
|
records = scene_payload.get(scene)
|
|
if records is None:
|
|
raise ValueError(f"Scene {scene!r} not found in {json_file}")
|
|
|
|
for index, record in enumerate(records):
|
|
has_direct_clip_ids, direct_clip_ids = _extract_direct_clip_ids(record, clip_ids_field)
|
|
event_id, resolved_id_field = _resolve_record_identifier(
|
|
record,
|
|
event_id_field,
|
|
allow_direct_clip_fallback=has_direct_clip_ids,
|
|
)
|
|
if not event_id:
|
|
continue
|
|
|
|
clip_source = "direct_clip_ids_field" if has_direct_clip_ids else "event_lookup"
|
|
if has_direct_clip_ids:
|
|
direct_clip_records += 1
|
|
else:
|
|
event_lookup_records += 1
|
|
lookup_event_ids.append(event_id)
|
|
|
|
pending_records.append(
|
|
(scene, index, record, event_id, resolved_id_field, direct_clip_ids, clip_source)
|
|
)
|
|
processed += 1
|
|
if max_events > 0 and processed >= max_events:
|
|
break
|
|
if max_events > 0 and processed >= max_events:
|
|
break
|
|
|
|
resolved_clip_map: dict[str, list[str]] = {}
|
|
lookup_stats = EventResolutionStats(
|
|
total_events=0,
|
|
cache_hits=0,
|
|
cache_misses=0,
|
|
request_workers=0,
|
|
cache_file=str(Path(cache_file).resolve()),
|
|
)
|
|
if lookup_event_ids:
|
|
resolved_clip_map, lookup_stats = resolve_event_clip_ids(
|
|
lookup_event_ids,
|
|
timeout=timeout,
|
|
cache_file=cache_file,
|
|
workers=workers,
|
|
max_retries=max_retries,
|
|
retry_backoff_sec=retry_backoff_sec,
|
|
)
|
|
|
|
resolved = [
|
|
ResolvedEventRecord(
|
|
scene=scene,
|
|
record_index=index,
|
|
event_id=event_id,
|
|
event_id_field_used=resolved_id_field,
|
|
source_record=record,
|
|
clip_ids=direct_clip_ids if clip_source == "direct_clip_ids_field" else resolved_clip_map.get(event_id, []),
|
|
clip_source=clip_source,
|
|
)
|
|
for scene, index, record, event_id, resolved_id_field, direct_clip_ids, clip_source in pending_records
|
|
]
|
|
stats = EventResolutionStats(
|
|
total_events=len(pending_records),
|
|
cache_hits=lookup_stats.cache_hits,
|
|
cache_misses=lookup_stats.cache_misses,
|
|
request_workers=lookup_stats.request_workers,
|
|
cache_file=str(Path(cache_file).resolve()),
|
|
direct_clip_records=direct_clip_records,
|
|
event_lookup_records=event_lookup_records,
|
|
)
|
|
return resolved, stats
|
|
|
|
|
|
def build_event_resolution_payload(
|
|
json_file: str,
|
|
resolved_records: list[ResolvedEventRecord],
|
|
event_id_field: str,
|
|
clip_ids_field: str,
|
|
stats: EventResolutionStats,
|
|
selection_summary: Optional[dict[str, Any]] = None,
|
|
selected_record_keys: Optional[set[tuple[str, int, str]]] = None,
|
|
) -> dict[str, Any]:
|
|
selection_enabled = bool(selection_summary and selection_summary.get("enabled"))
|
|
selected_keys = selected_record_keys or set()
|
|
scenes: dict[str, list[dict[str, Any]]] = {}
|
|
for record in resolved_records:
|
|
scenes.setdefault(record.scene, []).append(
|
|
{
|
|
"record_index": record.record_index,
|
|
"event_id_field": event_id_field,
|
|
"event_id_field_used": record.event_id_field_used,
|
|
"event_id": record.event_id,
|
|
"event_clip_ids_field": clip_ids_field,
|
|
"clip_ids": record.clip_ids,
|
|
"clip_count": len(record.clip_ids),
|
|
"clip_source": record.clip_source,
|
|
"selected_for_inference": True if not selection_enabled else _record_key(record) in selected_keys,
|
|
"source_record": record.source_record,
|
|
}
|
|
)
|
|
|
|
payload = {
|
|
"event_json_file": json_file,
|
|
"event_id_field": event_id_field,
|
|
"event_clip_ids_field": clip_ids_field,
|
|
"scene_count": len(scenes),
|
|
"record_count": len(resolved_records),
|
|
"selected_record_count": len(selected_keys) if selection_enabled else len(resolved_records),
|
|
"resolution_stats": _build_resolution_stats_payload(stats),
|
|
"scenes": scenes,
|
|
}
|
|
if selection_summary is not None:
|
|
payload["selection"] = selection_summary
|
|
return payload
|
|
|
|
|
|
def save_event_resolution_manifest(
|
|
output_root: Path,
|
|
json_file: str,
|
|
resolved_records: list[ResolvedEventRecord],
|
|
event_id_field: str,
|
|
clip_ids_field: str,
|
|
stats: EventResolutionStats,
|
|
selection_summary: Optional[dict[str, Any]] = None,
|
|
selected_record_keys: Optional[set[tuple[str, int, str]]] = None,
|
|
) -> Path:
|
|
manifest_path = output_root / "_status" / "event_scene_manifest.json"
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = build_event_resolution_payload(
|
|
json_file,
|
|
resolved_records,
|
|
event_id_field,
|
|
clip_ids_field,
|
|
stats,
|
|
selection_summary=selection_summary,
|
|
selected_record_keys=selected_record_keys,
|
|
)
|
|
with manifest_path.open("w", encoding="utf-8") as file:
|
|
json.dump(payload, file, indent=2, ensure_ascii=False)
|
|
return manifest_path
|
|
|
|
|
|
def run_event_json_inference_exported(
|
|
*,
|
|
context: Any,
|
|
args: argparse.Namespace,
|
|
load_env: Callable[[], Any],
|
|
run_case_inference: Callable[..., dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
resolved_records, resolution_stats = resolve_event_records(
|
|
json_file=args.event_json_file,
|
|
scene_filter=args.scene,
|
|
event_id_field=args.event_id_field,
|
|
clip_ids_field=args.event_clip_ids_field,
|
|
max_events=args.max_events,
|
|
timeout=args.event_request_timeout,
|
|
cache_file=args.event_cache_file,
|
|
workers=args.event_resolve_workers,
|
|
max_retries=args.event_request_retries,
|
|
retry_backoff_sec=args.event_request_retry_backoff_sec,
|
|
)
|
|
selected_records, selection_summary = _select_event_records_by_condition(
|
|
resolved_records,
|
|
condition_fields=[str(field).strip() for field in getattr(args, "condition_fields", []) if str(field).strip()],
|
|
max_records_per_condition=int(getattr(args, "max_records_per_condition", 0)),
|
|
selection_strategy=str(getattr(args, "condition_select_strategy", "first")),
|
|
)
|
|
selected_record_keys = {_record_key(record) for record in selected_records}
|
|
output_root = Path(args.output_dir).resolve()
|
|
output_root.mkdir(parents=True, exist_ok=True)
|
|
manifest_path = save_event_resolution_manifest(
|
|
output_root=output_root,
|
|
json_file=args.event_json_file,
|
|
resolved_records=resolved_records,
|
|
event_id_field=args.event_id_field,
|
|
clip_ids_field=args.event_clip_ids_field,
|
|
stats=resolution_stats,
|
|
selection_summary=selection_summary,
|
|
selected_record_keys=selected_record_keys,
|
|
)
|
|
|
|
summary_by_scene: dict[str, dict[str, Any]] = {}
|
|
for scene in sorted({record.scene for record in selected_records}):
|
|
scene_records = [record for record in selected_records if record.scene == scene]
|
|
scene_selection_summary = _filter_selection_summary_for_scene(selection_summary, scene)
|
|
scene_export_root = Path(args.export_root).resolve() / scene
|
|
scene_output_root = output_root / scene
|
|
scene_output_root.mkdir(parents=True, exist_ok=True)
|
|
scene_manifest_path = scene_output_root / "_status" / "scene_event_manifest.json"
|
|
scene_manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with scene_manifest_path.open("w", encoding="utf-8") as file:
|
|
json.dump(
|
|
_build_scene_manifest_payload(
|
|
args=args,
|
|
scene=scene,
|
|
scene_records=scene_records,
|
|
resolution_stats=resolution_stats,
|
|
selection_summary=scene_selection_summary,
|
|
),
|
|
file,
|
|
indent=2,
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
if args.selection_only:
|
|
summary_by_scene[scene] = {
|
|
"scene_output_dir": str(scene_output_root),
|
|
"scene_manifest_path": str(scene_manifest_path),
|
|
"selected_record_count": len(scene_records),
|
|
"selection_only": True,
|
|
}
|
|
continue
|
|
|
|
event_records_by_id: dict[str, list[ResolvedEventRecord]] = {}
|
|
event_order: list[str] = []
|
|
for record in scene_records:
|
|
if record.event_id not in event_records_by_id:
|
|
event_order.append(record.event_id)
|
|
event_records_by_id.setdefault(record.event_id, []).append(record)
|
|
|
|
selected_scene_clip_ids: set[str] = set()
|
|
event_results: dict[str, dict[str, Any]] = {}
|
|
event_dirs: dict[str, str] = {}
|
|
event_export_dirs: dict[str, str] = {}
|
|
for event_id in event_order:
|
|
event_records = event_records_by_id[event_id]
|
|
event_clip_ids = _dedupe_preserve_order(
|
|
[clip_id for record in event_records for clip_id in record.clip_ids]
|
|
)
|
|
if args.limit_clips > 0:
|
|
limited_clip_ids: list[str] = []
|
|
for clip_id in event_clip_ids:
|
|
if clip_id in selected_scene_clip_ids:
|
|
limited_clip_ids.append(clip_id)
|
|
continue
|
|
if len(selected_scene_clip_ids) >= args.limit_clips:
|
|
continue
|
|
selected_scene_clip_ids.add(clip_id)
|
|
limited_clip_ids.append(clip_id)
|
|
event_clip_ids = limited_clip_ids
|
|
|
|
clip_tasks = build_clip_tasks_from_clip_ids(event_clip_ids)
|
|
dir_prefix = "event_id"
|
|
if any(record.clip_source == "direct_clip_ids_field" for record in event_records):
|
|
dir_prefix = event_records[0].event_id_field_used or args.event_id_field or "record_id"
|
|
event_dir_name = _sanitize_identifier_for_path(event_id, prefix=dir_prefix)
|
|
event_export_root = scene_export_root / event_dir_name
|
|
event_output_root = scene_output_root / event_dir_name
|
|
event_export_dirs[event_id] = str(event_export_root)
|
|
event_dirs[event_id] = str(event_output_root)
|
|
|
|
def on_before_run(export_root: Path, inference_root: Path, tasks: list[Any], *, _event_id: str = event_id, _event_records: list[ResolvedEventRecord] = event_records, _event_clip_ids: list[str] = event_clip_ids) -> None:
|
|
payload = _build_event_manifest_payload(
|
|
args=args,
|
|
scene=scene,
|
|
event_id=_event_id,
|
|
event_records=_event_records,
|
|
clip_ids=_event_clip_ids,
|
|
resolution_stats=resolution_stats,
|
|
)
|
|
manifest_path = inference_root / "_status" / "event_manifest.json"
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with manifest_path.open("w", encoding="utf-8") as file:
|
|
json.dump(payload, file, indent=2, ensure_ascii=False)
|
|
|
|
event_results[event_id] = run_clip_tasks_inference_exported(
|
|
context=context,
|
|
args=args,
|
|
clip_tasks=clip_tasks,
|
|
load_env=load_env,
|
|
run_case_inference=run_case_inference,
|
|
export_root=event_export_root,
|
|
output_root=event_output_root,
|
|
on_before_run=on_before_run,
|
|
)
|
|
|
|
summary_by_scene[scene] = {
|
|
"scene_output_dir": str(scene_output_root),
|
|
"scene_manifest_path": str(scene_manifest_path),
|
|
"event_export_dirs": event_export_dirs,
|
|
"event_dirs": event_dirs,
|
|
"event_results": event_results,
|
|
}
|
|
|
|
result = {
|
|
"event_json_file": args.event_json_file,
|
|
"scene": args.scene or "",
|
|
"event_id_field": args.event_id_field,
|
|
"event_clip_ids_field": args.event_clip_ids_field,
|
|
"manifest_path": str(manifest_path),
|
|
"selection": selection_summary,
|
|
"selected_record_count": len(selected_records),
|
|
"resolution_stats": _build_resolution_stats_payload(resolution_stats),
|
|
"scene_results": summary_by_scene,
|
|
}
|
|
_print_event_json_summary(
|
|
args=args,
|
|
manifest_path=manifest_path,
|
|
resolved_records=resolved_records,
|
|
selected_records=selected_records,
|
|
selection_summary=selection_summary,
|
|
scene_results=summary_by_scene,
|
|
)
|
|
return result
|