Skip to content

geotrax.utils.file_utils

Where each stage writes its results, how they are written safely, and small helpers for location IDs, orthophoto folders and videos. The result files themselves are described in the extract and georeference guides.

Result paths

build_result_path

build_result_path(source: Path, result_type: str, output_cfg: dict | None = None, viz_mode: int | None = None, ext: str | None = None) -> Path | None

Return the path where a stage writes one kind of result for an input video.

Parameters:

Name Type Description Default
source Path

Input video.

required
result_type str

One of 'video' (the input itself), 'processed' (tracks .txt), 'video_transformations' (stabilization transforms), 'geo_transformations' (video-to-orthophoto homography), 'georeferenced' (trajectory .csv), 'visualized' (annotated video) and 'metadata' (run-metadata .yaml).

required
output_cfg dict | None

The output config section, which sets the folder and the filename postfixes; None uses DEFAULT_OUTPUT.

None
viz_mode int | None

Visualization mode, for 'visualized' only.

None
ext str | None

Video file extension without the dot, for 'visualized' only.

None

Returns:

Type Description
Path | None

The path, which need not exist yet, or None for an unknown result_type.

Examples:

>>> build_result_path(Path("D1/A1.mp4"), "georeferenced")
PosixPath('D1/results/A1.csv')
>>> build_result_path(Path("D1/A1.mp4"), "visualized", viz_mode=2, ext="mp4")
PosixPath('D1/results/A1_mode_2.mp4')
Source code in geotrax/utils/file_utils.py
def build_result_path(
    source: Path,
    result_type: str,
    output_cfg: dict | None = None,
    viz_mode: int | None = None,
    ext: str | None = None,
) -> Path | None:
    """Return the path where a stage writes one kind of result for an input video.

    Args:
        source: Input video.
        result_type: One of `'video'` (the input itself), `'processed'` (tracks `.txt`),
            `'video_transformations'` (stabilization transforms), `'geo_transformations'`
            (video-to-orthophoto homography), `'georeferenced'` (trajectory `.csv`),
            `'visualized'` (annotated video) and `'metadata'` (run-metadata `.yaml`).
        output_cfg: The `output` config section, which sets the folder and the filename postfixes;
            `None` uses [`DEFAULT_OUTPUT`][geotrax.utils.file_utils.DEFAULT_OUTPUT].
        viz_mode: Visualization mode, for `'visualized'` only.
        ext: Video file extension without the dot, for `'visualized'` only.

    Returns:
        The path, which need not exist yet, or `None` for an unknown `result_type`.

    Examples:
        ```python
        >>> build_result_path(Path("D1/A1.mp4"), "georeferenced")
        PosixPath('D1/results/A1.csv')
        >>> build_result_path(Path("D1/A1.mp4"), "visualized", viz_mode=2, ext="mp4")
        PosixPath('D1/results/A1_mode_2.mp4')
        ```
    """
    if result_type == 'video':
        return source
    cfg = output_cfg or DEFAULT_OUTPUT
    out_dir = get_output_dir(source, cfg)
    stem = source.stem
    if result_type == 'processed':
        return out_dir / f"{stem}{cfg.get('tracks_postfix', DEFAULT_OUTPUT['tracks_postfix'])}.txt"
    if result_type == 'video_transformations':
        return out_dir / f"{stem}{cfg.get('stab_transform_postfix', DEFAULT_OUTPUT['stab_transform_postfix'])}.txt"
    if result_type == 'geo_transformations':
        return out_dir / f"{stem}{cfg.get('geo_transform_postfix', DEFAULT_OUTPUT['geo_transform_postfix'])}.txt"
    if result_type == 'georeferenced':
        return out_dir / f"{stem}{cfg.get('georeferenced_postfix', DEFAULT_OUTPUT['georeferenced_postfix'])}.csv"
    if result_type == 'visualized':
        return out_dir / f"{stem}{cfg.get('visualization_postfix', DEFAULT_OUTPUT['visualization_postfix'])}_mode_{viz_mode}.{ext}"
    if result_type == 'metadata':
        return out_dir / f"{stem}{cfg.get('metadata_postfix', DEFAULT_OUTPUT['metadata_postfix'])}.yaml"
    return None

check_if_results_exist

check_if_results_exist(file: Path, result_type: str, viz_mode: int | None = None, ext: str | None = None, output_cfg: dict | None = None) -> tuple[bool, Path | None]

Check whether a result of an input video already exists; batch skips the stage if so.

Parameters:

Name Type Description Default
file Path

Input video.

required
result_type str

A result type accepted by build_result_path.

required
viz_mode int | None

Visualization mode, for 'visualized' only.

None
ext str | None

Video file extension, for 'visualized' only.

None
output_cfg dict | None

The output config section; None uses the defaults.

None

Returns:

Type Description
tuple[bool, Path | None]

(exists, path); path is None for an unknown result_type.

Source code in geotrax/utils/file_utils.py
def check_if_results_exist(
    file: Path,
    result_type: str,
    viz_mode: int | None = None,
    ext: str | None = None,
    output_cfg: dict | None = None,
) -> tuple[bool, Path | None]:
    """Check whether a result of an input video already exists; `batch` skips the stage if so.

    Args:
        file: Input video.
        result_type: A result type accepted by
            [`build_result_path`][geotrax.utils.file_utils.build_result_path].
        viz_mode: Visualization mode, for `'visualized'` only.
        ext: Video file extension, for `'visualized'` only.
        output_cfg: The `output` config section; `None` uses the defaults.

    Returns:
        `(exists, path)`; `path` is `None` for an unknown `result_type`.
    """
    result_path = build_result_path(file, result_type, output_cfg, viz_mode, ext)
    return (result_path.exists() if result_path else False), result_path

get_output_dir

get_output_dir(source: Path, output_cfg: dict | None = None) -> Path

Return the output directory for an input file.

Parameters:

Name Type Description Default
source Path

Input video (or any file whose results are looked up).

required
output_cfg dict | None

The output config section; None uses DEFAULT_OUTPUT.

None

Returns:

Type Description
Path

output.folder itself when it is absolute (one folder shared by every input), otherwise that name inside the input's own directory.

Source code in geotrax/utils/file_utils.py
def get_output_dir(source: Path, output_cfg: dict | None = None) -> Path:
    """Return the output directory for an input file.

    Args:
        source: Input video (or any file whose results are looked up).
        output_cfg: The `output` config section; `None` uses
            [`DEFAULT_OUTPUT`][geotrax.utils.file_utils.DEFAULT_OUTPUT].

    Returns:
        `output.folder` itself when it is absolute (one folder shared by every input), otherwise
            that name inside the input's own directory.
    """
    cfg = output_cfg or DEFAULT_OUTPUT
    folder = Path(cfg.get('folder', DEFAULT_OUTPUT['folder']))
    return folder if folder.is_absolute() else source.parent / folder

output_folder_exclusion

output_folder_exclusion(output_cfg: dict | None = None) -> tuple[str | None, Path | None]

Return how a directory scan should skip the configured output folder.

A relative folder is recreated next to every input, so it is skipped by name. An absolute folder is one shared directory, so it is skipped by its resolved path only: skipping its name as well would also drop every unrelated input folder that shares it, e.g. -of /mnt/out/D1 dropping every D1 drone folder.

Parameters:

Name Type Description Default
output_cfg dict | None

The output config section; None uses the default results.

None

Returns:

Type Description
tuple[str | None, Path | None]

(name, None) for a relative folder, (None, resolved_path) for an absolute one.

Source code in geotrax/utils/file_utils.py
def output_folder_exclusion(output_cfg: dict | None = None) -> tuple[str | None, Path | None]:
    """Return how a directory scan should skip the configured output folder.

    A relative folder is recreated next to every input, so it is skipped by name. An absolute
    folder is one shared directory, so it is skipped by its resolved path only: skipping its name as
    well would also drop every unrelated input folder that shares it, e.g. `-of /mnt/out/D1`
    dropping every `D1` drone folder.

    Args:
        output_cfg: The `output` config section; `None` uses the default `results`.

    Returns:
        `(name, None)` for a relative folder, `(None, resolved_path)` for an absolute one.
    """
    cfg = output_cfg or DEFAULT_OUTPUT
    folder = Path(cfg.get('folder', DEFAULT_OUTPUT['folder']))
    if folder.is_absolute():
        return None, folder.resolve()
    return folder.name or None, None

read_recorded_stab_anchor

read_recorded_stab_anchor(source: Path, output_cfg: dict | None = None) -> int | None

Return the stabilization anchor frame that extract recorded for an input video.

Stabilized track coordinates are relative to the processing.cut_frame_left that extract ran with, which it records in the run-metadata YAML. Stages that run separately (georeference, visualize) must use that frame rather than their own cut_frame_left.

Parameters:

Name Type Description Default
source Path

Input video.

required
output_cfg dict | None

The output config section, used to locate the metadata file.

None

Returns:

Type Description
int | None

The recorded frame index, or None when the metadata file is missing or unreadable (e.g. results from before v1.4.0) or holds no integer anchor; the fallback is up to the caller.

Source code in geotrax/utils/file_utils.py
def read_recorded_stab_anchor(source: Path, output_cfg: dict | None = None) -> int | None:
    """Return the stabilization anchor frame that `extract` recorded for an input video.

    Stabilized track coordinates are relative to the `processing.cut_frame_left` that `extract` ran
    with, which it records in the run-metadata YAML. Stages that run separately (`georeference`,
    `visualize`) must use that frame rather than their own `cut_frame_left`.

    Args:
        source: Input video.
        output_cfg: The `output` config section, used to locate the metadata file.

    Returns:
        The recorded frame index, or `None` when the metadata file is missing or unreadable (e.g.
            results from before v1.4.0) or holds no integer anchor; the fallback is up to the caller.
    """
    processing = _read_run_metadata(source, output_cfg).get('processing')
    anchor = processing.get('cut_frame_left') if isinstance(processing, dict) else None
    if not isinstance(anchor, int) or isinstance(anchor, bool):
        return None
    return anchor

no_vehicles_detected

no_vehicles_detected(source: Path, output_cfg: dict | None = None) -> bool

Tell whether extract finished on an input video without finding any vehicle.

Such a run writes no tracks file, the file whose presence marks a finished extraction, so its absence alone cannot separate a video in which nothing was detected from one that was never extracted. The run-metadata YAML settles it: extract records the number of saved vehicles there, and writes it even when there are none. batch uses this to count such a video as done instead of failed, and georeference and visualize to stop without an error.

Parameters:

Name Type Description Default
source Path

Input video.

required
output_cfg dict | None

The output config section, used to locate the result files.

None

Returns:

Type Description
bool

True when the metadata records zero vehicles and no tracks file exists; False when there are tracks, or when the metadata is missing, unreadable or from before v1.6.0.

Source code in geotrax/utils/file_utils.py
def no_vehicles_detected(source: Path, output_cfg: dict | None = None) -> bool:
    """Tell whether `extract` finished on an input video without finding any vehicle.

    Such a run writes no tracks file, the file whose presence marks a finished extraction, so its
    absence alone cannot separate a video in which nothing was detected from one that was never
    extracted. The run-metadata YAML settles it: `extract` records the number of saved vehicles
    there, and writes it even when there are none. `batch` uses this to count such a video as done
    instead of failed, and `georeference` and `visualize` to stop without an error.

    Args:
        source: Input video.
        output_cfg: The `output` config section, used to locate the result files.

    Returns:
        `True` when the metadata records zero vehicles and no tracks file exists; `False` when there
            are tracks, or when the metadata is missing, unreadable or from before v1.6.0.
    """
    if check_if_results_exist(source, 'processed', output_cfg=output_cfg)[0]:
        return False
    results = _read_run_metadata(source, output_cfg).get('results')
    vehicles = results.get('vehicles') if isinstance(results, dict) else None
    return isinstance(vehicles, int) and not isinstance(vehicles, bool) and vehicles == 0

DEFAULT_OUTPUT module-attribute

DEFAULT_OUTPUT = {'folder': 'results', 'tracks_postfix': '', 'georeferenced_postfix': '', 'stab_transform_postfix': '_vid_transf', 'geo_transform_postfix': '_geo_transf', 'visualization_postfix': '', 'metadata_postfix': ''}

Fallback output section, reproducing the historical results/ layout.

Used only when no config is passed to the path helpers (output_cfg=None), or for a key missing from a custom config. The pipeline stages always pass the loaded output section.

Writing files

atomic_output

atomic_output(path: Path, keep_suffix: bool = False) -> Iterator[Path]

Write a file atomically: yield a temporary path that replaces path only if the block completes.

batch treats any existing result as complete, so an interrupted write (Ctrl+C, a SLURM kill, a full disk) must not leave a truncated file behind. Inside this context, path holds either its previous content or the complete new one, and the temporary file is always removed.

Parameters:

Name Type Description Default
path Path

Final destination.

required
keep_suffix bool

Name the temporary file .name.tmp.ext instead of .name.ext.tmp, for writers such as GDAL that check the extension. The default suffix keeps the file out of every *.csv/*.txt result scan; use keep_suffix=True only for formats no scan globs.

False

Yields:

Type Description
Path

The temporary path to write to, in the same directory as path.

Examples:

with atomic_output(csv_path) as tmp:
    df.to_csv(tmp, index=False)
Source code in geotrax/utils/file_utils.py
@contextmanager
def atomic_output(path: Path, keep_suffix: bool = False) -> Iterator[Path]:
    """Write a file atomically: yield a temporary path that replaces `path` only if the block completes.

    `batch` treats any existing result as complete, so an interrupted write (Ctrl+C, a SLURM kill, a
    full disk) must not leave a truncated file behind. Inside this context, `path` holds either its
    previous content or the complete new one, and the temporary file is always removed.

    Args:
        path: Final destination.
        keep_suffix: Name the temporary file `.name.tmp.ext` instead of `.name.ext.tmp`, for writers
            such as GDAL that check the extension. The default suffix keeps the file out of every
            `*.csv`/`*.txt` result scan; use `keep_suffix=True` only for formats no scan globs.

    Yields:
        The temporary path to write to, in the same directory as `path`.

    Examples:
        ```python
        with atomic_output(csv_path) as tmp:
            df.to_csv(tmp, index=False)
        ```
    """
    tmp_path = path.with_name(f'.{path.stem}.tmp{path.suffix}' if keep_suffix else f'.{path.name}.tmp')
    try:
        yield tmp_path
        os.replace(tmp_path, path)
    finally:
        tmp_path.unlink(missing_ok=True)

convert_to_serializable

convert_to_serializable(obj: Any) -> Any

Recursively convert paths and argparse namespaces into YAML/JSON-serializable values.

Parameters:

Name Type Description Default
obj Any

A value, possibly nesting dicts, lists, Paths and argparse.Namespaces.

required

Returns:

Type Description
Any

obj with every Path turned into a string and every namespace into a dict.

Source code in geotrax/utils/file_utils.py
def convert_to_serializable(obj: Any) -> Any:
    """Recursively convert paths and argparse namespaces into YAML/JSON-serializable values.

    Args:
        obj: A value, possibly nesting dicts, lists, `Path`s and `argparse.Namespace`s.

    Returns:
        `obj` with every `Path` turned into a string and every namespace into a dict.
    """
    if isinstance(obj, Path):
        return str(obj)
    elif isinstance(obj, argparse.Namespace):
        return {k: convert_to_serializable(v) for k, v in vars(obj).items()}
    elif isinstance(obj, dict):
        return {k: convert_to_serializable(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [convert_to_serializable(i) for i in obj]
    else:
        return obj

detect_delimiter

detect_delimiter(filepath: Path, lines_to_check: int = 5) -> str

Guess whether a text file is comma, space or tab separated.

Parameters:

Name Type Description Default
filepath Path

File to inspect.

required
lines_to_check int

Number of leading lines to count delimiters in.

5

Returns:

Type Description
str

The most frequent of ',', ' ' and '\t' in those lines.

Source code in geotrax/utils/file_utils.py
def detect_delimiter(filepath: Path, lines_to_check: int = 5) -> str:
    """Guess whether a text file is comma, space or tab separated.

    Args:
        filepath: File to inspect.
        lines_to_check: Number of leading lines to count delimiters in.

    Returns:
        The most frequent of `','`, `' '` and `'\\t'` in those lines.
    """
    delimiters = {',': 0, ' ': 0, '\t': 0}
    with open(filepath) as file:
        for _ in range(lines_to_check):
            line = file.readline()
            if not line:
                break
            delimiters[','] += line.count(',')
            delimiters[' '] += line.count(' ')
            delimiters['\t'] += line.count('\t')
    return max(delimiters, key=lambda k: delimiters[k])

Locations

determine_location_id

determine_location_id(source: Path, logger: Logger = None) -> str

Extract the location ID from a filename: its first run of letters.

The location ID selects the orthophoto, segmentation and master frame of a recording site. A digit, _ or - ends the run.

Parameters:

Name Type Description Default
source Path

Video or result file named after its location.

required
logger Logger

Receives the detected ID; without one, an error is printed instead.

None

Returns:

Type Description
str

The location ID.

Raises:

Type Description
SystemExit

The filename contains no letters.

Examples:

>>> determine_location_id(Path("A1.mp4"))
'A'
>>> determine_location_id(Path("2025-01-01_A_PM1.mp4"))
'A'
>>> determine_location_id(Path("A1_AV.csv"))
'A'
Source code in geotrax/utils/file_utils.py
def determine_location_id(source: Path, logger: logging.Logger = None) -> str:
    """Extract the location ID from a filename: its first run of letters.

    The location ID selects the orthophoto, segmentation and master frame of a recording site. A
    digit, `_` or `-` ends the run.

    Args:
        source: Video or result file named after its location.
        logger: Receives the detected ID; without one, an error is printed instead.

    Returns:
        The location ID.

    Raises:
        SystemExit: The filename contains no letters.

    Examples:
        ```python
        >>> determine_location_id(Path("A1.mp4"))
        'A'
        >>> determine_location_id(Path("2025-01-01_A_PM1.mp4"))
        'A'
        >>> determine_location_id(Path("A1_AV.csv"))
        'A'
        ```
    """
    location_id = []
    for char in source.stem:
        if char.isalpha():
            location_id.append(char)
        elif len(location_id) and (char in '_-' or char.isdigit()):
            break
    location_id = ''.join(location_id)

    if not location_id:
        message = f"Error: Failed to extract location ID from the source filename {source}."
        if logger:
            logger.error(message)
        else:
            print(message)
        sys.exit(1)

    if logger:
        logger.info(f"Detected location ID: '{location_id}' from the source filename {source.name}.")

    return location_id

get_ortho_folder

get_ortho_folder(source: Path, ortho_folder: Path | None, logger: Logger, critical: bool = True) -> Path

Return the orthophoto folder for an input, given explicitly or found from the project layout.

Without an explicit folder, the nearest ancestor of source named PROCESSED or DATASET is looked up, and its sibling ORTHOPHOTOS folder is used.

Parameters:

Name Type Description Default
source Path

Input video or result file.

required
ortho_folder Path | None

Explicit folder (--ortho-folder), or None to use the project layout.

required
logger Logger

Receives the outcome.

required
critical bool

Exit on failure instead of returning None.

True

Returns:

Type Description
Path

The orthophoto folder, or None when it cannot be found and critical is false.

Raises:

Type Description
SystemExit

The folder cannot be found and critical is true.

Source code in geotrax/utils/file_utils.py
def get_ortho_folder(source: Path, ortho_folder: Path | None, logger: logging.Logger, critical: bool = True) -> Path:
    """Return the orthophoto folder for an input, given explicitly or found from the project layout.

    Without an explicit folder, the nearest ancestor of `source` named `PROCESSED` or `DATASET` is
    looked up, and its sibling `ORTHOPHOTOS` folder is used.

    Args:
        source: Input video or result file.
        ortho_folder: Explicit folder (`--ortho-folder`), or `None` to use the project layout.
        logger: Receives the outcome.
        critical: Exit on failure instead of returning `None`.

    Returns:
        The orthophoto folder, or `None` when it cannot be found and `critical` is false.

    Raises:
        SystemExit: The folder cannot be found and `critical` is true.
    """
    if ortho_folder is None:
        ortho_folder = source.parent

        while ortho_folder != ortho_folder.parent:
            if ortho_folder.name in ['PROCESSED', 'DATASET']:
                break
            ortho_folder = ortho_folder.parent

        if ortho_folder.name not in ['PROCESSED', 'DATASET']:
            if critical:
                logger.critical(
                    f"Failed to find the orthophoto folder for source '{source}'. "
                    f"Please either provide a custom path using the --ortho-folder argument, "
                    f"skip georeferencing with the --no-geo argument, "
                    f"or ensure that the default folder structure is in place."
                )
                sys.exit(1)
            else:
                logger.info(
                    f"Failed to find the orthophoto folder for source '{source}'. "
                    f"Please either provide a custom path using the --ortho-folder argument, "
                    f"skip georeferencing with the --no-geo argument, "
                    f"or ensure that the default folder structure is in place."
                )
                return None

        ortho_folder = ortho_folder.parent / 'ORTHOPHOTOS'

    if not ortho_folder.exists():
        if critical:
            logger.critical(f"Orthophoto folder '{ortho_folder}' not found. Use the '--ortho-folder' argument to provide a custom path or ensure the default folder structure.")
            sys.exit(1)
        else:
            logger.info(f"Orthophoto folder '{ortho_folder}' not found. Use the '--ortho-folder' argument to provide a custom path or ensure the default folder structure.")
            return None
    else:
        logger.info(f"Using orthophoto folder: '{ortho_folder}'.")

    return ortho_folder

Videos

get_video_dimensions

get_video_dimensions(video_path: Path) -> tuple[int, int]

Return the frame size of a video.

Parameters:

Name Type Description Default
video_path Path

Video file.

required

Returns:

Type Description
tuple[int, int]

(width, height) in pixels; (0, 0) if OpenCV cannot open the file.

Source code in geotrax/utils/file_utils.py
def get_video_dimensions(video_path: Path) -> tuple[int, int]:
    """Return the frame size of a video.

    Args:
        video_path: Video file.

    Returns:
        `(width, height)` in pixels; `(0, 0)` if OpenCV cannot open the file.
    """
    reader = cv2.VideoCapture(str(video_path))
    frame_w = int(reader.get(cv2.CAP_PROP_FRAME_WIDTH))
    frame_h = int(reader.get(cv2.CAP_PROP_FRAME_HEIGHT))
    reader.release()
    return frame_w, frame_h

get_keyframe_times

get_keyframe_times(video_path: Path) -> ndarray

Return the presentation times of a video's keyframes.

Reads ffprobe's per-packet pts_time,flags output and keeps the packets flagged K. ffprobe is run with an argument list rather than through a shell, so any path works on every platform.

Parameters:

Name Type Description Default
video_path Path

Video file; ffprobe must be on the PATH.

required

Returns:

Type Description
ndarray

Keyframe times in seconds, in stream order.

Raises:

Type Description
CalledProcessError

ffprobe failed.

Source code in geotrax/utils/file_utils.py
def get_keyframe_times(video_path: Path) -> np.ndarray:
    """Return the presentation times of a video's keyframes.

    Reads ffprobe's per-packet `pts_time,flags` output and keeps the packets flagged `K`. ffprobe is
    run with an argument list rather than through a shell, so any path works on every platform.

    Args:
        video_path: Video file; `ffprobe` must be on the `PATH`.

    Returns:
        Keyframe times in seconds, in stream order.

    Raises:
        subprocess.CalledProcessError: ffprobe failed.
    """
    output = subprocess.check_output(
        ['ffprobe', '-loglevel', 'error', '-select_streams', 'v:0', '-show_entries', 'packet=pts_time,flags',
         '-of', 'csv=print_section=0', str(video_path)],
        text=True,
    )
    times = []
    for line in output.splitlines():
        fields = line.strip().split(',')
        if len(fields) >= 2 and 'K' in fields[1] and fields[0] not in ('', 'N/A'):
            times.append(float(fields[0]))
    return np.array(times)

determine_suffix_and_fourcc

determine_suffix_and_fourcc() -> tuple[str, str]

Return the container suffix and FourCC codec for annotated videos on this platform.

Returns:

Type Description
tuple[str, str]

('mp4', 'avc1') on macOS, ('avi', 'WMV2') on Windows and ('mp4', 'mp4v') elsewhere.

Source code in geotrax/utils/file_utils.py
def determine_suffix_and_fourcc() -> tuple[str, str]:
    """Return the container suffix and FourCC codec for annotated videos on this platform.

    Returns:
        `('mp4', 'avc1')` on macOS, `('avi', 'WMV2')` on Windows and `('mp4', 'mp4v')` elsewhere.
    """
    suffix = 'mp4' if MACOS else 'avi' if WINDOWS else 'mp4'
    fourcc = 'avc1' if MACOS else 'WMV2' if WINDOWS else 'mp4v'
    return suffix, fourcc