Skip to content

deeplabcut.generate_training_dataset.metadata

File containing methods to load and parse shuffle metadata.

Classes:

Name Description
DataSplit

Class representing the metadata for a shuffle.

ShuffleMetadata

Class representing the metadata for a shuffle.

TrainingDatasetMetadata

An immutable class containing the metadata for a dataset.

Functions:

Name Description
find_engines_from_model_folders

Determines which engines are used with a given shuffle.

get_shuffle_engine

Args:

update_metadata

Updates the metadata for a training-dataset.

DataSplit dataclass

Class representing the metadata for a shuffle.

Methods:

Name Description
__post_init__

Raises:

Source code in deeplabcut/generate_training_dataset/metadata.py
@dataclass(frozen=True)
class DataSplit:
    """Class representing the metadata for a shuffle."""

    train_indices: tuple[int, ...]
    test_indices: tuple[int, ...]

    def __post_init__(self) -> None:
        """
        Raises:
            ValueError if the indices are not sorted in increasing
        """
        for indices in [self.train_indices, self.test_indices]:
            idx = np.array(indices)
            if not np.all(idx[:-1] < idx[1:]):
                raise RuntimeError(
                    "The training and test indices in a data split must be sorted in strictly ascending order."
                )

__post_init__

__post_init__() -> None
Source code in deeplabcut/generate_training_dataset/metadata.py
def __post_init__(self) -> None:
    """
    Raises:
        ValueError if the indices are not sorted in increasing
    """
    for indices in [self.train_indices, self.test_indices]:
        idx = np.array(indices)
        if not np.all(idx[:-1] < idx[1:]):
            raise RuntimeError(
                "The training and test indices in a data split must be sorted in strictly ascending order."
            )

ShuffleMetadata dataclass

Class representing the metadata for a shuffle.

Methods:

Name Description
load_split

Loads the data split for this shuffle.

Source code in deeplabcut/generate_training_dataset/metadata.py
@dataclass(frozen=True)
class ShuffleMetadata:
    """Class representing the metadata for a shuffle."""

    name: str
    train_fraction: float
    index: int
    engine: Engine
    split: DataSplit | None

    def load_split(self, cfg: dict, trainset_path: Path) -> ShuffleMetadata:
        """Loads the data split for this shuffle.

        Args:
            cfg: the config for the DeepLabCut project
            trainset_path: the path to the training dataset folder

        Returns:
            a new instance with the data split defined
        """
        _, doc_path = auxiliaryfunctions.get_data_and_metadata_filenames(
            trainset_path, self.train_fraction, self.index, cfg
        )
        if not Path(doc_path).exists():
            raise ValueError(
                f"Could not load the metadata file for {self} as {doc_path} does not "
                f"exist. If you deleted the shuffle, you also need to delete the "
                f"shuffle from metadata.yaml or recreate the metadata.yaml file."
            )

        with open(doc_path, "rb") as f:
            _, train_idx, test_idx, _ = pickle.load(f)
        return ShuffleMetadata(
            name=self.name,
            train_fraction=self.train_fraction,
            index=self.index,
            engine=self.engine,
            split=DataSplit(
                train_indices=tuple(sorted([int(idx) for idx in train_idx])),
                test_indices=tuple(sorted([int(idx) for idx in test_idx])),
            ),
        )

load_split

load_split(cfg: dict, trainset_path: Path) -> ShuffleMetadata

Loads the data split for this shuffle.

Parameters:

Name Type Description Default

cfg

dict

the config for the DeepLabCut project

required

trainset_path

Path

the path to the training dataset folder

required

Returns:

Type Description
ShuffleMetadata

a new instance with the data split defined

Source code in deeplabcut/generate_training_dataset/metadata.py
def load_split(self, cfg: dict, trainset_path: Path) -> ShuffleMetadata:
    """Loads the data split for this shuffle.

    Args:
        cfg: the config for the DeepLabCut project
        trainset_path: the path to the training dataset folder

    Returns:
        a new instance with the data split defined
    """
    _, doc_path = auxiliaryfunctions.get_data_and_metadata_filenames(
        trainset_path, self.train_fraction, self.index, cfg
    )
    if not Path(doc_path).exists():
        raise ValueError(
            f"Could not load the metadata file for {self} as {doc_path} does not "
            f"exist. If you deleted the shuffle, you also need to delete the "
            f"shuffle from metadata.yaml or recreate the metadata.yaml file."
        )

    with open(doc_path, "rb") as f:
        _, train_idx, test_idx, _ = pickle.load(f)
    return ShuffleMetadata(
        name=self.name,
        train_fraction=self.train_fraction,
        index=self.index,
        engine=self.engine,
        split=DataSplit(
            train_indices=tuple(sorted([int(idx) for idx in train_idx])),
            test_indices=tuple(sorted([int(idx) for idx in test_idx])),
        ),
    )

TrainingDatasetMetadata dataclass

An immutable class containing the metadata for a dataset.

When creating a new "training-datasets" folder (e.g., when creating the first training set for a project, or when creating the first training for a given iteration of a project), TrainingDatasetMetadata.create(cfg) should be called when the "training-datasets" folder is still empty.

For existing projects (created with DeepLabCut < 3.0), calling TrainingDatasetMetadata.create(cfg) will go over documentation data for all existing shuffles in the training-datasets folder and add them to a new metadata instance. All shuffles will be given Engine.TF as an engine.

Examples:

Creating the metadata file for an existing project

config = "/data/my-dlc-project/config.yaml" trainset_metadata = TrainingDatasetMetadata.create(config) trainset_metadata.save()

Adding a new shuffle to the metadata file

config = "/data/my-dlc-project-2008-06-17/config.yaml" trainset_metadata = TrainingDatasetMetadata.load(config) new_shuffle = ShuffleMetadata( name="my-dlc-projectJun17-trainset60shuffle5", train_fraction=0.6, index=5, engine=compat.Engine.PYTORCH, split=DataSplit(train_indices=(1, 3, 4), test_indices=(0, 2)), ) trainset_metadata = trainset_metadata.add(new_shuffle) trainset_metadata.save() # saves to disk

Methods:

Name Description
__post_init__

Raises:

add

Adds a new shuffle to the metadata file.

create

Function to create the metadata file.

get

Args:

load

Loads the metadata from disk.

path

Args:

save

Saves the training dataset metadata to disk.

Source code in deeplabcut/generate_training_dataset/metadata.py
@dataclass(frozen=True)
class TrainingDatasetMetadata:
    """An immutable class containing the metadata for a dataset.

    When creating a new "training-datasets" folder (e.g., when creating the first
    training set for a project, or when creating the first training for a given
    iteration of a project), TrainingDatasetMetadata.create(cfg) should be called when
    the "training-datasets" folder is still empty.

    For existing projects (created with DeepLabCut < 3.0), calling
    TrainingDatasetMetadata.create(cfg) will go over documentation data for all existing
    shuffles in the training-datasets folder and add them to a new metadata instance.
    All shuffles will be given Engine.TF as an engine.

    Examples:
        # Creating the metadata file for an existing project
        config = "/data/my-dlc-project/config.yaml"
        trainset_metadata = TrainingDatasetMetadata.create(config)
        trainset_metadata.save()

        # Adding a new shuffle to the metadata file
        config = "/data/my-dlc-project-2008-06-17/config.yaml"
        trainset_metadata = TrainingDatasetMetadata.load(config)
        new_shuffle = ShuffleMetadata(
            name="my-dlc-projectJun17-trainset60shuffle5",
            train_fraction=0.6,
            index=5,
            engine=compat.Engine.PYTORCH,
            split=DataSplit(train_indices=(1, 3, 4), test_indices=(0, 2)),
        )
        trainset_metadata = trainset_metadata.add(new_shuffle)
        trainset_metadata.save()  # saves to disk
    """

    project_config: dict
    shuffles: tuple[ShuffleMetadata, ...]
    file_header: tuple[str] = (
        "# This file is automatically generated - DO NOT EDIT",
        "# It contains the information about the shuffles created for the dataset",
        "---",
    )

    def __post_init__(self) -> None:
        """
        Raises:
            ValueError if the indices are not sorted in increasing order
        """
        indices = [[s.train_fraction, s.index] for s in self.shuffles]
        for (frac1, idx1), (frac2, idx2) in zip(indices[:-1], indices[1:], strict=False):
            if not (frac1 < frac2 or (frac1 == frac2 and idx1 < idx2)):
                raise RuntimeError(
                    "The shuffles given must be sorted in order of ascending training "
                    f"fraction and index. Found {self.shuffles}"
                )

    def add(
        self,
        shuffle: ShuffleMetadata,
        overwrite: bool = False,
    ) -> TrainingDatasetMetadata:
        """Adds a new shuffle to the metadata file.

        Args:
            shuffle: the shuffle to add
            overwrite: if a shuffle with the same index is already stored in the
                metadata file, whether to overwrite it

        Returns:
            A new instance of TrainingDatasetMetadata with updated shuffles

        Raises:
            ValueError: if overwrite=False and there is already a shuffle with the given
                index in the metadata file.
        """
        existing_indices = [s.index for s in self.shuffles if s.train_fraction == shuffle.train_fraction]
        if shuffle.index in existing_indices:
            if not overwrite:
                raise RuntimeError(
                    f"Cannot add {shuffle} to the meta: a shuffle with index "
                    f"{shuffle.index} and train_fraction {shuffle.train_fraction} "
                    f"already exists: {self.shuffles}."
                )

        existing_shuffles = [
            s for s in self.shuffles if (s.index != shuffle.index or s.train_fraction != shuffle.train_fraction)
        ]
        shuffles = existing_shuffles + [shuffle]
        return TrainingDatasetMetadata(
            project_config=self.project_config,
            shuffles=tuple(sorted(shuffles, key=lambda s: (s.train_fraction, s.index))),
        )

    def get(self, trainset_index: int = 0, index: int = 0) -> ShuffleMetadata:
        """
        Args:
            trainset_index: the index of the trainset fraction as defined in config.yaml
            index: the index of the shuffle

        Returns:
            the shuffle with the given trainset index and shuffle index

        Raises:
            ValueError if trainset_index is out of bounds or the shuffle is not present
        """
        fractions = self.project_config["TrainingFraction"]
        if trainset_index >= len(fractions):
            raise ValueError(
                f"trainset_index={trainset_index} is out of bounds for "
                f"TrainingFraction={fractions} (length {len(fractions)})."
            )
        train_fraction = fractions[trainset_index]
        for shuffle in self.shuffles:
            if shuffle.train_fraction == train_fraction and shuffle.index == index:
                return shuffle

        known = [(s.train_fraction, s.index) for s in self.shuffles] or "none"
        raise ValueError(
            f"Could not find a shuffle with train_fraction={train_fraction} and "
            f"index={index}. Known shuffles (fraction, index): {known}."
        )

    def save(self) -> None:
        """Saves the training dataset metadata to disk."""
        metadata = {"shuffles": {}}
        data_splits: dict[DataSplit, int] = {}
        trainset_path = self.path(self.project_config).parent
        for s in self.shuffles:
            if s.split is None:
                s = s.load_split(cfg=self.project_config, trainset_path=trainset_path)

            split_index = data_splits.get(s.split)
            if split_index is None:
                split_index = len(data_splits) + 1
                data_splits[s.split] = split_index

            metadata["shuffles"][s.name] = {
                "train_fraction": s.train_fraction,
                "index": s.index,
                "split": split_index,
                "engine": s.engine.aliases[0],
            }

        with open(self.path(self.project_config), "w") as file:
            file.write("\n".join(self.file_header) + "\n")
            YAML().dump(metadata, file)

    @staticmethod
    def load(
        config: str | Path | dict,
        load_splits: bool = False,
    ) -> TrainingDatasetMetadata:
        """Loads the metadata from disk.

        Args:
            config: the config for the DeepLabCut project (or its path)
            load_splits: whether to load the data split for each shuffle
        """
        if isinstance(config, (str, Path)):
            cfg = auxiliaryfunctions.read_config(config)
        else:
            cfg = config

        metadata_path = TrainingDatasetMetadata.path(cfg)
        if not metadata_path.exists():
            raise FileNotFoundError(f"No metadata.yaml found at {metadata_path}.")
        with open(metadata_path) as file:
            metadata = YAML(typ="safe", pure=True).load(file)

        shuffles = []
        for shuffle_name, shuffle_metadata in metadata["shuffles"].items():
            shuffle = ShuffleMetadata(
                name=shuffle_name,
                train_fraction=shuffle_metadata["train_fraction"],
                index=shuffle_metadata["index"],
                engine=Engine(shuffle_metadata["engine"]),
                split=None,
            )
            if load_splits:
                shuffle = shuffle.load_split(cfg, metadata_path.parent)

            shuffles.append(shuffle)

        shuffles.sort(key=lambda s: (s.train_fraction, s.index))
        return TrainingDatasetMetadata(project_config=cfg, shuffles=tuple(shuffles))

    @staticmethod
    def create(config: str | Path | dict) -> TrainingDatasetMetadata:
        """Function to create the metadata file.

        Assumes that all existing shuffles use the TensorFlow engine, as this file
        should have already been created for PyTorch shuffles.

        Args;
            config: the config for the DeepLabCut project (or its path)
            default_engine: the default engine to set for shuffles in the project

        Returns:
            the metadata for the existing shuffles in the project
        """
        if isinstance(config, (str, Path)):
            cfg = auxiliaryfunctions.read_config(config)
        else:
            cfg = config

        trainset_path = TrainingDatasetMetadata.path(cfg).parent
        if trainset_path.exists():
            shuffle_docs = [
                f for f in trainset_path.iterdir() if re.match(r"Documentation_data-.+shuffle[0-9]+\.pickle", f.name)
            ]
        else:
            trainset_path.mkdir(parents=True)
            shuffle_docs = []

        prefix = cfg["Task"] + cfg["date"]
        shuffles = []
        existing_splits: dict[tuple[tuple[int, ...], tuple[int, ...]], int] = {}
        for doc_path in shuffle_docs:
            index = int(doc_path.stem.split("shuffle")[-1])
            with open(doc_path, "rb") as f:
                _, train_idx, test_idx, train_frac = pickle.load(f)

            engine = Engine.TF
            train_idx = tuple(sorted([int(idx) for idx in train_idx]))
            test_idx = tuple(sorted([int(idx) for idx in test_idx]))
            split_idx = existing_splits.get((train_idx, test_idx))
            if split_idx is None:
                split_idx = len(existing_splits) + 1
                existing_splits[(train_idx, test_idx)] = split_idx

            shuffles.append(
                ShuffleMetadata(
                    name=f"{prefix}-trainset{int(100 * train_frac)}shuffle{index}",
                    train_fraction=train_frac,
                    index=index,
                    engine=engine,
                    split=DataSplit(train_indices=train_idx, test_indices=test_idx),
                )
            )

        shuffles = tuple(sorted(shuffles, key=lambda s: (s.train_fraction, s.index)))
        return TrainingDatasetMetadata(
            project_config=cfg,
            shuffles=shuffles,
        )

    @staticmethod
    def path(cfg: dict) -> Path:
        """
        Args:
            cfg: the config for the DeepLabCut project

        Returns:
            the path to the training dataset metadata file
        """
        meta_path = auxiliaryfunctions.get_training_set_folder(cfg) / "metadata.yaml"
        return Path(cfg["project_path"]) / meta_path

__post_init__

__post_init__() -> None
Source code in deeplabcut/generate_training_dataset/metadata.py
def __post_init__(self) -> None:
    """
    Raises:
        ValueError if the indices are not sorted in increasing order
    """
    indices = [[s.train_fraction, s.index] for s in self.shuffles]
    for (frac1, idx1), (frac2, idx2) in zip(indices[:-1], indices[1:], strict=False):
        if not (frac1 < frac2 or (frac1 == frac2 and idx1 < idx2)):
            raise RuntimeError(
                "The shuffles given must be sorted in order of ascending training "
                f"fraction and index. Found {self.shuffles}"
            )

add

add(shuffle: ShuffleMetadata, overwrite: bool = False) -> TrainingDatasetMetadata

Adds a new shuffle to the metadata file.

Parameters:

Name Type Description Default

shuffle

ShuffleMetadata

the shuffle to add

required

overwrite

bool

if a shuffle with the same index is already stored in the metadata file, whether to overwrite it

False

Returns:

Type Description
TrainingDatasetMetadata

A new instance of TrainingDatasetMetadata with updated shuffles

Raises:

Type Description
ValueError

if overwrite=False and there is already a shuffle with the given index in the metadata file.

Source code in deeplabcut/generate_training_dataset/metadata.py
def add(
    self,
    shuffle: ShuffleMetadata,
    overwrite: bool = False,
) -> TrainingDatasetMetadata:
    """Adds a new shuffle to the metadata file.

    Args:
        shuffle: the shuffle to add
        overwrite: if a shuffle with the same index is already stored in the
            metadata file, whether to overwrite it

    Returns:
        A new instance of TrainingDatasetMetadata with updated shuffles

    Raises:
        ValueError: if overwrite=False and there is already a shuffle with the given
            index in the metadata file.
    """
    existing_indices = [s.index for s in self.shuffles if s.train_fraction == shuffle.train_fraction]
    if shuffle.index in existing_indices:
        if not overwrite:
            raise RuntimeError(
                f"Cannot add {shuffle} to the meta: a shuffle with index "
                f"{shuffle.index} and train_fraction {shuffle.train_fraction} "
                f"already exists: {self.shuffles}."
            )

    existing_shuffles = [
        s for s in self.shuffles if (s.index != shuffle.index or s.train_fraction != shuffle.train_fraction)
    ]
    shuffles = existing_shuffles + [shuffle]
    return TrainingDatasetMetadata(
        project_config=self.project_config,
        shuffles=tuple(sorted(shuffles, key=lambda s: (s.train_fraction, s.index))),
    )

create staticmethod

create(config: str | Path | dict) -> TrainingDatasetMetadata

Function to create the metadata file.

Assumes that all existing shuffles use the TensorFlow engine, as this file should have already been created for PyTorch shuffles.

Args; config: the config for the DeepLabCut project (or its path) default_engine: the default engine to set for shuffles in the project

Returns:

Type Description
TrainingDatasetMetadata

the metadata for the existing shuffles in the project

Source code in deeplabcut/generate_training_dataset/metadata.py
@staticmethod
def create(config: str | Path | dict) -> TrainingDatasetMetadata:
    """Function to create the metadata file.

    Assumes that all existing shuffles use the TensorFlow engine, as this file
    should have already been created for PyTorch shuffles.

    Args;
        config: the config for the DeepLabCut project (or its path)
        default_engine: the default engine to set for shuffles in the project

    Returns:
        the metadata for the existing shuffles in the project
    """
    if isinstance(config, (str, Path)):
        cfg = auxiliaryfunctions.read_config(config)
    else:
        cfg = config

    trainset_path = TrainingDatasetMetadata.path(cfg).parent
    if trainset_path.exists():
        shuffle_docs = [
            f for f in trainset_path.iterdir() if re.match(r"Documentation_data-.+shuffle[0-9]+\.pickle", f.name)
        ]
    else:
        trainset_path.mkdir(parents=True)
        shuffle_docs = []

    prefix = cfg["Task"] + cfg["date"]
    shuffles = []
    existing_splits: dict[tuple[tuple[int, ...], tuple[int, ...]], int] = {}
    for doc_path in shuffle_docs:
        index = int(doc_path.stem.split("shuffle")[-1])
        with open(doc_path, "rb") as f:
            _, train_idx, test_idx, train_frac = pickle.load(f)

        engine = Engine.TF
        train_idx = tuple(sorted([int(idx) for idx in train_idx]))
        test_idx = tuple(sorted([int(idx) for idx in test_idx]))
        split_idx = existing_splits.get((train_idx, test_idx))
        if split_idx is None:
            split_idx = len(existing_splits) + 1
            existing_splits[(train_idx, test_idx)] = split_idx

        shuffles.append(
            ShuffleMetadata(
                name=f"{prefix}-trainset{int(100 * train_frac)}shuffle{index}",
                train_fraction=train_frac,
                index=index,
                engine=engine,
                split=DataSplit(train_indices=train_idx, test_indices=test_idx),
            )
        )

    shuffles = tuple(sorted(shuffles, key=lambda s: (s.train_fraction, s.index)))
    return TrainingDatasetMetadata(
        project_config=cfg,
        shuffles=shuffles,
    )

get

get(trainset_index: int = 0, index: int = 0) -> ShuffleMetadata

Parameters:

Name Type Description Default

trainset_index

int

the index of the trainset fraction as defined in config.yaml

0

index

int

the index of the shuffle

0

Returns:

Type Description
ShuffleMetadata

the shuffle with the given trainset index and shuffle index

Source code in deeplabcut/generate_training_dataset/metadata.py
def get(self, trainset_index: int = 0, index: int = 0) -> ShuffleMetadata:
    """
    Args:
        trainset_index: the index of the trainset fraction as defined in config.yaml
        index: the index of the shuffle

    Returns:
        the shuffle with the given trainset index and shuffle index

    Raises:
        ValueError if trainset_index is out of bounds or the shuffle is not present
    """
    fractions = self.project_config["TrainingFraction"]
    if trainset_index >= len(fractions):
        raise ValueError(
            f"trainset_index={trainset_index} is out of bounds for "
            f"TrainingFraction={fractions} (length {len(fractions)})."
        )
    train_fraction = fractions[trainset_index]
    for shuffle in self.shuffles:
        if shuffle.train_fraction == train_fraction and shuffle.index == index:
            return shuffle

    known = [(s.train_fraction, s.index) for s in self.shuffles] or "none"
    raise ValueError(
        f"Could not find a shuffle with train_fraction={train_fraction} and "
        f"index={index}. Known shuffles (fraction, index): {known}."
    )

load staticmethod

load(config: str | Path | dict, load_splits: bool = False) -> TrainingDatasetMetadata

Loads the metadata from disk.

Parameters:

Name Type Description Default

config

str | Path | dict

the config for the DeepLabCut project (or its path)

required

load_splits

bool

whether to load the data split for each shuffle

False
Source code in deeplabcut/generate_training_dataset/metadata.py
@staticmethod
def load(
    config: str | Path | dict,
    load_splits: bool = False,
) -> TrainingDatasetMetadata:
    """Loads the metadata from disk.

    Args:
        config: the config for the DeepLabCut project (or its path)
        load_splits: whether to load the data split for each shuffle
    """
    if isinstance(config, (str, Path)):
        cfg = auxiliaryfunctions.read_config(config)
    else:
        cfg = config

    metadata_path = TrainingDatasetMetadata.path(cfg)
    if not metadata_path.exists():
        raise FileNotFoundError(f"No metadata.yaml found at {metadata_path}.")
    with open(metadata_path) as file:
        metadata = YAML(typ="safe", pure=True).load(file)

    shuffles = []
    for shuffle_name, shuffle_metadata in metadata["shuffles"].items():
        shuffle = ShuffleMetadata(
            name=shuffle_name,
            train_fraction=shuffle_metadata["train_fraction"],
            index=shuffle_metadata["index"],
            engine=Engine(shuffle_metadata["engine"]),
            split=None,
        )
        if load_splits:
            shuffle = shuffle.load_split(cfg, metadata_path.parent)

        shuffles.append(shuffle)

    shuffles.sort(key=lambda s: (s.train_fraction, s.index))
    return TrainingDatasetMetadata(project_config=cfg, shuffles=tuple(shuffles))

path staticmethod

path(cfg: dict) -> Path

Parameters:

Name Type Description Default

cfg

dict

the config for the DeepLabCut project

required

Returns:

Type Description
Path

the path to the training dataset metadata file

Source code in deeplabcut/generate_training_dataset/metadata.py
@staticmethod
def path(cfg: dict) -> Path:
    """
    Args:
        cfg: the config for the DeepLabCut project

    Returns:
        the path to the training dataset metadata file
    """
    meta_path = auxiliaryfunctions.get_training_set_folder(cfg) / "metadata.yaml"
    return Path(cfg["project_path"]) / meta_path

save

save() -> None

Saves the training dataset metadata to disk.

Source code in deeplabcut/generate_training_dataset/metadata.py
def save(self) -> None:
    """Saves the training dataset metadata to disk."""
    metadata = {"shuffles": {}}
    data_splits: dict[DataSplit, int] = {}
    trainset_path = self.path(self.project_config).parent
    for s in self.shuffles:
        if s.split is None:
            s = s.load_split(cfg=self.project_config, trainset_path=trainset_path)

        split_index = data_splits.get(s.split)
        if split_index is None:
            split_index = len(data_splits) + 1
            data_splits[s.split] = split_index

        metadata["shuffles"][s.name] = {
            "train_fraction": s.train_fraction,
            "index": s.index,
            "split": split_index,
            "engine": s.engine.aliases[0],
        }

    with open(self.path(self.project_config), "w") as file:
        file.write("\n".join(self.file_header) + "\n")
        YAML().dump(metadata, file)

find_engines_from_model_folders

find_engines_from_model_folders(cfg: dict, trainingsetindex: int, shuffle: int, modelprefix: str = '') -> set[Engine]

Determines which engines are used with a given shuffle.

This method can be useful when using modelprefix, as the engine for a shuffle stored under a "modelprefix" might not be the same as the base shuffle (for which the engine is stored in the training-datasets folder).

Parameters:

Name Type Description Default

cfg

dict

the config for the DeepLabCut project

required

trainingsetindex

int

the training set index used

required

shuffle

int

the shuffle for which to get the engine

required

modelprefix

str

the model prefix, if there is one

''

Returns:

Type Description
set[Engine]

the engines for which a model folder exists for the given shuffle

Source code in deeplabcut/generate_training_dataset/metadata.py
def find_engines_from_model_folders(
    cfg: dict,
    trainingsetindex: int,
    shuffle: int,
    modelprefix: str = "",
) -> set[Engine]:
    """Determines which engines are used with a given shuffle.

    This method can be useful when using modelprefix, as the engine for a shuffle stored
    under a "modelprefix" might not be the same as the base shuffle (for which the
    engine is stored in the training-datasets folder).

    Args:
        cfg: the config for the DeepLabCut project
        trainingsetindex: the training set index used
        shuffle: the shuffle for which to get the engine
        modelprefix: the model prefix, if there is one

    Returns:
        the engines for which a model folder exists for the given shuffle
    """
    project_path = Path(cfg["project_path"])
    train_fraction = cfg["TrainingFraction"][trainingsetindex]

    existing_engines = set()
    for engine in Engine:
        expected_model_folder = project_path / auxiliaryfunctions.get_model_folder(
            trainFraction=train_fraction,
            shuffle=shuffle,
            cfg=cfg,
            engine=engine,
            modelprefix=modelprefix,
        )
        if expected_model_folder.exists():
            existing_engines.add(engine)

    return existing_engines

get_shuffle_engine

get_shuffle_engine(cfg: dict, trainingsetindex: int, shuffle: int, modelprefix: str = '') -> Engine

Parameters:

Name Type Description Default

cfg

dict

the config for the DeepLabCut project

required

trainingsetindex

int

the training set index used

required

shuffle

int

the shuffle for which to get the engine

required

modelprefix

str

the model prefix, if there is one

''

Returns:

Type Description
Engine

the engine that the shuffle was created with

Source code in deeplabcut/generate_training_dataset/metadata.py
def get_shuffle_engine(
    cfg: dict,
    trainingsetindex: int,
    shuffle: int,
    modelprefix: str = "",
) -> Engine:
    """
    Args:
        cfg: the config for the DeepLabCut project
        trainingsetindex: the training set index used
        shuffle: the shuffle for which to get the engine
        modelprefix: the model prefix, if there is one

    Returns:
        the engine that the shuffle was created with

    Raises:
        ValueError if the engine for the shuffle cannot be determined or the shuffle
        doesn't exist
    """
    if not TrainingDatasetMetadata.path(cfg).exists():
        metadata = TrainingDatasetMetadata.create(cfg)
        if metadata.shuffles:
            # only persist when there is actual content to avoid writing empty files
            metadata.save()
    else:
        metadata = TrainingDatasetMetadata.load(cfg)

    # Try to resolve the shuffle from metadata; fall through to model-folder detection
    # on failure so that inference works even when metadata is incomplete.
    shuffle_metadata = None
    try:
        shuffle_metadata = metadata.get(trainingsetindex, shuffle)
    except ValueError as e:
        logging.warning(
            "Could not read shuffle metadata for trainingsetindex=%s, shuffle=%s: %s. "
            "Falling back to detecting the engine from model folders.",
            trainingsetindex,
            shuffle,
            e,
        )

    if shuffle_metadata is not None:
        return shuffle_metadata.engine

    engines = find_engines_from_model_folders(cfg, trainingsetindex, shuffle, modelprefix)
    if len(engines) == 0:
        prefix_str = f" and modelprefix={modelprefix}" if modelprefix else ""
        raise ValueError(
            f"Couldn't find any shuffles with trainingsetindex={trainingsetindex}, "
            f"shuffle={shuffle}{prefix_str}. The shuffle was not found "
            "in metadata.yaml and no model folder exists for it. Please check that "
            "such a shuffle is defined."
        )

    engine = list(engines)[0]  # Get any engine from the set
    if len(engines) > 1:
        logging.warning(
            f"Found multiple engines for trainingsetindex={trainingsetindex}, "
            f"shuffle={shuffle} and modelprefix={modelprefix}. Using engine={engine}. "
            f"To select another engine, please specify it in your API call."
        )
    return engine

update_metadata

update_metadata(
    cfg: dict,
    train_fraction: float,
    shuffle: int,
    engine: Engine,
    train_indices: list[int],
    test_indices: list[int],
    overwrite: bool = False,
) -> None

Updates the metadata for a training-dataset.

Parameters:

Name Type Description Default

cfg

dict

the config for the DeepLabCut project

required

train_fraction

float

the train_fraction of the new shuffle

required

shuffle

int

the index of the shuffle to add

required

engine

Engine

the engine for the shuffle

required

train_indices

list[int]

the indices of images in the training set

required

test_indices

list[int]

the indices of images in the test set

required

overwrite

bool

whether to overwrite a shuffle with the same index and train fraction if one exists

False

Raises:

Type Description
ValueError

if overwrite=False and there is already a shuffle with the given index in the metadata file.

Source code in deeplabcut/generate_training_dataset/metadata.py
def update_metadata(
    cfg: dict,
    train_fraction: float,
    shuffle: int,
    engine: Engine,
    train_indices: list[int],
    test_indices: list[int],
    overwrite: bool = False,
) -> None:
    """Updates the metadata for a training-dataset.

    Args:
        cfg: the config for the DeepLabCut project
        train_fraction: the train_fraction of the new shuffle
        shuffle: the index of the shuffle to add
        engine: the engine for the shuffle
        train_indices: the indices of images in the training set
        test_indices: the indices of images in the test set
        overwrite: whether to overwrite a shuffle with the same index and train fraction
            if one exists

    Raises:
        ValueError: if overwrite=False and there is already a shuffle with the given
            index in the metadata file.
    """
    prefix = cfg["Task"] + cfg["date"]
    metadata = TrainingDatasetMetadata.load(cfg, load_splits=True)
    new_shuffle = ShuffleMetadata(
        name=f"{prefix}-trainset{int(100 * train_fraction)}shuffle{shuffle}",
        train_fraction=train_fraction,
        index=shuffle,
        engine=engine,
        split=DataSplit(
            train_indices=tuple(sorted([int(i) for i in train_indices])),
            test_indices=tuple(sorted([int(i) for i in test_indices])),
        ),
    )
    metadata = metadata.add(shuffle=new_shuffle, overwrite=overwrite)
    metadata.save()