# SPDX-FileCopyrightText: 2020-2026 ReproNim ReproStim Team <reprostim@repronim.org>
#
# SPDX-License-Identifier: MIT
"""
Core logic for bids-inject: cross-reference ReproStim videos.tsv with BIDS
_scans.tsv files to slice and inject per-acquisition video clips into a BIDS
dataset.
See .ai/bids/inject-spec.md for the full specification.
"""
import contextlib
import csv
import io
import json
import logging
import os
import re
from datetime import datetime, time, timedelta, timezone, tzinfo
from enum import Enum
from functools import lru_cache
from importlib.resources import files
from typing import Callable, List, Optional, Tuple
from zoneinfo import ZoneInfo
from pydantic import BaseModel, Field
from reprostim.__about__ import __version__
from reprostim.bids.properties import bids_properties_from_ffprobe
from reprostim.video.audit import find_video_audit_by_timerange
# initialize the logger
logger = logging.getLogger(__name__)
logger.debug(f"name={__name__}")
# Try to import datalad-fuse, but don't require it
try:
from datalad_fuse.fsspec import FsspecAdapter
DATALAD_FUSE_AVAILABLE = True
except ImportError:
DATALAD_FUSE_AVAILABLE = False
FsspecAdapter = None
logger.debug(
"datalad-fuse available: FsspecAdapter enabled for annexed file I/O"
if DATALAD_FUSE_AVAILABLE
else "datalad-fuse not available: using plain file I/O"
)
####################################################################
# Enumerations
####################################################################
[docs]
class BufferPolicy(str, Enum):
"""Policy for handling buffer overflow beyond video boundaries."""
STRICT = "strict"
FLEXIBLE = "flexible"
[docs]
class QrMode(str, Enum):
"""QR code-based timing refinement mode."""
NONE = "none"
AUTO = "auto"
EMBED_EXISTING = "embed-existing"
PARSE = "parse"
[docs]
class LayoutMode(str, Enum):
"""Output file placement layout within the BIDS dataset."""
NEARBY = "nearby"
TOP_STIMULI = "top-stimuli"
[docs]
class OverwriteMode(str, Enum):
"""Policy for handling existing output files."""
SKIP = "skip"
FORCE = "force"
ALWAYS = "always"
ERROR = "error"
[docs]
class ScanRecordKind(str, Enum):
"""Classification of a ``_scans.tsv`` row's ``filename`` by file type.
Distinguishes NIfTI acquisition rows (compressed/uncompressed) from
dedicated ReproStim media rows (Output A, see inject-spec.md Output D)
and any other/unrecognized row type (e.g. a future
``_events.tsv``/``_physio.tsv``). Computed once, from a raw filename, via
:func:`_calc_scan_record_kind` — see :attr:`ScanRecord.kind`. Replaces the
earlier ad hoc ``_is_media_row(filename)`` boolean check with a single,
inspectable classification reused everywhere a row's type matters.
"""
NIFTI_GZ = "nifti_gz"
NIFTI = "nifti"
REPROSTIM_VIDEO = "reprostim_video"
REPROSTIM_AUDIO = "reprostim_audio"
REPROSTIM_AUDIOVIDEO = "reprostim_audiovideo"
UNKNOWN = "unknown"
@property
def is_reprostim_media(self) -> bool:
"""``True`` for any of the three ReproStim media row kinds."""
return self in (
ScanRecordKind.REPROSTIM_VIDEO,
ScanRecordKind.REPROSTIM_AUDIO,
ScanRecordKind.REPROSTIM_AUDIOVIDEO,
)
@property
def is_nifti(self) -> bool:
"""``True`` for either NIfTI variant (compressed or uncompressed)."""
return self in (ScanRecordKind.NIFTI_GZ, ScanRecordKind.NIFTI)
####################################################################
# Data models
####################################################################
[docs]
class BiSummary(BaseModel):
"""Mutable counters accumulating per-run injection statistics."""
n_processed: int = Field(
default=0,
description="Scan records considered for injection (matched by --match filter).",
)
n_injected: int = Field(
default=0,
description="Scan records successfully injected (or planned in dry-run).",
)
n_skipped: int = Field(
default=0,
description="Scan records skipped (no video match, no media stream, etc.).",
)
n_errors: int = Field(
default=0,
description="Scan records that encountered errors (ambiguous match, "
"split-video failure, etc.).",
)
errors: List[str] = Field(
default_factory=list,
description="Human-readable description of each error encountered.",
)
[docs]
class BiContext(BaseModel):
"""Context for bids-inject processing of scan records."""
dataset_home: str = Field(
default=".",
description="Home directory of the BIDS dataset being injected into "
"(e.g. contains scans.json). Defaults to the current directory.",
)
dry_run: bool = Field(
..., description="Whether to skip actual file writes and print planned actions"
)
recursive: bool = Field(
..., description="Whether to search for _scans.tsv files recursively"
)
match: str = Field(
default=".*",
description="Regex pattern matched against ScanRecord.filename, "
"only matching records are processed",
)
videos_tsv: Optional[str] = Field(
default=None,
description="Optional path to videos.tsv audit file used to look up "
"matching video records by time range",
)
time_offset: float = Field(
default=0.0,
description="Clock offset in seconds added to acq_time values after "
"timezone normalisation to handle security functionality if any.",
)
qr: QrMode = Field(
default=QrMode.NONE,
description="QR code-based timing refinement mode. "
"Only 'none' is currently implemented; all other modes raise "
"NotImplementedError.",
)
buffer_before: str = Field(
default="0",
description="Extra video to include before scan onset. "
"Accepts seconds (e.g. '10') or ISO 8601 duration (e.g. 'PT10S').",
)
buffer_after: str = Field(
default="0",
description="Extra video to include after scan end. "
"Accepts seconds (e.g. '10') or ISO 8601 duration (e.g. 'PT10S').",
)
buffer_policy: str = Field(
default="flexible",
description="Policy for handling buffer overflow beyond video boundaries: "
"'strict' to error, 'flexible' to trim.",
)
layout: LayoutMode = Field(
default=LayoutMode.NEARBY,
description="Output file placement layout within the BIDS dataset. "
"'nearby': place output next to the NIfTI in the same datatype folder. "
"'top-stimuli': place output under stimuli/ at the BIDS root, "
"mirroring the subject/session/datatype hierarchy.",
)
overwrite: OverwriteMode = Field(
default=OverwriteMode.SKIP,
description="Policy for handling existing output files. "
"'skip': skip if output exists (default). "
"'force': remove existing file/symlink then re-inject. "
"'always': run split-video as-is without existence check. "
"'error': treat existing output as an error.",
)
reprostim_timezone: str = Field(
default="local",
description="Timezone name for naive ReproStim timestamps in videos.tsv. "
"Use 'local' for the OS timezone or an IANA name (e.g. 'America/New_York'). "
"Resolved to a tzinfo via reprostim_tz property (cached).",
)
bids_timezone: str = Field(
default="local",
description="Timezone name for naive BIDS acq_time values in _scans.tsv. "
"Use 'local' for the OS timezone or an IANA name (e.g. 'America/New_York'). "
"Resolved to a tzinfo via bids_tz property (cached).",
)
@property
def reprostim_tz(self) -> tzinfo:
"""Resolved :class:`tzinfo` for :attr:`reprostim_timezone` (cached)."""
return dt_resolve_tz(self.reprostim_timezone)
@property
def bids_tz(self) -> tzinfo:
"""Resolved :class:`tzinfo` for :attr:`bids_timezone` (cached)."""
return dt_resolve_tz(self.bids_timezone)
lock: bool = Field(
default=True,
description="When True (default), acquire the advisory file lock before "
"reading videos.tsv. When False, skip the lock (dirty-read mode) — "
"useful when the lock is owned by a different OS user.",
)
metadata_only: bool = Field(
default=False,
description="When True, refresh _scans.tsv/scans.json annotations for "
"already-generated media without touching the .mkv/sidecar .json "
"(see inject-spec.md Metadata-Only Mode). ",
)
verbose: bool = Field(
default=False,
description="When True, emit verbose progress output.",
)
out_func: Optional[Callable] = Field(
default=None,
description="Callable used for user-facing output (e.g. click.echo). "
"When None, output is suppressed.",
)
summary: BiSummary = Field(
default_factory=BiSummary,
description="Mutable run statistics accumulating injected/skipped/error counts.",
)
[docs]
class ScanRecord(BaseModel):
"""A single row from a BIDS _scans.tsv file with optionally expanded metadata."""
filename: str = Field(
..., description="Relative path to NIfTI within subject/session dir"
)
acq_time: str = Field(..., description="ISO 8601 acquisition start datetime")
duration: Optional[float] = Field(
default=None,
description="Wallclock duration of the recording in seconds, optional.",
)
kind: ScanRecordKind = Field(
default=ScanRecordKind.UNKNOWN,
description="Classification of this row's filename (NIfTI vs. ReproStim "
"media vs. other). Computed once via _calc_scan_record_kind(filename) "
"when a ScanRecord is parsed/built — not auto-recomputed if filename "
"changes afterwards.",
)
extra: dict = Field(
default_factory=dict,
description="All remaining columns from the TSV row as key/value pairs",
)
metadata: Optional[ScanMetadata] = Field(
default=None,
description="Parsed BIDS JSON sidecar metadata for this scan record.",
)
duration_sec: Optional[float] = Field(
default=None,
description="Scan duration in seconds, computed from sidecar metadata.",
)
# reprostim_* annotation columns — read from _scans.tsv when present (prior run),
# written back after successful injection.
reprostim_path: Optional[str] = Field(
default=None,
description="Path to source .mkv file relative to videos.tsv location.",
)
reprostim_offset: Optional[float] = Field(
default=None,
description="Offset of the buffer-segment start into the source video, "
"in seconds.",
)
reprostim_buffer_before: Optional[float] = Field(
default=None,
description="Actual buffer prepended before scan onset, in seconds.",
)
reprostim_buffer_after: Optional[float] = Field(
default=None,
description="Actual buffer appended after scan end, in seconds.",
)
[docs]
class ScansModel(BaseModel):
"""Parsed representation of a single BIDS ``*_scans.tsv`` file."""
path: str = Field(
..., description="Absolute or relative path to the ``*_scans.tsv`` file"
)
records: List[ScanRecord] = Field(
default_factory=list,
description="Ordered list of scan records parsed from the file",
)
orig_content: Optional[str] = Field(
None, repr=False, description="Raw content of the original ``*_scans.tsv`` file"
)
####################################################################
# Internal API
####################################################################
# Specify ordered list of basic _scans.tsv columns used by bids-inject tool
_BASIC_COLS = ["filename", "acq_time", "duration"]
# Ordered list of reprostim_* annotation columns read from and written to _scans.tsv.
_REPROSTIM_COLS = [
"reprostim_path",
"reprostim_offset",
"reprostim_buffer_before",
"reprostim_buffer_after",
]
def _parse_bids_float(value: Optional[str]) -> Optional[float]:
"""Parse an optional TSV cell as float, returning ``None`` for ``'n/a'`` or absent.
:param value: Raw cell string from a TSV row, or ``None`` when the column
was not present in the file.
:type value: Optional[str]
:returns: Parsed float, or ``None``.
:rtype: Optional[float]
"""
return float(value) if value and value != "n/a" else None
def _parse_bids_str(value: Optional[str]) -> Optional[str]:
"""Return a TSV cell string as-is, or ``None`` for ``'n/a'`` or absent.
:param value: Raw cell string from a TSV row, or ``None`` when the column
was not present in the file.
:type value: Optional[str]
:returns: The string value, or ``None``.
:rtype: Optional[str]
"""
return value if value and value != "n/a" else None
def _format_bids_str(value) -> str:
"""Serialise a field value to a TSV cell string, using ``'n/a'`` for ``None``.
:param value: Value to serialise; any type accepted by ``str()``.
:returns: String representation, or ``'n/a'`` when *value* is ``None``.
:rtype: str
"""
return "n/a" if value is None else str(value)
@contextlib.contextmanager
def _open_dataset_file(path: str, encoding: str = "utf-8", newline=None):
"""Open a BIDS text file for reading, routing through FsspecAdapter when available.
When :mod:`datalad_fuse` is installed the file is opened via
``FsspecAdapter`` so that git-annex–managed files (e.g. ``_scans.tsv``
sidecars, JSON sidecars) can be read without a local fetch. Falls back to
a plain :func:`open` call when the package is not present.
The adapter is read-only; callers that need to *write* a file (e.g.
:func:`_save_scans_model`) use a plain :func:`open` call directly.
:param path: Path to the file to open.
:type path: str
:param encoding: Text encoding, forwarded to :class:`io.TextIOWrapper` or
:func:`open`.
:type encoding: str
:param newline: Newline handling forwarded to :class:`io.TextIOWrapper` or
:func:`open`. Pass ``""`` when the caller handles line endings itself
(e.g. :mod:`csv`).
"""
f = None
raw = None
if DATALAD_FUSE_AVAILABLE:
try:
adapter = FsspecAdapter(os.path.dirname(path), caching=False)
raw = adapter.open(path)
f = io.TextIOWrapper(raw, encoding=encoding, newline=newline)
logger.debug(f"Opened via FsspecAdapter: {path}")
except Exception as e:
logger.debug(
f"FsspecAdapter failed ({e}); falling back to plain open: {path}"
)
if raw is not None:
try:
raw.close()
except Exception:
pass
raw = None
if f is None:
with open(path, encoding=encoding, newline=newline) as f:
yield f
else:
try:
yield f
finally:
f.close()
def _is_scans_file(path: str) -> bool:
"""Check if the given path points to a BIDS ``*_scans.tsv`` file.
Returns ``True`` only when *path* refers to an existing regular file
whose name ends with ``_scans.tsv``.
:param path: Path to test.
:type path: str
:returns: ``True`` if *path* is an existing ``*_scans.tsv`` file,
``False`` otherwise.
:rtype: bool
"""
return os.path.isfile(path) and path.endswith("_scans.tsv")
# Matches the Output A media filename suffix, e.g.
# "..._recording-reprostim_audiovideo.mkv". Capture group 1 ("video" / "audio"
# / "audiovideo") maps directly onto the ScanRecordKind.REPROSTIM_* values.
_MEDIA_FILENAME_RE = re.compile(r"_recording-reprostim_(video|audio|audiovideo)\.mkv$")
def _calc_scan_record_kind(filename: str) -> ScanRecordKind:
"""Classify a ``_scans.tsv`` ``filename`` value into a :class:`ScanRecordKind`.
Backs :attr:`ScanRecord.kind`. Dedicated ReproStim media rows (the
injected ``.mkv`` files, see inject-spec.md Output A/D) are excluded from
the main per-acquisition processing loop (via ``kind.is_reprostim_media``)
and are only ever touched via the insert-or-update mechanism, keyed by
this same filename.
:param filename: ``filename`` column value of a :class:`ScanRecord`.
:type filename: str
:returns: The row's classification.
:rtype: ScanRecordKind
"""
if filename.endswith(".nii.gz"):
return ScanRecordKind.NIFTI_GZ
if filename.endswith(".nii"):
return ScanRecordKind.NIFTI
m = _MEDIA_FILENAME_RE.search(filename)
if m:
return ScanRecordKind(f"reprostim_{m.group(1)}")
return ScanRecordKind.UNKNOWN
def _calc_sidecar_json(sr: ScanRecord) -> Optional[str]:
"""Derive a scan record's JSON sidecar filename from its ``filename`` and ``kind``.
Sidecar naming follows the same convention as the record's own file
extension:
- NIfTI rows (``NIFTI_GZ``/``NIFTI``) — standard BIDS convention: the
NIfTI extension (``.nii.gz`` or ``.nii``) is replaced by ``.json``
(e.g. ``*_bold.nii.gz`` -> ``*_bold.json``).
- ReproStim media rows (``REPROSTIM_VIDEO``/``REPROSTIM_AUDIO``/
``REPROSTIM_AUDIOVIDEO``) — the ``.mkv`` extension is replaced by
``.json`` (see inject-spec.md Output B).
- ``UNKNOWN`` — no sidecar naming convention is known; returns ``None``.
:param sr: Scan record whose JSON sidecar filename to derive.
:type sr: ScanRecord
:returns: Sidecar filename, relative in the same way as
:attr:`ScanRecord.filename`, or ``None`` when *sr*'s ``kind`` has no
known sidecar naming convention.
:rtype: Optional[str]
"""
filename = sr.filename
if sr.kind == ScanRecordKind.NIFTI_GZ:
return filename[: -len(".nii.gz")] + ".json"
if sr.kind == ScanRecordKind.NIFTI:
return filename[: -len(".nii")] + ".json"
if sr.kind.is_reprostim_media:
return filename[: -len(".mkv")] + ".json"
logger.warning(f"Cannot derive JSON sidecar path from: {filename}")
return None
def _parse_scan_metadata(
model: ScansModel, record: ScanRecord
) -> Optional[ScanMetadata]:
"""Parse BIDS JSON sidecar metadata for a given scan record.
Locates the JSON sidecar via :func:`_calc_sidecar_json`, resolved
relative to the directory containing the ``*_scans.tsv`` file.
The four duration-relevant keys (``FrameAcquisitionDuration``,
``AcquisitionTime``, ``RepetitionTime``, ``NumberOfVolumes``) are
mapped to typed fields on :class:`ScanMetadata`; all remaining keys are
stored verbatim in :attr:`ScanMetadata.extra`.
:param model: Parent :class:`ScansModel` whose ``path`` is used to
resolve the sidecar location.
:type model: ScansModel
:param record: Scan record whose JSON sidecar to parse.
:type record: ScanRecord
:returns: Parsed :class:`ScanMetadata` when the sidecar exists and can
be read, ``None`` otherwise.
:rtype: Optional[ScanMetadata]
"""
if not record.kind.is_nifti:
logger.warning(f"Skipping non-NIfTI scan record: {record.filename}")
return None
json_name = _calc_sidecar_json(record)
if json_name is None:
logger.warning(f"Cannot derive JSON sidecar path for: {record.filename}")
return None
scans_dir = os.path.dirname(os.path.abspath(model.path))
json_path = os.path.join(scans_dir, json_name)
if not os.path.isfile(json_path):
logger.warning(f"JSON sidecar not found: {json_path}")
return None
with _open_dataset_file(json_path) as f:
data = json.load(f)
# Per-volume AcquisitionTime array lives under time.samples, not at the
# top level (which holds only the first volume's time as a plain string).
acq_time_list: Optional[List[str]] = (
data.get("time", {}).get("samples", {}).get("AcquisitionTime")
)
known_keys = {
"TaskName",
"FrameAcquisitionDuration",
"RepetitionTime",
"NumberOfVolumes",
}
return ScanMetadata(
TaskName=data.get("TaskName"),
FrameAcquisitionDuration=data.get("FrameAcquisitionDuration"),
AcquisitionTime=acq_time_list,
RepetitionTime=data.get("RepetitionTime"),
NumberOfVolumes=data.get("NumberOfVolumes"),
extra={k: v for k, v in data.items() if k not in known_keys},
)
def _parse_scans_model(path: str) -> ScansModel:
"""Parse a ``*_scans.tsv`` file and return a :class:`ScansData` instance.
Reads a tab-separated BIDS scans file. The columns ``filename`` and
``acq_time`` are required. ``duration`` is optional and parsed as float;
absent/``'n/a'``/empty values become ``None``. Each record's ``kind`` is
computed once here, from its raw ``filename``, via
:func:`_calc_scan_record_kind`.
:param path: Absolute or relative path to a ``*_scans.tsv`` file.
:type path: str
:returns: Parsed scans data containing the file path and one
:class:`ScanRecord` per data row.
:rtype: ScansModel
:raises FileNotFoundError: If *path* does not exist.
:raises KeyError: If a required column (``filename`` or ``acq_time``) is
missing from the TSV header.
"""
_known = set(_BASIC_COLS) | set(_REPROSTIM_COLS)
records: List[ScanRecord] = []
raw_content: Optional[str] = None
# cache file content
with _open_dataset_file(path, newline="") as f:
raw_content = f.read()
reader = csv.DictReader(io.StringIO(raw_content), delimiter="\t")
for row in reader:
records.append(
ScanRecord(
filename=row["filename"],
acq_time=row["acq_time"],
duration=_parse_bids_float(row.get("duration")),
kind=_calc_scan_record_kind(row["filename"]),
extra={k: v for k, v in row.items() if k not in _known},
reprostim_path=_parse_bids_str(row.get("reprostim_path")),
reprostim_offset=_parse_bids_float(row.get("reprostim_offset")),
reprostim_buffer_before=_parse_bids_float(
row.get("reprostim_buffer_before")
),
reprostim_buffer_after=_parse_bids_float(
row.get("reprostim_buffer_after")
),
)
)
logger.debug(f"Parsed {len(records)} scan records from: {path}")
return ScansModel(path=path, records=records, orig_content=raw_content)
def _save_scans_model(model: ScansModel) -> None:
"""Write :class:`ScansModel` records back to the ``*_scans.tsv`` file.
Derives column order from the already-parsed model data: ``filename`` and
``acq_time`` first, then any extra columns (in their original order from
:attr:`ScanRecord.extra`), then ``reprostim_*`` annotation columns from
:data:`_REPROSTIM_COLS` appended on the right. ``None`` field values are
serialised as ``'n/a'`` via :func:`_format_bids_str`.
The content is serialised in memory first and compared against
:attr:`ScansModel.orig_content`; if identical, the write is skipped
entirely, so an unchanged file (including a git-annex symlink) is left
untouched. After a successful write, :attr:`ScansModel.orig_content` is
updated to the new content, so saving the same model again is a no-op.
A model with no ``orig_content`` (e.g. not loaded via
:func:`_parse_scans_model`) is always written.
:param model: Scans model to persist. :attr:`~ScansModel.path` must be
writable, or removable if it is a read-only git-annex symlink.
:type model: ScansModel
"""
# Derive extra column names from the first record (order preserved by dict).
extra_cols = list(model.records[0].extra.keys()) if model.records else []
fieldnames = (
_BASIC_COLS + extra_cols + [c for c in _REPROSTIM_COLS if c not in extra_cols]
)
rows = []
for record in model.records:
row = {
"filename": record.filename,
"acq_time": record.acq_time,
"duration": _format_bids_str(record.duration),
**record.extra,
"reprostim_path": _format_bids_str(record.reprostim_path),
"reprostim_offset": _format_bids_str(record.reprostim_offset),
"reprostim_buffer_before": _format_bids_str(record.reprostim_buffer_before),
"reprostim_buffer_after": _format_bids_str(record.reprostim_buffer_after),
}
rows.append(row)
# first serialize content to str
out_str = io.StringIO(newline="")
writer = csv.DictWriter(
out_str,
fieldnames=fieldnames,
delimiter="\t",
extrasaction="ignore",
lineterminator="\n",
)
writer.writeheader()
writer.writerows(rows)
# skip the write if content is unchanged:
new_content = out_str.getvalue()
if new_content == model.orig_content:
logger.debug(
f"Skip saving {len(rows)} scan records to: {model.path} as "
f"nothing to update"
)
return
# `_scans.tsv` can itself be annexed (a DataLad dataset may annex any file
# by .gitattributes policy, not only large binaries), in which case it is
# a read-only symlink into .git/annex/objects/ and a plain open(..., "w")
# raises PermissionError. Since this function always rewrites the entire
# file's content, remove any existing file/symlink first — os.remove()
# unlinks the symlink without touching the annex object — mirroring the
# --overwrite force remedy used for Output A/B media files.
if os.path.lexists(model.path):
os.remove(model.path)
with open(model.path, "w", newline="", encoding="utf-8") as f:
f.write(new_content)
# also update content in model after file saved
model.orig_content = new_content
logger.debug(f"Saved {len(rows)} scan records to: {model.path}")
def _calc_media_acq_time(acq_time: str, buffer_before: float) -> str:
"""Compute the dedicated media row's ``acq_time`` (see inject-spec.md Output D).
The media row's ``acq_time`` is the acquisition's own raw ``acq_time``
string (as read from the file, *before* ``--time-offset``/timezone
normalization) minus ``buffer_before`` seconds — i.e. the actual
wall-clock start of the sliced clip. Always formatted with microsecond
precision, regardless of the input string's own precision, so every
reprostim media row's ``acq_time`` has a consistent format.
:param acq_time: Raw ``acq_time`` string from the acquisition row.
:type acq_time: str
:param buffer_before: Actual buffer prepended before scan onset, in
seconds (``SplitResult.buffer_before``).
:type buffer_before: float
:returns: ISO 8601 datetime string (microsecond precision) for the media
row's ``acq_time``.
:rtype: str
"""
dt = dt_parse_bids(acq_time) - timedelta(seconds=buffer_before)
return dt.isoformat(timespec="microseconds")
def _upsert_media_row(scans: ScansModel, media_record: ScanRecord) -> None:
"""Insert or update the dedicated media row for an injected media file.
Looks up an existing record in ``scans.records`` whose ``filename``
equals ``media_record.filename`` (see inject-spec.md Output D /
"Insert-or-update (idempotency) and sort order"). If found, it is
replaced in place so re-running injection over an already-annotated file
updates rather than duplicates the row; otherwise *media_record* is
appended.
:param scans: Parsed scans model to update in place.
:type scans: ScansModel
:param media_record: Dedicated media row to insert or update with.
:type media_record: ScanRecord
"""
for i, existing in enumerate(scans.records):
if existing.filename == media_record.filename:
scans.records[i] = media_record
return
scans.records.append(media_record)
def _calc_scan_duration_sec(record: ScanRecord) -> Optional[float]:
"""Calculate scan duration in seconds from a :class:`ScanRecord`.
Resolution follows the priority order defined in the spec:
1. ``duration`` from ``_scans.tsv`` when provided.
2. ``FrameAcquisitionDuration`` (ms) — most reliable; divided by 1000.
3. ``AcquisitionTime`` array of DICOM TM strings —
``(t_last − t_first) + TR`` where ``TR = t_times[1] − t_times[0]``.
Requires at least two elements.
4. ``RepetitionTime`` (s) × ``NumberOfVolumes``.
Returns ``None`` and logs a warning when none of the four sources are
available or sufficient.
:param record: Scan record with optional metadata and/or ``duration``.
:type record: ScanRecord
:returns: Scan duration in seconds, or ``None`` if it cannot be determined.
:rtype: Optional[float]
"""
# Priority 1: duration specified in _scans.tsv
if record.duration is not None:
logger.debug(
f"Duration from scans.tsv: {record.duration:.3f} s" f" ({record.filename})"
)
return record.duration
md = record.metadata
if md is None:
logger.warning(f"No metadata available for: {record.filename}")
return None
# Priority 2: FrameAcquisitionDuration (ms → seconds)
if md.FrameAcquisitionDuration is not None:
duration = md.FrameAcquisitionDuration / 1000.0
logger.debug(
f"Duration from FrameAcquisitionDuration: {duration:.3f} s"
f" ({record.filename})"
)
return duration
# Priority 3: AcquisitionTime array — (t_last - t_first) + TR
if md.AcquisitionTime is not None and len(md.AcquisitionTime) >= 2:
times = [dt_time_to_sec(dt_parse_dicom_time(s)) for s in md.AcquisitionTime]
tr = times[1] - times[0]
duration = times[-1] - times[0] + tr
logger.debug(
f"Duration from AcquisitionTime array: {duration:.3f} s"
f" (TR={tr:.3f} s, {record.filename})"
)
return duration
# Priority 4: RepetitionTime × NumberOfVolumes
if md.RepetitionTime is not None and md.NumberOfVolumes is not None:
duration = md.RepetitionTime * md.NumberOfVolumes
logger.debug(
f"Duration from RepetitionTime × NumberOfVolumes: {duration:.3f} s"
f" ({record.filename})"
)
return duration
logger.warning(f"Cannot determine scan duration for: {record.filename}")
return None
def _calc_scan_start_end_ts(
record: ScanRecord,
time_offset: float = 0.0,
reprostim_tz: Optional[tzinfo] = None,
bids_tz: Optional[tzinfo] = None,
) -> Optional[Tuple[datetime, datetime]]:
"""Calculate scan start and end timestamps from a :class:`ScanRecord`.
Parses :attr:`ScanRecord.acq_time` as the scan start datetime, applies
timezone conversion from BIDS domain to ReproStim domain (when
*reprostim_tz* and *bids_tz* are provided), optionally applies
*time_offset*, then derives the end timestamp from
:attr:`ScanRecord.duration_sec`.
Both timestamps are returned as naive :class:`datetime.datetime` objects
in the ReproStim timezone, matching the convention used in ``videos.tsv``.
Returns ``None`` when :attr:`ScanRecord.duration_sec` is ``None``.
:param record: Scan record with ``acq_time`` and ``duration_sec`` populated.
:type record: ScanRecord
:param time_offset: Clock offset in seconds added to the start timestamp.
Defaults to ``0.0``.
:type time_offset: float
:param reprostim_tz: ReproStim capture machine timezone
(from ``--reprostim-timezone``). When ``None``, no conversion is done.
:type reprostim_tz: Optional[datetime.tzinfo]
:param bids_tz: BIDS dataset timezone (from ``--bids-timezone``).
When ``None``, no conversion is done.
:type bids_tz: Optional[datetime.tzinfo]
:returns: ``(start_ts, end_ts)`` as naive :class:`datetime.datetime` objects,
or ``None`` if the duration is unavailable.
:rtype: Optional[Tuple[datetime, datetime]]
"""
if record.duration_sec is None:
logger.warning(
f"Cannot compute scan timestamps: duration_sec is None"
f" ({record.filename})"
)
return None
start_dt = dt_parse_bids(record.acq_time)
if reprostim_tz is not None and bids_tz is not None:
start_dt = dt_bids_to_reprostim(start_dt, bids_tz, reprostim_tz)
if time_offset:
start_dt += timedelta(seconds=time_offset)
end_dt = start_dt + timedelta(seconds=record.duration_sec)
return start_dt, end_dt
def _calc_media_suffix(va) -> Optional[MediaSuffix]:
"""Determine the BEP044 recording-type suffix from a video-audit record.
Inspects ``video_res_detected`` and ``audio_sr`` fields of *va* per the
decision table in the spec:
+----------------------+----------------+---------------+
| ``video_res_detected``| ``audio_sr`` | Suffix |
+======================+================+===============+
| present | present | ``_audiovideo``|
+----------------------+----------------+---------------+
| present | absent / ``n/a``| ``_video`` |
+----------------------+----------------+---------------+
| absent / ``n/a`` | present | ``_audio`` |
+----------------------+----------------+---------------+
| absent / ``n/a`` | absent / ``n/a``| ``None`` |
+----------------------+----------------+---------------+
Returns ``None`` when neither stream is detected; the caller should skip
the injection and log a warning.
:param va: Video-audit record from ``videos.tsv``.
:returns: Appropriate :class:`MediaSuffix`, or ``None`` to skip.
:rtype: Optional[MediaSuffix]
"""
has_video = bool(va.video_res_detected and va.video_res_detected != "n/a")
has_audio = bool(va.audio_sr and va.audio_sr != "n/a")
if has_video and has_audio:
return MediaSuffix.AUDIOVIDEO
elif has_video:
return MediaSuffix.VIDEO
elif has_audio:
return MediaSuffix.AUDIO
logger.warning(
f"Cannot determine media suffix: no video or audio stream detected "
f"in {va.name!r}"
)
return None
def _find_bids_root(scans_path: str) -> str:
"""Find the BIDS dataset root directory from a ``*_scans.tsv`` path.
Walks upward from the directory containing *scans_path*, looking for a
``dataset_description.json`` file, which is required at the BIDS root.
Falls back to the parent of the first ``sub-`` path component when the
dataset description file is not found.
:param scans_path: Absolute or relative path to a ``*_scans.tsv`` file.
:type scans_path: str
:returns: Absolute path to the BIDS dataset root.
:rtype: str
"""
scans_dir = os.path.dirname(os.path.abspath(scans_path))
# Primary: walk upward looking for dataset_description.json
current = scans_dir
while True:
if os.path.isfile(os.path.join(current, "dataset_description.json")):
return current
parent = os.path.dirname(current)
if parent == current:
break
current = parent
# Fallback: parent of the first sub-XX component in the absolute path
parts = os.path.abspath(scans_dir).split(os.sep)
for i, part in enumerate(parts):
if re.match(r"^sub-", part):
return os.sep.join(parts[:i]) or os.sep
logger.warning(
f"Cannot determine BIDS root for {scans_path}; "
f"using scans file parent directory as fallback."
)
return os.path.dirname(scans_dir)
def _calc_bids_output_stem(filename: str) -> Tuple[str, str]:
"""Derive the output filename stem by stripping the BIDS datatype suffix.
Removes the ``.nii`` / ``.nii.gz`` extension, extracts any trailing
ReproIn-style suffix (double-underscore convention, e.g. ``__dup-01``),
and then strips the last BIDS suffix token (e.g. ``_bold``, ``_T1w``) —
the trailing ``_[A-Za-z][A-Za-z0-9]*`` segment — so the caller can
insert a recording-type suffix such as ``_recording-reprostim_audiovideo``
before the reproin suffix.
Examples::
# Standard BIDS name — no reproin suffix
_calc_bids_output_stem("func/sub-qa_ses-20250814_acq-faX77_bold.nii.gz")
# → ("func/sub-qa_ses-20250814_acq-faX77", "")
# ReproIn name with __dup-01 suffix
_calc_bids_output_stem(
"func/sub-qa_ses-20250814_task-rest_acq-p2_bold__dup-01.nii.gz"
)
# → ("func/sub-qa_ses-20250814_task-rest_acq-p2", "__dup-01")
The caller assembles the final output path as::
base_stem + "_recording-reprostim" + media_suffix + reproin_suffix + ".mkv"
:param filename: Relative NIfTI path from the scan record, e.g.
``func/sub-qa_ses-20250814_acq-faX77_bold.nii.gz``.
:type filename: str
:returns: ``(base_stem, reproin_suffix)`` where *base_stem* is the path
without the NIfTI extension, BIDS suffix, or reproin suffix, and
*reproin_suffix* is the extracted double-underscore token (e.g.
``"__dup-01"``) or an empty string when none is present.
:rtype: Tuple[str, str]
"""
stem = re.sub(r"\.nii(\.gz)?$", "", filename)
# Extract ReproIn-style suffix: double-underscore followed by non-whitespace
# characters at the end of the stem (e.g. "__dup-01").
reproin_suffix = ""
m = re.search(r"(__\S+)$", stem)
if m:
reproin_suffix = m.group(1)
stem = stem[: m.start()]
# Strip the BIDS suffix token (e.g. _bold, _T1w).
stem = re.sub(r"_[A-Za-z][A-Za-z0-9]*$", "", stem)
return stem, reproin_suffix
def _call_split_video(
ctx: BiContext,
scans_path: str,
record: ScanRecord,
va,
start_ts: datetime,
end_ts: datetime,
) -> Optional[ScanRecord]:
"""Invoke split-video for a single matched video record.
Resolves the input video path from the ``videos.tsv`` directory and
derives the output path from the ``*_scans.tsv`` location and scan
filename stem. Respects ``ctx.dry_run`` — when set, logs the planned
action without executing the split.
When ``ctx.metadata_only`` is set, takes a different path entirely: skips
``--overwrite`` (never writes the output file, so it's moot), errors if
the media file doesn't already exist on disk, and otherwise calls
``split-video`` with ``phantom_mode=True`` and no sidecar JSON to refresh
``_scans.tsv`` annotations without touching the ``.mkv``/sidecar
(see inject-spec.md "Metadata-Only Mode").
:param ctx: Processing context with buffer, policy, and dry-run settings.
:type ctx: BiContext
:param scans_path: Absolute path to the ``*_scans.tsv`` file being processed.
:type scans_path: str
:param record: Scan record being injected. Never mutated — the
``reprostim_*`` annotation now always lives on the dedicated media row
(see inject-spec.md Output D), not on the acquisition record.
:type record: ScanRecord
:param va: Matched video-audit record from ``videos.tsv``.
:param start_ts: Scan start timestamp (after time-offset applied).
:type start_ts: datetime
:param end_ts: Scan end timestamp.
:type end_ts: datetime
:returns: The dedicated media :class:`ScanRecord` to insert/update via
:func:`_upsert_media_row` on success, or ``None`` when nothing should
be upserted (skipped, dry-run, error, or ``--layout top-stimuli``,
which isn't supported yet — see inject-spec.md Open Questions #16).
:rtype: Optional[ScanRecord]
"""
media_suffix = _calc_media_suffix(va)
if media_suffix is None:
logger.warning(
f"Skipping injection: cannot determine media suffix ({record.filename})"
)
ctx.summary.n_skipped += 1
return
videos_dir = os.path.dirname(os.path.abspath(ctx.videos_tsv))
input_path = os.path.join(videos_dir, va.path)
scans_dir = os.path.dirname(os.path.abspath(scans_path))
base_stem, reproin_suffix = _calc_bids_output_stem(record.filename)
output_name = (
base_stem + f"_recording-reprostim{media_suffix.value}{reproin_suffix}.mkv"
)
if ctx.layout == LayoutMode.TOP_STIMULI:
bids_root = _find_bids_root(scans_path)
rel_session = os.path.relpath(scans_dir, bids_root)
output_path = os.path.join(bids_root, "stimuli", rel_session, output_name)
else: # LayoutMode.NEARBY (default)
output_path = os.path.join(scans_dir, output_name)
sidecar_path = output_path[: -len(".mkv")] + ".json"
logger.info(f"Input video path : {input_path}")
logger.info(f"Output video path : {output_path}")
logger.info(f"Sidecar JSON path : {sidecar_path}")
from reprostim.video.split import SidecarFormat
from reprostim.video.split import do_main as split_video_main
captured_errors: List[str] = []
def _capturing_out_func(msg: str) -> None:
if msg.startswith("ERROR:"):
captured_errors.append(msg)
if ctx.out_func:
ctx.out_func(msg)
sidecar_metadata: dict = {}
if ctx.metadata_only:
# --metadata-only refreshes _scans.tsv/scans.json for media that was
# already generated by an earlier real run; it never generates media
# itself. --overwrite governs whether to *write* output_path/
# sidecar_path, which never happens here, so it's bypassed entirely
# (all of skip/force/error/always are irrelevant). The existence
# check below applies only to this ReproStim media file — the one
# split-video/SplitResult produces — never to the NIfTI acquisition
# file or the sidecar JSON. See inject-spec.md "Metadata-Only Mode".
if not (os.path.exists(output_path) or os.path.islink(output_path)):
err_msg = (
f"--metadata-only: media file does not exist ({output_path}). "
"--metadata-only only refreshes annotations for already-"
"generated media; run without it first to generate this file."
)
logger.error(err_msg)
ctx.summary.errors.append(f"{record.filename}: {err_msg}")
ctx.summary.n_errors += 1
return None
else:
# Apply overwrite policy when any output file already exists
output_exists = os.path.exists(output_path) or os.path.islink(output_path)
sidecar_exists = os.path.exists(sidecar_path) or os.path.islink(sidecar_path)
if output_exists or sidecar_exists:
existing = ", ".join(
p
for p, e in [
(output_path, output_exists),
(sidecar_path, sidecar_exists),
]
if e
)
if ctx.overwrite == OverwriteMode.SKIP:
logger.info(f"Output already exists, skipping ({existing}).")
ctx.summary.n_skipped += 1
return None
elif ctx.overwrite == OverwriteMode.FORCE:
logger.info(
f"Output already exists, removing before re-inject ({existing})."
)
if output_exists:
os.remove(output_path)
if sidecar_exists:
os.remove(sidecar_path)
elif ctx.overwrite == OverwriteMode.ERROR:
err_msg = f"Output already exists ({existing})"
logger.error(err_msg)
ctx.summary.errors.append(f"{record.filename}: {err_msg}")
ctx.summary.n_errors += 1
return None
# OverwriteMode.ALWAYS: no action, fall through
if record.metadata and record.metadata.TaskName:
sidecar_metadata["TaskName"] = record.metadata.TaskName
if ctx.dry_run:
if ctx.out_func:
if ctx.metadata_only:
ctx.out_func(f"[DRY-RUN] scan (metadata-only): {record.filename}")
ctx.out_func(f" output : {output_path}")
else:
duration_sec = (end_ts - start_ts).total_seconds()
ctx.out_func(f"[DRY-RUN] scan : {record.filename}")
ctx.out_func(f" onset : {start_ts.isoformat()}")
ctx.out_func(f" duration : {duration_sec:.3f} s")
ctx.out_func(f" buf_before: {ctx.buffer_before}")
ctx.out_func(f" buf_after : {ctx.buffer_after}")
ctx.out_func(f" input : {input_path}")
ctx.out_func(f" output : {output_path}")
ctx.out_func(f" sidecar : {sidecar_path}")
dry_run_sidecar = "n/a (metadata-only)" if ctx.metadata_only else sidecar_path
logger.info(
f"Dry-run : split-video"
f" --video-audit-file {ctx.videos_tsv}"
f" --buffer-before {ctx.buffer_before}"
f" --buffer-after {ctx.buffer_after}"
f" --buffer-policy {ctx.buffer_policy}"
f" --sidecar-json {dry_run_sidecar}"
f" --start {start_ts.isoformat()}"
f" --end {end_ts.isoformat()}"
f" --lock {'yes' if ctx.lock else 'no'}"
f" --input {input_path}"
f" --output {output_path}"
)
ctx.summary.n_injected += 1
return None
if not ctx.metadata_only:
os.makedirs(os.path.dirname(output_path), exist_ok=True)
# TODO: in the future, read these fields from dedicated columns in videos.tsv
# populated by video-audit, avoiding the extra ffprobe call here.
try:
bids_properties_from_ffprobe(input_path, props=sidecar_metadata)
except Exception as e:
logger.error(f"Failed to get video info for {input_path}: {e}")
# NOTE: in --metadata-only mode, the buffer/offset/acq_time values below
# are recomputed fresh from *this* invocation's --buffer-before/
# --buffer-after/--time-offset/timezone arguments, not cross-checked
# against what's actually baked into the existing sidecar JSON. A
# mismatch between this run's arguments and the original real run's
# would silently drift the refreshed _scans.tsv annotation out of sync
# with the real file. Deliberately deferred — see inject-spec.md
# "Metadata-Only Mode" / "Known limitation — no drift protection".
ret, split_results = split_video_main(
input_path=input_path,
output_path=output_path,
start_time=start_ts.isoformat(),
end_time=end_ts.isoformat(),
buffer_before=ctx.buffer_before,
buffer_after=ctx.buffer_after,
buffer_policy=ctx.buffer_policy,
sidecar_json=None if ctx.metadata_only else sidecar_path,
sidecar_format=SidecarFormat.BIDS,
sidecar_metadata=sidecar_metadata,
video_audit_file=ctx.videos_tsv,
phantom_mode=ctx.metadata_only,
lock=ctx.lock,
verbose=ctx.verbose,
out_func=_capturing_out_func,
)
if ret != 0:
captured = "; ".join(captured_errors) if captured_errors else f"exit code {ret}"
err_msg = f"split-video failed for {record.filename}: {captured}"
logger.error(err_msg)
ctx.summary.errors.append(err_msg)
ctx.summary.n_errors += 1
return None
logger.info(f"split-video completed successfully ({record.filename})")
ctx.summary.n_injected += 1
if not split_results:
return None
if ctx.layout == LayoutMode.TOP_STIMULI:
# The media file lives outside the subject/session tree (under a
# top-level stimuli/ dir), so it has no session-relative `filename`
# yet — dedicated media-row insertion is nearby-only for now.
# See inject-spec.md Output D / Open Questions #16.
return None
split_result = split_results[0]
media_filename = os.path.relpath(output_path, scans_dir)
media_acq_time = _calc_media_acq_time(record.acq_time, split_result.buffer_before)
media_duration = (
split_result.buffer_duration
) # entire duration with pre- and post- buffer
media_extra = {k: "n/a" for k in record.extra}
if "operator" in media_extra:
media_extra["operator"] = f"reprostim:{__version__}"
return ScanRecord(
filename=media_filename,
acq_time=media_acq_time,
duration=media_duration,
# Computed explicitly here (not left to the field default) because
# _upsert_media_row may append this record onto scans.records while
# _do_inject_scans is still iterating it — Python's list iterator
# will visit the appended record later in the same loop, and it must
# already carry the correct kind so kind.is_reprostim_media routes it
# away from the acquisition-processing branch.
kind=_calc_scan_record_kind(media_filename),
extra=media_extra,
reprostim_path=va.path,
reprostim_offset=split_result.orig_buffer_offset,
reprostim_buffer_before=split_result.buffer_before,
reprostim_buffer_after=split_result.buffer_after,
)
def _do_inject_scans(ctx: BiContext, path: str):
"""Process all scan records from a single ``*_scans.tsv`` file.
Verifies that *path* is a valid BIDS scans file and delegates per-record
injection logic. Non-matching paths are logged as warnings and skipped.
Dedicated media rows (see inject-spec.md Output D) are skipped by the main
loop and only touched via :func:`_upsert_media_row`. Every acquisition
row's ``reprostim_*`` columns are unconditionally reset to ``n/a`` before
that row is (re-)evaluated, regardless of ``--match`` or outcome, which
migrates files previously annotated by the pre-#276 same-row design.
Records are sorted by ``acq_time`` before the file is rewritten.
:param ctx: Processing context carrying flags such as ``dry_run``.
:type ctx: BiContext
:param path: Path to the ``*_scans.tsv`` file to process.
:type path: str
"""
if _is_scans_file(path):
logger.info(f"Processing scans file : {path}")
if ctx.qr != QrMode.NONE:
raise NotImplementedError(
f"QR mode '{ctx.qr.value}' is not implemented yet. "
f"Only '{QrMode.NONE.value}' is currently supported."
)
scans: ScansModel = _parse_scans_model(path)
for sr in scans.records:
if sr.kind.is_reprostim_media:
# Dedicated media rows (Output D) are never treated as
# acquisitions — they have no sidecar/duration/video match of
# their own. They're only touched via _upsert_media_row below,
# driven by the acquisition row they correspond to.
logger.debug(f"Skipping media row (not an acquisition): {sr.filename}")
continue
# Unconditionally migrate/reset this acquisition row's reprostim_*
# annotation to n/a, regardless of --match or the outcome below —
# see inject-spec.md Output D / "Migration from the pre-#276 layout".
sr.reprostim_path = None
sr.reprostim_offset = None
sr.reprostim_buffer_before = None
sr.reprostim_buffer_after = None
if re.search(ctx.match, sr.filename):
ctx.summary.n_processed += 1
logger.info(f"Processing scan record : {sr}")
sr.metadata = _parse_scan_metadata(scans, sr)
logger.debug(f"Scan metadata : {sr.metadata}")
sr.duration_sec = _calc_scan_duration_sec(sr)
logger.info(f"Scan duration : {sr.duration_sec} s")
ts_pair = _calc_scan_start_end_ts(
sr, ctx.time_offset, ctx.reprostim_tz, ctx.bids_tz
)
start_ts, end_ts = ts_pair if ts_pair else (None, None)
logger.info(
f"Scan start_ts : "
f"{start_ts.isoformat() if start_ts else None}"
)
logger.info(
f"Scan end_ts : {end_ts.isoformat() if end_ts else None}"
)
# find matching video records in videos.tsv if any
if ctx.videos_tsv and start_ts and end_ts:
va_records = find_video_audit_by_timerange(
ctx.videos_tsv,
start_ts,
end_ts,
cached=True,
use_lock=ctx.lock,
)
logger.info(f"Matching video records : {len(va_records)}")
for va in va_records:
logger.info(f" : {va.name}")
logger.debug(f" : {va.model_dump()}")
# Note: only process when 1 record provided, when 0 skip and
# in case more than 1 records, log an error that multiple videos
# are not supported yet and skip as well (ambiguous match)
n_va = len(va_records)
if n_va == 0:
logger.info("No matching video records found, skipping.")
ctx.summary.n_skipped += 1
elif n_va > 1:
va_list = ", ".join(
f"{va.name} " f"[{va.start_time} -- {va.end_time}]"
for va in va_records
)
err_msg = (
f"Ambiguous match: {n_va} video records overlap "
f"the scan window [{start_ts} -- {end_ts}]. "
f"Multiple videos per scan are not yet supported; "
f"skipping. Matched videos: {va_list}."
)
logger.error(err_msg)
ctx.summary.errors.append(f"{sr.filename}: {err_msg}")
ctx.summary.n_errors += 1
else:
media_record = _call_split_video(
ctx, path, sr, va_records[0], start_ts, end_ts
)
if media_record is not None:
_upsert_media_row(scans, media_record)
else:
if not (start_ts and end_ts):
logger.warning(
f"Skipping scan record: cannot compute timestamps"
f" ({sr.filename})"
)
ctx.summary.n_skipped += 1
else:
logger.debug(f"Skipping scan record (no match): {sr.filename}")
ctx.summary.n_skipped += 1
if not ctx.dry_run:
scans.records.sort(key=lambda r: dt_parse_bids(r.acq_time))
_save_scans_model(scans)
else:
logger.warning(f"Skipping non-_scans.tsv file: {path}")
def _do_inject_dir(ctx: BiContext, path: str):
"""Process all ``*_scans.tsv`` files found directly inside a directory.
Iterates over the immediate entries of *path*. Regular files are passed
to :func:`_do_inject_scans` (which filters for ``*_scans.tsv`` names).
Subdirectories are recursed into only when ``ctx.recursive`` is ``True``.
:param ctx: Processing context; ``ctx.recursive`` controls whether
subdirectories are visited.
:type ctx: BiContext
:param path: Path to the directory to scan.
:type path: str
"""
logger.info(f"Processing scans dir : {path}")
for entry in os.scandir(path):
if entry.is_file():
if _is_scans_file(entry.path):
_do_inject_scans(ctx, entry.path)
elif entry.is_dir() and ctx.recursive:
_do_inject_dir(ctx, entry.path)
def _load_default_scans_json() -> dict:
"""Load the packaged default BIDS ``scans.json`` data-dictionary sample.
:returns: Parsed contents of ``assets/bids/scans.json``.
:rtype: dict
"""
text = (files("reprostim") / "assets" / "bids" / "scans.json").read_text()
return json.loads(text)
def _do_inject_scans_json(ctx: BiContext) -> None:
"""Ensure ``<dataset_home>/scans.json`` exists and has every default field.
Compares ``<dataset_home>/scans.json`` against the packaged default
sample (``assets/bids/scans.json``):
- If ``scans.json`` doesn't exist yet, it is created from the default
sample verbatim.
- If it exists and already has every top-level field from the default
sample, nothing is written.
- If it exists but is missing one or more default fields, those fields
(and only those — existing fields are left untouched) are appended
and the file is rewritten.
Honours ``ctx.dry_run``: when set, logs/reports what would change but
writes nothing. A ``scans.json`` that fails to parse as JSON is reported
as an error (via ``ctx.summary``) and left untouched rather than
overwritten or raised as an exception.
:param ctx: Processing context; uses ``ctx.dataset_home``, ``ctx.dry_run``,
``ctx.out_func``, and ``ctx.summary``.
:type ctx: BiContext
"""
scans_json_path = os.path.join(ctx.dataset_home, "scans.json")
default = _load_default_scans_json()
if not os.path.isfile(scans_json_path):
if ctx.dry_run:
msg = f"Would create {scans_json_path}"
else:
with open(scans_json_path, "w") as f:
json.dump(default, f, indent=2)
f.write("\n")
msg = f"Created {scans_json_path}"
logger.info(msg)
if ctx.out_func:
ctx.out_func(f"[DRY-RUN] {msg}" if ctx.dry_run else msg)
return
try:
with open(scans_json_path) as f:
existing = json.load(f)
except json.JSONDecodeError as e:
err_msg = f"Failed to parse {scans_json_path}: {e}"
logger.error(err_msg)
ctx.summary.errors.append(err_msg)
ctx.summary.n_errors += 1
if ctx.out_func:
ctx.out_func(f"ERROR: {err_msg}")
return
missing = {k: v for k, v in default.items() if k not in existing}
if not missing:
logger.debug(f"{scans_json_path} already has all default fields")
return
field_list = ", ".join(sorted(missing))
if ctx.dry_run:
msg = f"Would add missing field(s) to {scans_json_path}: {field_list}"
else:
existing.update(missing)
with open(scans_json_path, "w") as f:
json.dump(existing, f, indent=2)
f.write("\n")
msg = f"Added missing field(s) to {scans_json_path}: {field_list}"
logger.info(msg)
if ctx.out_func:
ctx.out_func(f"[DRY-RUN] {msg}" if ctx.dry_run else msg)
def _do_inject_all(ctx: BiContext, paths: List[str]):
"""Dispatch injection across a mixed list of file and directory paths.
First ensures ``<dataset_home>/scans.json`` exists and has every default
field (see :func:`_do_inject_scans_json`). Then, for each entry in
*paths*: regular files are forwarded to :func:`_do_inject_scans`;
directories are forwarded to :func:`_do_inject_dir` (which honours
``ctx.recursive``); anything else is logged as a warning and skipped.
:param ctx: Processing context propagated to all subordinate calls.
:type ctx: BiContext
:param paths: Sequence of file or directory paths supplied by the caller.
:type paths: List[str]
"""
_do_inject_scans_json(ctx)
# iterate over paths and depending on whether it's a file or directory,
# process accordingly
for path in paths:
if os.path.isfile(path):
_do_inject_scans(ctx, path)
elif os.path.isdir(path):
_do_inject_dir(ctx, path)
else:
logger.warning(f"Skipping invalid path: {path}")
####################################################################
# Datetime / Timezone Public API
####################################################################
[docs]
@lru_cache(maxsize=32)
def dt_resolve_tz(name: str) -> tzinfo:
"""Resolve a timezone name string to a :class:`datetime.tzinfo` object.
Results are cached via :func:`functools.lru_cache`, so repeated calls
with the same *name* are free after the first resolution.
:param name: ``'local'`` to use the OS system timezone, or any IANA
timezone name (e.g. ``'America/New_York'``, ``'UTC'``).
:type name: str
:returns: Corresponding :class:`tzinfo` instance.
:rtype: datetime.tzinfo
:raises ZoneInfoNotFoundError: If *name* is not a recognised IANA timezone.
"""
if name.lower() == "local":
return datetime.now().astimezone().tzinfo
return ZoneInfo(name)
[docs]
def dt_tz_label(name: str) -> str:
"""Return the current UTC offset for a timezone name, e.g. ``UTC-05:00``.
Resolves the timezone name via :func:`dt_resolve_tz` and formats the
UTC offset as ``"UTC±HH:MM"``, e.g. ``"UTC-05:00"`` or ``"UTC+00:00"``.
:param name: Timezone name accepted by :func:`dt_resolve_tz`.
:type name: str
:returns: UTC offset string.
:rtype: str
"""
tz = dt_resolve_tz(name)
offset = datetime.now(tz).utcoffset()
total_sec = int(offset.total_seconds())
sign = "+" if total_sec >= 0 else "-"
h, m = divmod(abs(total_sec), 3600)
return f"UTC{sign}{h:02d}:{m // 60:02d}"
[docs]
def dt_parse_dicom_time(value: str) -> time:
"""Parse a DICOM TM (Time) string into a :class:`datetime.time` object.
Accepts the DICOM TM format ``HHMMSS.FFFFFF`` where:
- ``HH`` — hours, 00–23
- ``MM`` — minutes, 00–59
- ``SS`` — seconds, 00–60 (60 permitted for leap seconds; clamped to 59
for :class:`datetime.time` compatibility)
- ``.FFFFFF`` — optional fractional seconds (1–6 digits, right-padded
with zeros to form microseconds)
:param value: DICOM TM string, e.g. ``'151953.397500'`` or ``'151953'``.
:type value: str
:returns: Corresponding :class:`datetime.time` instance.
:rtype: datetime.time
:raises ValueError: If *value* does not match the expected DICOM TM format.
"""
m = re.fullmatch(r"(\d{2})(\d{2})(\d{2})(?:\.(\d{1,6}))?", value.strip())
if not m:
raise ValueError(f"Invalid DICOM TM value: {value!r}")
hh, mm, ss = int(m.group(1)), int(m.group(2)), int(m.group(3))
frac = m.group(4)
microsecond = int(frac.ljust(6, "0")) if frac else 0
# datetime.time does not support leap second 60; clamp to 59
if ss == 60:
ss = 59
return time(hh, mm, ss, microsecond)
[docs]
def dt_time_to_sec(t: time) -> float:
"""Convert a :class:`datetime.time` to total seconds since midnight.
:param t: Time value to convert.
:type t: datetime.time
:returns: Total seconds since midnight, including microseconds as a
fractional part.
:rtype: float
"""
return t.hour * 3600 + t.minute * 60 + t.second + t.microsecond / 1_000_000
[docs]
def dt_parse_bids(s: str) -> datetime:
"""Parse a BIDS ``acq_time`` ISO 8601 string to a naive :class:`datetime`.
Any UTC offset present in *s* is stripped so the returned object has no
``tzinfo`` attached, matching the naive-datetime convention used throughout
this module.
:param s: ISO 8601 datetime string, e.g. ``'2025-08-14T15:06:09.742500'``.
:type s: str
:returns: Naive :class:`datetime` (no ``tzinfo``).
:rtype: datetime.datetime
:raises ValueError: If *s* cannot be parsed as an ISO 8601 datetime.
"""
return datetime.fromisoformat(s).replace(tzinfo=None)
[docs]
def dt_convert(dt: datetime, tz_from: tzinfo, tz_to: tzinfo) -> datetime:
"""Convert a naive datetime from one timezone context to another.
Attaches *tz_from* to *dt*, converts to *tz_to*, then strips ``tzinfo``
to return a naive datetime. This is the primitive that all higher-level
helpers delegate to.
:param dt: Naive source datetime (no ``tzinfo``).
:type dt: datetime.datetime
:param tz_from: Timezone the naive *dt* implicitly represents.
:type tz_from: datetime.tzinfo
:param tz_to: Target timezone to convert into.
:type tz_to: datetime.tzinfo
:returns: Naive datetime in *tz_to*.
:rtype: datetime.datetime
"""
return dt.replace(tzinfo=tz_from).astimezone(tz_to).replace(tzinfo=None)
[docs]
def dt_reprostim_to_utc(dt: datetime, tz: tzinfo) -> datetime:
"""Convert a naive ReproStim datetime to a naive UTC datetime.
:param dt: Naive ReproStim datetime from ``videos.tsv``.
:type dt: datetime.datetime
:param tz: Timezone the ReproStim capture machine was using.
:type tz: datetime.tzinfo
:returns: Naive UTC datetime.
:rtype: datetime.datetime
"""
return dt_convert(dt, tz, timezone.utc)
[docs]
def dt_bids_to_utc(dt: datetime, tz: tzinfo) -> datetime:
"""Convert a naive BIDS datetime to a naive UTC datetime.
:param dt: Naive BIDS ``acq_time`` datetime.
:type dt: datetime.datetime
:param tz: Timezone assumed for the BIDS timestamps.
:type tz: datetime.tzinfo
:returns: Naive UTC datetime.
:rtype: datetime.datetime
"""
return dt_convert(dt, tz, timezone.utc)
[docs]
def dt_utc_to_reprostim(dt: datetime, tz: tzinfo) -> datetime:
"""Convert a naive UTC datetime to a naive ReproStim-domain datetime.
:param dt: Naive UTC datetime.
:type dt: datetime.datetime
:param tz: Target timezone (ReproStim capture machine timezone).
:type tz: datetime.tzinfo
:returns: Naive datetime in *tz*.
:rtype: datetime.datetime
"""
return dt_convert(dt, timezone.utc, tz)
[docs]
def dt_utc_to_bids(dt: datetime, tz: tzinfo) -> datetime:
"""Convert a naive UTC datetime to a naive BIDS-domain datetime.
:param dt: Naive UTC datetime.
:type dt: datetime.datetime
:param tz: Target timezone (BIDS dataset timezone).
:type tz: datetime.tzinfo
:returns: Naive datetime in *tz*.
:rtype: datetime.datetime
"""
return dt_convert(dt, timezone.utc, tz)
[docs]
def dt_reprostim_to_bids(
dt: datetime, reprostim_tz: tzinfo, bids_tz: tzinfo
) -> datetime:
"""Convert a naive ReproStim datetime directly to a naive BIDS datetime.
Routes through UTC internally:
``dt_utc_to_bids(dt_reprostim_to_utc(dt, reprostim_tz), bids_tz)``.
:param dt: Naive ReproStim datetime.
:type dt: datetime.datetime
:param reprostim_tz: ReproStim capture machine timezone.
:type reprostim_tz: datetime.tzinfo
:param bids_tz: BIDS dataset timezone.
:type bids_tz: datetime.tzinfo
:returns: Naive datetime in the BIDS timezone.
:rtype: datetime.datetime
"""
return dt_utc_to_bids(dt_reprostim_to_utc(dt, reprostim_tz), bids_tz)
[docs]
def dt_bids_to_reprostim(
dt: datetime, bids_tz: tzinfo, reprostim_tz: tzinfo
) -> datetime:
"""Convert a naive BIDS datetime directly to a naive ReproStim datetime.
Routes through UTC internally:
``dt_utc_to_reprostim(dt_bids_to_utc(dt, bids_tz), reprostim_tz)``.
:param dt: Naive BIDS ``acq_time`` datetime.
:type dt: datetime.datetime
:param bids_tz: BIDS dataset timezone.
:type bids_tz: datetime.tzinfo
:param reprostim_tz: ReproStim capture machine timezone.
:type reprostim_tz: datetime.tzinfo
:returns: Naive datetime in the ReproStim timezone.
:rtype: datetime.datetime
"""
return dt_utc_to_reprostim(dt_bids_to_utc(dt, bids_tz), reprostim_tz)
####################################################################
# Public API
####################################################################
[docs]
def do_main(
paths: List[str],
videos_tsv: str,
dataset_home: str,
recursive: bool,
match: str,
buffer_before: str,
buffer_after: str,
buffer_policy: str,
time_offset: float,
qr: str,
layout: str,
reprostim_timezone: str,
bids_timezone: str,
dry_run: bool,
overwrite: str,
lock: bool,
verbose: bool,
out_func: Callable,
metadata_only: bool = False,
) -> int:
"""Main entry point for the bids-inject command.
Orchestrates loading of videos.tsv, discovery of ``*_scans.tsv`` files,
per-scan matching, slicing, and injection.
:param paths: List of file or directory paths from the CLI ``PATHS`` argument.
:type paths: List[str]
:param videos_tsv: Path to ``videos.tsv`` produced by ``video-audit``.
Video file paths inside the TSV are resolved relative to this file's location.
:type videos_tsv: str
:param dataset_home: Home directory of the BIDS dataset being injected into
(e.g. contains ``scans.json``). Defaults to the current directory.
:type dataset_home: str
:param recursive: When ``True``, recurse into subdirectories when searching
for ``*_scans.tsv`` files.
:type recursive: bool
:param match: Regular expression matched against the ``filename`` field of
each :class:`ScanRecord`. Only matching records are processed; others
are skipped. Default ``'.*'`` matches every record.
:type match: str
:param buffer_before: Extra video to include before scan onset.
Accepts seconds (e.g. ``'10'``) or ISO 8601 duration (e.g. ``'PT10S'``).
:type buffer_before: str
:param buffer_after: Extra video to include after scan end.
Accepts seconds (e.g. ``'10'``) or ISO 8601 duration (e.g. ``'PT10S'``).
:type buffer_after: str
:param buffer_policy: ``'strict'`` to error when buffers exceed video
boundaries; ``'flexible'`` to trim them instead.
:type buffer_policy: str
:param time_offset: Clock offset in seconds added to ``acq_time`` values
after timezone normalisation.
:type time_offset: float
:param qr: QR timing refinement mode: ``'none'``, ``'auto'``,
``'embed-existing'``, or ``'parse'``.
:type qr: str
:param layout: Output placement layout: ``'nearby'`` or ``'top-stimuli'``.
:type layout: str
:param reprostim_timezone: Timezone for naive ReproStim timestamps in
``videos.tsv``. Use ``'local'`` for the OS timezone or an IANA name
(e.g. ``'America/New_York'``).
:type reprostim_timezone: str
:param bids_timezone: Timezone for naive BIDS ``acq_time`` values.
Use ``'local'`` or an IANA name.
:type bids_timezone: str
:param dry_run: When ``True``, resolve matches and print planned actions
but skip ``split-video`` and all file writes.
:type dry_run: bool
:param lock: When ``True`` (default), acquire the advisory file lock
before reading ``videos.tsv``. When ``False``, skip the lock
(dirty-read mode) — useful when the lock file is owned by a
different OS user.
:type lock: bool
:param layout: Output file placement layout: ``'nearby'`` places the output
next to the NIfTI in the same BIDS datatype folder; ``'top-stimuli'``
places it under a ``stimuli/`` directory at the BIDS root, mirroring
the subject/session/datatype hierarchy.
:type layout: str
:param overwrite: Policy for existing output files: ``'skip'`` (default)
skips the scan, ``'force'`` removes existing file/symlink then
re-injects, ``'always'`` runs split-video as-is without an existence
check, ``'error'`` treats existing output as an error.
:type overwrite: str
:param verbose: When ``True``, emit verbose progress output.
:type verbose: bool
:param out_func: Callable used for user-facing output (e.g. ``click.echo``).
:type out_func: Callable
:param metadata_only: When ``True``, refresh ``_scans.tsv``/``scans.json``
annotations for already-generated media without touching the
``.mkv``/sidecar ``.json`` (see inject-spec.md Metadata-Only Mode).
:type metadata_only: bool
:returns: Exit code — ``0`` on success, non-zero on error.
:rtype: int
"""
if not os.path.isdir(dataset_home):
err_msg = (
f"--dataset path does not exist or is not a directory: {dataset_home!r}"
)
logger.error(err_msg)
if out_func:
out_func(f"ERROR: {err_msg}")
return 1
ctx: BiContext = BiContext(
dataset_home=dataset_home,
dry_run=dry_run,
recursive=recursive,
match=match,
videos_tsv=videos_tsv,
time_offset=time_offset,
qr=QrMode(qr),
buffer_before=buffer_before,
buffer_after=buffer_after,
buffer_policy=buffer_policy,
layout=LayoutMode(layout),
overwrite=OverwriteMode(overwrite),
reprostim_timezone=reprostim_timezone,
bids_timezone=bids_timezone,
lock=lock,
verbose=verbose,
out_func=out_func,
metadata_only=metadata_only,
)
_do_inject_all(ctx, paths)
s = ctx.summary
prefix = "[DRY-RUN] " if dry_run else ""
summary_line = (
f"{prefix}{s.n_processed} processed, "
f"{s.n_injected} injected, "
f"{s.n_skipped} skipped, "
f"{s.n_errors} errors"
)
logger.debug(summary_line)
if out_func:
out_func(summary_line)
if verbose and s.errors:
out_func("Errors:")
for i, err in enumerate(s.errors, 1):
out_func(f" [{i}] {err}")
return 1 if s.n_errors > 0 else 0
