reprostim.qr.video_audit

API to analyze video files recorded by reprostim-videocapture, along with their corresponding log files and QR/audio metadata. It extracts key information about each recording and produces a summary table (videos.tsv) suitable for quality control, sharing, and further analysis.

Functions

audio_codec_to_rfc6381(codec, profile)

Return the RFC 6381 codec string for an audio stream.

check_coherent(vr)

Check if the video record is coherent.

check_ffprobe()

Check if ffprobe is installed and available in PATH.

do_audit(ctx, paths_dir_or_file)

Generator that audits files and applies all external tools to each record if any, depending on context and options.

do_audit_dir(ctx, path)

Audit video files in directory with .mkv, .mp4, .avi extensions.

do_audit_file(ctx, path)

Audit a single video file.

do_audit_internal(ctx, paths_dir_or_file)

Audit a single video file or all video files in a directory.

do_ext(ctx, recs, paths_dir_or_file)

Generator that runs external tools on existing records depending on context and options.

do_main(paths, path_tsv[, recursive, mode, ...])

The main function invoked by CLI to analyze video files with logs and save the results to a TSV file.

find_metadata_json(path, key, value)

Find the first metadata JSON entry with a specific key-value pair.

find_video_audit_by_timerange(path_tsv, ...)

Find all VaRecords whose recording interval intersects the given time range, sorted by record start time ascending.

format_date(dt)

Extract date from datetime as 'YYYY-MM-DD' string.

format_duration(duration_sec)

Convert duration in seconds to human-readable string HH:MM:SS or HH:MM:SS.mmm.

format_time(dt)

Extract time from datetime as 'HH:MM:SS.mmm' string.

format_tts(t)

Converts a time.time() timestamp into an ISO 8601 formatted string with local timezone.

get_audio_video_info_ffprobe(path)

Extract audio and video stream information from the video file using ffprobe.

get_file_video_audit(path[, path_tsv, ...])

Get a single VaRecord by auditing a single video file.

iter_metadata_json(log_path)

Iterate over all REPROSTIM-METADATA-JSON lines in the log file.

run_ext_all(ctx, vr)

Run all external tools on the specified VaRecord.

run_ext_nosignal(ctx, vr)

Run detect-noscreen external tools on the specified VaRecord.

run_ext_qr(ctx, vr)

Run qr-parse external tool on the specified VaRecord.

video_codec_to_rfc6381(codec, profile, level)

Return the RFC 6381 codec string for a video stream.

Classes

AudioInfo(*[, bits_per_sample, channels, ...])

Audio stream information extracted from the video file with ffprobe.

VaContext(*[, c_internal, c_nosignal, c_qr, ...])

Context for video audit processing.

VaMode(value)

Video audit processing mode constants.

VaRecord(*[, path, present, complete, name, ...])

A single record in the videos.tsv audit summary.

VaSource(value)

Video audit source constants.

VideoInfo(*[, bit_depth, codec, codec_long, ...])

Video stream information extracted from the video file with ffprobe.
class reprostim.qr.video_audit.AudioInfo(*, bits_per_sample: int | None = None, channels: int | None = None, codec: str | None = None, codec_long: str | None = None, codec_rfc6381: str | None = None, duration_sec: float | None = None, profile: str | None = None, sample_rate: int | None = None, start_time: float | None = None, tag_str: str | None = None)[source]

Audio stream information extracted from the video file with ffprobe.

bits_per_sample: int | None
channels: int | None
codec: str | None
codec_long: str | None
codec_rfc6381: str | None
duration_sec: float | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

profile: str | None
sample_rate: int | None
start_time: float | None
tag_str: str | None
class reprostim.qr.video_audit.VaContext(*, c_internal: int | None = 0, c_nosignal: int | None = 0, c_qr: int | None = 0, log_level: str | None = None, max_counter: int | None = -1, mode: VaMode | None = VaMode.INCREMENTAL, nosignal_data_dir: str | None = 'derivatives/nosignal', nosignal_log_dir: str | None = 'logs/nosignal', nosignal_opts: List | None = ['--number-of-checks', '100', '--truncated', 'fixup', '--invalid-timing', 'fixup', '--threshold', '1.01'], path_mask: str | None = None, qr_data_dir: str | None = 'derivatives/qr', qr_log_dir: str | None = 'logs/qr', qr_opts: List | None = [], recursive: bool | None = False, skip_names: set | None = None, source: Set[VaSource] | None = {VaSource.INTERNAL}, updated_paths: set | None = {})[source]

Context for video audit processing.

c_internal: int | None

Count of files processed with INTERNAL source

c_nosignal: int | None

Count of files processed with NOSIGNAL source

c_qr: int | None

Count of files processed with QR source

log_level: str | None

Logging level to be used in external tool, one of DEBUG, INFO, WARNING, ERROR, CRITICAL (default: None)

max_counter: int | None

Max number of records to process or -1 for unlimited (default: -1)

mode: VaMode | None

INCREMENTAL)

Type:

Operation mode, one of VaMode values (default

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nosignal_data_dir: str | None

Directory to store nosignal output data in JSON format.

nosignal_log_dir: str | None

Directory to store nosignal logs.

nosignal_opts: List | None

Additional options to pass to detect-noscreen tool.

path_mask: str | None

Optional path mask to filter video files based on their paths.

qr_data_dir: str | None

Directory to store qr-parse output data in JSON format.

qr_log_dir: str | None

Directory to store qr-parse logs.

qr_opts: List | None

Additional options to pass to qr-parse tool if any.

recursive: bool | None

False

Type:

Whether to scan directories recursively. Default

skip_names: set | None

Optional set of file base names to skip (for incremental mode)

source: Set[VaSource] | None

INTERNAL)

Type:

One of VaSource values to specify audit source (default

updated_paths: set | None

Optional set of updated record paths

class reprostim.qr.video_audit.VaMode(value)[source]

Video audit processing mode constants.

FORCE = 'force'

Force redo/overwrite specified records in TSV

FULL = 'full'

Process all files, overwrite existing records in TSV completely.

INCREMENTAL = 'incremental'

Process only new files not present in existing TSV and keep existing records.

RERUN_FOR_NA = 'rerun-for-na'

Process only records with ‘n/a’ values in existing TSV in columns that are usually filled by external tools. Intended for run external slow tools like detect-noscreen or qr-parser.

RESET_TO_NA = 'reset-to-na'

Reset specified columns to ‘n/a’ in existing TSV. Intended to clear results of external tools like detect-noscreen or qr-parser to rerun them from scratch.

class reprostim.qr.video_audit.VaRecord(*, path: str = 'n/a', present: bool = False, complete: bool = False, name: str = 'n/a', start_date: str = 'n/a', start_time: str = 'n/a', end_date: str = 'n/a', end_time: str = 'n/a', video_res_detected: str = 'n/a', video_fps_detected: str = 'n/a', video_dur_detected: str = 'n/a', video_res_recorded: str = 'n/a', video_fps_recorded: str = 'n/a', video_dur_recorded: str = 'n/a', video_size_mb: str = 'n/a', video_rate_mbpm: str = 'n/a', audio_sr: str = 'n/a', audio_dur: str = 'n/a', duration: str = 'n/a', duration_h: str = 'n/a', no_signal_frames: str = 'n/a', qr_records_number: str = 'n/a', file_log_coherent: bool = False, no_signal_updated_on: str = 'n/a', qr_updated_on: str = 'n/a', updated_on: str = 'n/a')[source]

A single record in the videos.tsv audit summary.

audio_dur: str
audio_sr: str
complete: bool
duration: str
duration_h: str
end_date: str
end_time: str
file_log_coherent: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
no_signal_frames: str
no_signal_updated_on: str
path: str
present: bool
qr_records_number: str
qr_updated_on: str
start_date: str
start_time: str
updated_on: str
video_dur_detected: str
video_dur_recorded: str
video_fps_detected: str
video_fps_recorded: str
video_rate_mbpm: str
video_res_detected: str
video_res_recorded: str
video_size_mb: str
class reprostim.qr.video_audit.VaSource(value)[source]

Video audit source constants.

ALL = 'all'

Run all available audit sources.

INTERNAL = 'internal'

Basic and default behaviour to process quickly video files using only mediainfo and logs metadata.

NOSIGNAL = 'nosignal'

Process video files to detect no-signal frames. This is slow process and can take some time depending on the video length.

QR = 'qr'

Process video files to extract QR codes metadata. This is very slow process and time is almost the same as video duration at this moment.

class reprostim.qr.video_audit.VideoInfo(*, bit_depth: int | None = None, codec: str | None = None, codec_long: str | None = None, codec_rfc6381: str | None = None, duration_sec: float | None = None, fps: float | None = None, height: int | None = None, level: int | None = None, pix_fmt: str | None = None, profile: str | None = None, start_time: float | None = None, tag_str: str | None = None, width: int | None = None)[source]

Video stream information extracted from the video file with ffprobe.

bit_depth: int | None
codec: str | None
codec_long: str | None
codec_rfc6381: str | None
duration_sec: float | None
fps: float | None
height: int | None
level: int | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pix_fmt: str | None
profile: str | None
start_time: float | None
tag_str: str | None
width: int | None
reprostim.qr.video_audit.audio_codec_to_rfc6381(codec: str, profile: str | None) → str | None[source]

Return the RFC 6381 codec string for an audio stream.

Parameters:

  • codec (str) – Short codec name as reported by ffprobe (e.g. "aac").

  • profile (Optional[str]) – Codec profile string (e.g. "LC"), or None.

Returns:

RFC 6381 string (e.g. "mp4a.40.2"), or None when the codec is not recognized.

Return type:

Optional[str]

reprostim.qr.video_audit.check_coherent(vr: VaRecord) → bool[source]

Check if the video record is coherent.

reprostim.qr.video_audit.check_ffprobe()[source]

Check if ffprobe is installed and available in PATH. :return: True if ffprobe is available, False otherwise :rtype: bool

reprostim.qr.video_audit.do_audit(ctx: VaContext, paths_dir_or_file: List[str]) → Generator[VaRecord, None, None][source]

Generator that audits files and applies all external tools to each record if any, depending on context and options.

reprostim.qr.video_audit.do_audit_dir(ctx: VaContext, path: str) → Generator[VaRecord, None, None][source]

Audit video files in directory with .mkv, .mp4, .avi extensions.

Parameters:

  • ctx (VaContext) – VaContext object with processing context

  • path (str) – Path to the directory

Returns:

Generator of VaRecord objects

Return type:

Generator[VaRecord, None, None]

reprostim.qr.video_audit.do_audit_file(ctx: VaContext, path: str) → Generator[VaRecord, None, None][source]

Audit a single video file.

Parameters:

  • ctx (VaContext) – VaContext object with processing context

  • path (str) – Path to the video file

Returns:

Generator of VaRecord objects

Return type:

Generator[VaRecord, None, None]

reprostim.qr.video_audit.do_audit_internal(ctx: VaContext, paths_dir_or_file: List[str]) → Generator[VaRecord, None, None][source]

Audit a single video file or all video files in a directory.

Parameters:

  • ctx (VaContext) – VaContext object with processing context

  • paths_dir_or_file (List[str]) – List of path to the video file or directory

Returns:

Generator of VaRecord objects

Return type:

Generator[VaRecord, None, None]

reprostim.qr.video_audit.do_ext(ctx: VaContext, recs: List[VaRecord], paths_dir_or_file: List[str]) → Generator[VaRecord, None, None][source]

Generator that runs external tools on existing records depending on context and options.

reprostim.qr.video_audit.do_main(paths: ~typing.List[str], path_tsv: str, recursive: bool = False, mode: ~reprostim.qr.video_audit.VaMode = VaMode.INCREMENTAL, va_src: ~typing.Set[~reprostim.qr.video_audit.VaSource] | None = None, max_files: int = -1, path_mask: str | None = None, verbose: bool = False, out_func=<built-in function print>, nosignal_opts: str | None = None, qr_opts: str | None = None)[source]

The main function invoked by CLI to analyze video files with logs and save the results to a TSV file.

Parameters:

  • paths (List[str]) – One or more paths to the video file or directory

  • path_tsv (str) – Path to the output TSV file, default ‘videos.tsv’.

  • recursive (bool) – Whether to scan directories recursively. Default: False

  • mode (VaMode) – Operation mode, one of VaMode values (default: INCREMENTAL)

  • va_src (Set[VaSource]) – Set of VaSource values to specify audit sources. Default: {VaSource.INTERNAL}.

  • max_files (int) – Maximum number of video files/records to process. Use -1 for unlimited (default: -1)

  • path_mask (str) – Optional fnmatch-style mask to filter files

  • verbose (bool) – Whether to print verbose JSON output to stdout (default: False)

  • out_func (Callable[[str], None]) – Function to stdout results (default: print)

  • nosignal_opts (str) – Optional string of extra options to pass to detect-noscreen, parsed via shlex. Overrides the built-in defaults when provided.

  • qr_opts (str) – Optional string of extra options to pass to qr-parse, parsed via shlex. No extra options by default.

Returns:

0 on success, 1 on failure

Return type:

int

reprostim.qr.video_audit.find_metadata_json(path: str, key: str, value) → Dict | None[source]

Find the first metadata JSON entry with a specific key-value pair.

Parameters:

  • path (str) – Path to the log file

  • key (str) – Key to search for

  • value (Any) – Value to match

Returns:

The first matching dictionary or None if not found

Return type:

Optional[Dict]

reprostim.qr.video_audit.find_video_audit_by_timerange(path_tsv: str, start: datetime, end: datetime, cached: bool = False, use_lock: bool = True) → List[VaRecord][source]

Find all VaRecords whose recording interval intersects the given time range, sorted by record start time ascending.

Only records with both present and complete flags set to True are considered. A record intersects [start, end] when its start is before end and its end is after start.

Parameters:

  • path_tsv (str) – Path to the TSV file to search.

  • start (datetime) – Start of the query time range (inclusive).

  • end (datetime) – End of the query time range (inclusive).

  • cached (bool) – If True, return cached TSV data when available instead of reloading from disk. Default: False.

  • use_lock (bool) – If True (default), acquire the advisory lock before loading from disk. If False, skip the lock (dirty-read mode). Default: True.

Returns:

List of matching VaRecord objects sorted by start time ascending.

Return type:

List[VaRecord]

reprostim.qr.video_audit.format_date(dt: datetime) → str[source]

Extract date from datetime as ‘YYYY-MM-DD’ string. Ignores timezone.

Parameters:

dt (datetime) – datetime object

Returns:

formatted date string

Return type:

str

reprostim.qr.video_audit.format_duration(duration_sec: float) → str[source]

Convert duration in seconds to human-readable string HH:MM:SS or HH:MM:SS.mmm. If duration_sec is None, return “n/a”

Parameters:

duration_sec (float) – Duration in seconds

Returns:

Formatted duration string

Return type:

str

reprostim.qr.video_audit.format_time(dt: datetime) → str[source]

Extract time from datetime as ‘HH:MM:SS.mmm’ string. Ignores timezone.

Parameters:

dt (datetime) – datetime object

Returns:

formatted time string

Return type:

str

reprostim.qr.video_audit.format_tts(t: float) → str[source]

Converts a time.time() timestamp into an ISO 8601 formatted string with local timezone.

This function takes a timestamp in seconds and converts it into a datetime object with local timezone.

Parameters:

t (float) – The timestamp to be converted in seconds.

Returns:

The ISO 8601 formatted string with local timezone.

Return type:

str

reprostim.qr.video_audit.get_audio_video_info_ffprobe(path: str) → Tuple[AudioInfo, VideoInfo][source]

Extract audio and video stream information from the video file using ffprobe.

Issues a single ffprobe call that reads all streams, then splits results into an AudioInfo (first audio stream) and a VideoInfo (first video stream).

Parameters:

path (str) – Path to the video file (.mkv, .mp4, .avi)

Returns:

Tuple of (AudioInfo, VideoInfo) with extracted stream information. Fields are None when the corresponding stream is absent or a value cannot be parsed.

Return type:

Tuple[AudioInfo, VideoInfo]

reprostim.qr.video_audit.get_file_video_audit(path: str, path_tsv: str | None = None, cached: bool = False, use_lock: bool = True) → VaRecord[source]

Get a single VaRecord by auditing a single video file.

Parameters:

  • path (str) – Path to the video file.

  • path_tsv (str) – Optional path to existing TSV file to load existing records from. Default: None.

  • cached (bool) – If True, return cached TSV data when available instead of reloading from disk. Passed through to _get_tsv_records(). Default: False.

  • use_lock (bool) – If True (default), acquire the advisory file lock before reading path_tsv. If False, skip the lock (dirty-read mode).

Returns:

VaRecord object.

Return type:

VaRecord

reprostim.qr.video_audit.iter_metadata_json(log_path: str) → Generator[Dict, None, None][source]

Iterate over all REPROSTIM-METADATA-JSON lines in the log file. Yields parsed JSON dictionaries.

Parameters:

log_path (str) – Path to the log file

Returns:

Generator of parsed JSON dictionaries

Return type:

Generator[Dict, None, None]

reprostim.qr.video_audit.run_ext_all(ctx: VaContext, vr: VaRecord) → VaRecord[source]

Run all external tools on the specified VaRecord.

Parameters:

  • ctx (VaContext) – VaContext object with processing context

  • vr (VaRecord) – VaRecord object to process

Returns:

Updated VaRecord object

Return type:

VaRecord

reprostim.qr.video_audit.run_ext_nosignal(ctx: VaContext, vr: VaRecord) → VaRecord[source]

Run detect-noscreen external tools on the specified VaRecord.

Parameters:

  • ctx (VaContext) – VaContext object with processing context

  • vr (VaRecord) – VaRecord object to process

Returns:

Updated VaRecord object

Return type:

VaRecord

reprostim.qr.video_audit.run_ext_qr(ctx: VaContext, vr: VaRecord) → VaRecord[source]

Run qr-parse external tool on the specified VaRecord.

Parameters:

  • ctx (VaContext) – VaContext object with processing context

  • vr (VaRecord) – VaRecord object to process

Returns:

Updated VaRecord object

Return type:

VaRecord

reprostim.qr.video_audit.video_codec_to_rfc6381(codec: str, profile: str | None, level: int | None) → str | None[source]

Return the RFC 6381 codec string for a video stream.

Parameters:

  • codec (str) – Short codec name as reported by ffprobe (e.g. "h264").

  • profile (Optional[str]) – Codec profile string (e.g. "High"), or None.

  • level (Optional[int]) – Codec level integer as reported by ffprobe (e.g. 42), or None.

Returns:

RFC 6381 string (e.g. "avc1.64002A"), or None when the codec is not recognised.

Return type:

Optional[str]