Skip to content

capturegraph.data.load.loaders #

PType Loader Registry#

Maps CaptureGraph PType names to Python loader functions. Each loader reads a JSON file (or returns a Path for binary files) and converts it to the appropriate Python type.

JSON structures are based on the Swift PType implementations: - PBool: {"bool": value} - PNumber: {"number": value, "is_time_interval": bool} - PString: {"string": value} - PTime: {"time_since_1970": seconds} - PUserID: {"name": str, "identifier": uuid} → returns dict for vectorized access

Design Guideline

Loaders should return dicts or primitives (not tuples/lists) to enable chained vectorized access via List. For example: sessions.user_id.name # Works when user_id is a dict sessions.user_id[0] # Less intuitive when user_id is a tuple

load_file(directory, name, manifest) #

Load a single file from a directory based on its manifest.

Tries each file extension in order until a file is found. Uses the appropriate loader based on type_name, or returns the Path for binary files without a registered loader.

Parameters:

Name Type Description Default
directory Path

The directory containing the file

required
name str

The base filename (without extension)

required
manifest FileManifest

The FileManifest describing the file's type and extensions

required

Returns:

Type Description
Any

The loaded value, a Dict for unknown JSON types,

Any

a Path for binary files, or None if not found.

Source code in capturegraph-lib/capturegraph/data/load/loaders.py
def load_file(
    directory: Path,
    name: str,
    manifest: FileManifest,
) -> Any:
    """Load a single file from a directory based on its manifest.

    Tries each file extension in order until a file is found.
    Uses the appropriate loader based on type_name, or returns
    the Path for binary files without a registered loader.

    Args:
        directory: The directory containing the file
        name: The base filename (without extension)
        manifest: The FileManifest describing the file's type and extensions

    Returns:
        The loaded value, a Dict for unknown JSON types,
        a Path for binary files, or None if not found.
    """
    type_name = manifest.type_name
    extensions = manifest.file_extensions

    for ext in extensions:
        file_path = directory / f"{name}.{ext}"
        if file_path.exists():
            if type_name in PTYPE_LOADERS:
                return PTYPE_LOADERS[type_name](file_path)
            elif ext == "json":
                return _load_json(file_path)
            else:
                # Return Path for non-JSON types without a loader (images, etc.)
                return file_path

    return MissingType(FileNotFoundError(f"File not found: {directory / name}"))

load_sequence(directory, name, manifest) #

Load a numbered file sequence from a directory.

Looks for files named {name}_0.ext, {name}_1.ext, etc. Files are sorted numerically by the number in the filename.

Parameters:

Name Type Description Default
directory Path

The directory containing the sequence files

required
name str

The base sequence name (e.g., "sequence" for sequence_0.dng)

required
manifest FileManifest

The FileManifest describing the file type and extensions

required

Returns:

Type Description
list[Any]

List of loaded values (may be empty if no files found).

Source code in capturegraph-lib/capturegraph/data/load/loaders.py
def load_sequence(
    directory: Path,
    name: str,
    manifest: FileManifest,
) -> list[Any]:
    """Load a numbered file sequence from a directory.

    Looks for files named {name}_0.ext, {name}_1.ext, etc.
    Files are sorted numerically by the number in the filename.

    Args:
        directory: The directory containing the sequence files
        name: The base sequence name (e.g., "sequence" for sequence_0.dng)
        manifest: The FileManifest describing the file type and extensions

    Returns:
        List of loaded values (may be empty if no files found).
    """
    import re

    type_name = manifest.type_name
    extensions = manifest.file_extensions

    def extract_number(path: Path) -> int:
        """Extract the numeric suffix from filename like 'name_123.ext' -> 123."""
        match = re.search(rf"{re.escape(name)}_(\d+)", path.stem)
        return int(match.group(1)) if match else 0

    result = []

    for ext in extensions:
        files = sorted(directory.glob(f"{name}_*.{ext}"), key=extract_number)
        for file_path in files:
            if type_name in PTYPE_LOADERS:
                result.append(PTYPE_LOADERS[type_name](file_path))
            elif ext == "json":
                result.append(_load_json(file_path))
            else:
                result.append(file_path)

        if result:
            break  # Found files with this extension, stop searching

    return result