#!/usr/bin/env python3 # SPDX-License-Identifier: MIT """Operavue video timestamp audit v1.0. Python 3.9+, standard library only. Inspect local FFprobe JSON for one selected video stream, in the returned presentation order. No media is opened and no network requests are made. Supply a task-specific --max-gap-ms threshold. Decimal results are JSON strings. Missing keys, null, and FFprobe's N/A sentinel count as missing timestamps. Decimals support at most 128 coefficient digits and an exponent from -1000 to 1000. Zero comparable adjacent pairs is an insufficient_adjacent_pairs finding. Exit 0: no timestamp findings; 1: findings; 2: invalid input or CLI arguments. No findings is not a quality pass. Read the scope statement in every report. MIT License Copyright (c) 2026 Operavue Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import argparse from decimal import Decimal, InvalidOperation, localcontext, MAX_EMAX, MIN_EMIN import json from pathlib import Path import re import sys VERSION = "1.0" PREVIEW_LIMIT = 20 MAX_COEFFICIENT_DIGITS = 128 MAX_ABSOLUTE_EXPONENT = 1000 TIME_FIELD = "best_effort_timestamp_time" DECIMAL_SECONDS = re.compile(r"[+-]?(?:[0-9]+(?:\.[0-9]*)?|\.[0-9]+)(?:[eE][+-]?[0-9]+)?\Z") SCOPE = ( "One selected video stream in the supplied presentation order; timestamps are not sorted. " "Stream metadata coverage is reported; absent metadata leaves selection unverified. " "A gap candidate is an adjacent timestamp interval above the supplied threshold, not proof " "of dropped frames. Variable frame rate and timestamp changes can produce such intervals. " "Duplicate timestamps do not prove duplicate image content. Missing or invalid times break " "adjacency comparisons. Zero comparable adjacent pairs produces an insufficient_adjacent_pairs " "finding because no interval comparison was possible. This report cannot establish capture-clock synchronization, stereo " "alignment, content duplicates, or clinical suitability. No timestamp findings is not a " "quality pass. First and last valid timestamps are observations, not recording duration." ) def decimal_value(value): """Accept finite decimal numbers, excluding booleans and non-decimal strings.""" if isinstance(value, bool) or not isinstance(value, (str, int, float, Decimal)): raise ValueError("Expected finite decimal seconds.") text = str(value).strip() if not DECIMAL_SECONDS.fullmatch(text): raise ValueError("Expected finite decimal seconds.") try: number = Decimal(text) except InvalidOperation: raise ValueError("Expected finite decimal seconds.") from None if not number.is_finite(): raise ValueError("Expected finite decimal seconds.") parts = number.as_tuple() if len(parts.digits) > MAX_COEFFICIENT_DIGITS or abs(parts.exponent) > MAX_ABSOLUTE_EXPONENT: raise ValueError("Decimal exceeds 128 coefficient digits or the supported exponent range -1000 to 1000.") return number def positive_threshold(value): try: number = decimal_value(value) except ValueError: raise argparse.ArgumentTypeError("--max-gap-ms must be finite and positive.") from None if number <= 0: raise argparse.ArgumentTypeError("--max-gap-ms must be finite and positive.") return number def decimal_text(value): text = str(value) if "E" not in text and "." in text: text = text.rstrip("0").rstrip(".") return "0" if value == 0 else text def precision_for(left, right): """Enough coefficient digits for exact addition/subtraction, including unlike scales.""" first, second = left.as_tuple(), right.as_tuple() exponent = min(first.exponent, second.exponent) return max(len(first.digits) + first.exponent - exponent, len(second.digits) + second.exponent - exponent) + 4 def interval_ms(current, previous): with localcontext() as context: context.prec = precision_for(current, previous) context.Emax, context.Emin = MAX_EMAX, MIN_EMIN return (current - previous) * Decimal(1000) def interval_summary(intervals): if not intervals: return {"min": None, "median": None, "max": None} # Sort only the interval sample for its median; frame timestamps stay in input order. ordered = sorted(intervals) midpoint = len(ordered) // 2 median = ordered[midpoint] if len(ordered) % 2 == 0: left, right = ordered[midpoint - 1], ordered[midpoint] with localcontext() as context: context.prec = precision_for(left, right) context.Emax, context.Emin = MAX_EMAX, MIN_EMIN median = (left + right) / Decimal(2) return {"min": decimal_text(ordered[0]), "median": decimal_text(median), "max": decimal_text(ordered[-1])} def audit(document, max_gap_ms, input_name=None): threshold = decimal_value(max_gap_ms) if threshold <= 0: raise ValueError("The gap threshold must be positive.") if not isinstance(document, dict) or not isinstance(document.get("frames"), list): raise ValueError("Expected an object containing a frames array.") frames = document["frames"] if not frames: raise ValueError("The frames array is empty.") stream_indices = set() frames_with_stream_index = frames_with_media_type = 0 for frame in frames: if not isinstance(frame, dict): raise ValueError("Every frame must be a JSON object.") if "media_type" in frame and frame["media_type"] != "video": raise ValueError("Select one video stream; non-video frame metadata was found.") if "media_type" in frame: frames_with_media_type += 1 if "stream_index" in frame: index = frame["stream_index"] if isinstance(index, bool) or not isinstance(index, int) or index < 0: raise ValueError("A supplied stream_index must be a nonnegative integer.") stream_indices.add(index) frames_with_stream_index += 1 if len(stream_indices) > 1: raise ValueError("Select one video stream; multiple stream indices were found.") counts = dict(frames=len(frames), valid_timestamps=0, missing_timestamps=0, invalid_timestamps=0, adjacent_pairs=len(frames) - 1, compared_pairs=0, skipped_pairs=0, positive_intervals=0, duplicate_intervals=0, backward_intervals=0, gap_candidates=0, issues_total=0) preview, positive_intervals = [], [] previous = first_valid = last_valid = None def issue(kind, frame_index=None, **details): counts["issues_total"] += 1 if len(preview) < PREVIEW_LIMIT: preview.append({"type": kind, **({"frame_index": frame_index} if frame_index is not None else {}), **details}) for index, frame in enumerate(frames): raw = frame.get(TIME_FIELD) current = None if raw is None or (isinstance(raw, str) and raw.strip() == "N/A"): counts["missing_timestamps"] += 1 issue("missing_timestamp", index) else: try: current = decimal_value(raw) except ValueError as error: counts["invalid_timestamps"] += 1 issue("invalid_timestamp", index, reason=str(error)) if current is not None: counts["valid_timestamps"] += 1 if first_valid is None: first_valid = current last_valid = current if index: if previous is None or current is None: counts["skipped_pairs"] += 1 else: counts["compared_pairs"] += 1 delta = interval_ms(current, previous) details = {"previous_frame_index": index - 1, "interval_ms": decimal_text(delta)} if delta < 0: counts["backward_intervals"] += 1 issue("backward_timestamp", index, **details) elif delta == 0: counts["duplicate_intervals"] += 1 issue("duplicate_timestamp", index, **details) else: counts["positive_intervals"] += 1 positive_intervals.append(delta) if delta > threshold: counts["gap_candidates"] += 1 issue("gap_candidate", index, **details) previous = current if counts["compared_pairs"] == 0: issue("insufficient_adjacent_pairs", reason="No interval comparison was possible; at least one adjacent pair needs two valid timestamps.") return { "tool": "Operavue video timestamp audit", "version": VERSION, "status": "timestamp_findings" if counts["issues_total"] else "no_timestamp_findings", "input": Path(input_name).name if input_name is not None else None, "timestamp_field": TIME_FIELD, "frame_index_base": 0, "stream_metadata": {"observed_indices": sorted(stream_indices), "frames_with_stream_index": frames_with_stream_index, "frames_with_media_type": frames_with_media_type}, "max_gap_ms": decimal_text(threshold), "counts": counts, "first_valid_timestamp_seconds": decimal_text(first_valid) if first_valid is not None else None, "last_valid_timestamp_seconds": decimal_text(last_valid) if last_valid is not None else None, "positive_adjacent_intervals_ms": interval_summary(positive_intervals), "issue_preview": preview, "issue_preview_limit": PREVIEW_LIMIT, "omitted_issues": counts["issues_total"] - len(preview), "scope": SCOPE } def reject_constant(_value): raise ValueError("Nonfinite JSON constants are not valid input.") def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("Duplicate JSON keys are not supported.") result[key] = value return result def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("input", help="Local FFprobe JSON for one selected video stream") parser.add_argument("--max-gap-ms", type=positive_threshold, required=True, help="Flag positive adjacent intervals strictly above this task-specific threshold") args = parser.parse_args(argv) try: with open(args.input, encoding="utf-8-sig") as handle: document = json.load(handle, parse_float=Decimal, parse_constant=reject_constant, object_pairs_hook=unique_object) report = audit(document, args.max_gap_ms, input_name=args.input) except (OSError, UnicodeError): report = {"status": "invalid_input", "input": Path(args.input).name, "error": "Could not read the input as a UTF-8 JSON file."} print(json.dumps(report)) return 2 except (ValueError, ArithmeticError) as error: message = "Input is not valid JSON." if isinstance(error, json.JSONDecodeError) else str(error) print(json.dumps({"status": "invalid_input", "input": Path(args.input).name, "error": message})) return 2 print(json.dumps(report, indent=2, ensure_ascii=True, allow_nan=False)) return 1 if report["counts"]["issues_total"] else 0 if __name__ == "__main__": sys.exit(main())