Directory

datamate.directory

This module exports the Directory class, an array- and metadata-friendly view into a directory.

Instances of the base Directory class have methods to simplify reading/writing collections of arrays.

Directory

Array- and metadata-friendly view into a directory.

Provides a dictionary-like interface for working with arrays and metadata stored in a directory structure.

Parameters:

Name	Type	Description	Default
path	Optional[Path]	Path at which the Directory is/should be stored. Can be relative to current root_dir. If not provided, the Directory is created relative to the current root_dir.	required
config	Optional[Dict[str, Any]]	Configuration dictionary. When including a type field, indicates the Directory type to search for and construct.	required

Attributes:

Name	Type	Description
path	Path	Path where Directory is stored. Type: pathlib.Path
config	Config	Directory configuration. Type: Config
meta	Namespace	Metadata stored in _meta.yaml. Type: Namespace
status	Literal['running', 'done', 'stopped']	Build status from metadata. Type: Literal["running", "done", "stopped"]
parent	Directory	Parent directory. Type: Directory

Valid constructors

# Auto-name relative to root_dir:
Directory()
Directory(config={"type": "MyType"})

# Name relative to root_dir or absolute:
Directory("/path/to/dir")
Directory("/path/to/dir", config={"type": "MyType"})

After instantiation, Directories act as string-keyed mutable dictionaries containing: - ArrayFiles: Single-entry HDF5 files in SWMR mode - Paths: Non-array files in other formats - Directories: Subdirectories

Array-like numeric and byte-string data written via __setitem__, __setattr__, or extend is stored as an array file.

Example

# Create directory with arrays
dir = Directory("my_data")
dir["array1"] = np.array([1,2,3])
dir.array2 = np.array([[4,5],[6,7]])

# Access data
arr1 = dir["array1"]  # Returns array([1,2,3])
arr2 = dir.array2     # Returns array([[4,5],[6,7]])

Source code in datamate/directory.py

class Directory(metaclass=NonExistingDirectory):
    """Array- and metadata-friendly view into a directory.

    Provides a dictionary-like interface for working with arrays and metadata stored in
    a directory structure.

    Args:
        path (Optional[Path]): Path at which the Directory is/should be stored.
            Can be relative to current `root_dir`. If not provided, the Directory is
            created relative to the current `root_dir`.
        config (Optional[Dict[str, Any]]): Configuration dictionary. When including a
            `type` field, indicates the Directory type to search for and construct.

    Attributes:
        path: Path where Directory is stored.
            Type: `pathlib.Path`
        config: Directory configuration.
            Type: `Config`
        meta: Metadata stored in `_meta.yaml`.
            Type: `Namespace`
        status: Build status from metadata.
            Type: `Literal["running", "done", "stopped"]`
        parent: Parent directory.
            Type: `Directory`

    Valid constructors:
        ```python
        # Auto-name relative to root_dir:
        Directory()
        Directory(config={"type": "MyType"})

        # Name relative to root_dir or absolute:
        Directory("/path/to/dir")
        Directory("/path/to/dir", config={"type": "MyType"})
        ```

    After instantiation, Directories act as string-keyed mutable dictionaries
    containing:
    - ArrayFiles: Single-entry HDF5 files in SWMR mode
    - Paths: Non-array files in other formats
    - Directories: Subdirectories

    Array-like numeric and byte-string data written via `__setitem__`, `__setattr__`, or
    `extend` is stored as an array file.

    Example:
        ```python
        # Create directory with arrays
        dir = Directory("my_data")
        dir["array1"] = np.array([1,2,3])
        dir.array2 = np.array([[4,5],[6,7]])

        # Access data
        arr1 = dir["array1"]  # Returns array([1,2,3])
        arr2 = dir.array2     # Returns array([[4,5],[6,7]])
        ```
    """

    class Config(Protocol):
        """Protocol defining the configuration interface for Directory classes.

        This protocol defines the structure of the `config` argument to the
        `Directory` constructor, and provides type hints for the `config`
        attribute of `Directory` instances.

        Note:
            Subclasses should implement this protocol to define their configuration
            interface, or implement `__init__` with typed parameters.
        """

        pass

    path: Path
    config: Config

    @overload
    def __new__(cls: type[T]) -> T: ...

    @overload
    def __new__(cls: type[T], path: Union[str, Path]) -> T: ...

    @overload
    def __new__(cls: type[T], config: ConfigType) -> T: ...

    @overload
    def __new__(cls: type[T], path: Union[str, Path], config: ConfigType) -> T: ...

    def __new__(_type: type[T], *args: object, **kwargs: object) -> T:
        """Implementation of overloaded constructors."""
        path, config = _parse_directory_args(args, kwargs)

        if path is not None and isinstance(path, Path) and path.exists():
            # case 1: path exists and global context is deleting if exists
            if context.delete_if_exists.get():
                shutil.rmtree(path)
            # case 2: path exists and local kwargs are deleting if exists
            if (
                config is not None
                and "delete_if_exists" in config
                and config["delete_if_exists"]
            ):
                shutil.rmtree(path)

            if config is not None and "delete_if_exists" in config:
                # always remove the deletion flag from the config
                config.pop("delete_if_exists")

        cls = _directory(_type)
        _check_implementation(cls)

        defaults = get_defaults(cls)

        if config is None and defaults:  # and _implements_init(cls):
            # to initialize from defaults if no config or path is provided
            if path is None or path is not None and not path.exists():
                config = defaults
            # if a non-empty path is provided, we cannot initialize from defaults
            else:
                pass
        # breakpoint()
        if path is not None and config is None:
            cls = _directory_from_path(cls, _resolve_path(path))
        elif path is None and config is not None:
            cls = _directory_from_config(cls, config)
        elif path is not None and config is not None:
            cls = _directory_from_path_and_config(cls, _resolve_path(path), config)
        elif path is None and config is None and _implements_init(cls):
            # raise ValueError("no configuration provided")
            pass

        if context.check_size_on_init.get():
            cls.check_size()

        return cls

    def __init__(self) -> None:
        """Implement to compile `Directory` from a configuration.

        Note:
            Subclasses can either implement `Config` to determine the interface,
            types and defaults of `config`, or implement `__init__` with keyword
            arguments. If both are implemented, the config is created from the joined
            interface as long as defaults are not conflicting.
        """
        pass

    def __init_subclass__(cls, **kwargs) -> None:
        """Initializes a Directory subclass.

        Automatically generates documentation for the subclass.

        Args:
            **kwargs: Additional keyword arguments passed to parent __init_subclass__
        """
        super().__init_subclass__(**kwargs)
        cls.__doc__ = _auto_doc(cls)

    @property
    def meta(self) -> Namespace:
        """The metadata stored in `{self.path}/_meta.yaml`."""
        return read_meta(self.path)

    @property
    def config(self) -> Config:
        """The directory configuration."""
        return self.meta.config or self._config

    @config.setter
    def config(self, value: Config) -> None:
        self.__manual_config(value)

    @property
    def status(self) -> Literal["running", "done", "stopped"]:
        """The build status from metadata."""
        return self.meta.status

    @property
    def size(self) -> int:
        """Total size of directory in bytes."""
        return check_size(self.path, warning_at=float("inf"), print_size=False)

    @property
    def is_empty(self) -> bool:
        """Whether directory contains any files."""
        return len(self) == 0

    @property
    def modified(self) -> bool:
        """Whether directory has been modified after initialization."""
        return getattr(self.meta, "modified", False)

    # -- MutableMapping methods ----------------------------

    def __len__(self) -> int:
        """Returns the number of public files in `self.path`.

        Non-public files (files whose names start with "_") are not counted.

        Returns:
            Number of public files in the directory
        """
        return sum(1 for _ in self.path.glob("[!_]*"))

    def __iter__(self) -> Iterator[str]:
        """Yields field names corresponding to the public files in `self.path`.

        Entries it understands (subdirectories and HDF5 files) are yielded
        without extensions. Non-public files (files whose names start with "_")
        are ignored.

        Yields:
            Field names for each public file
        """
        for p in self.path.glob("[!_]*"):
            yield p.name.rpartition(".")[0] if p.suffix in [".h5", ".csv"] else p.name

    def __copy__(self) -> "Directory":
        """Creates a shallow copy of the directory.

        Returns:
            New `Directory` instance pointing to the same path
        """
        return Directory(self.path)

    def __deepcopy__(self, memodict={}):
        return self.__copy__()

    def keys(self) -> Iterator[str]:
        """Returns an iterator over public file names in the directory.

        Returns:
            Iterator yielding public file names
        """
        return self.__iter__()

    def items(self) -> Iterator[Tuple[str, ArrayFile]]:
        """Returns an iterator over (key, value) pairs in the directory.

        Returns:
            Iterator yielding tuples of (filename, file content)
        """
        for key in self.keys():
            yield (key, self[key])

    @classmethod
    def from_df(
        cls: type[T], df: DataFrame, dtypes: Dict[str, Any], *args, **kwargs
    ) -> T:
        """Create a Directory from a DataFrame by splitting into column arrays.

        Each column is stored as a separate HDF5 array with specified dtype.
        This is different from storing the DataFrame directly, which uses CSV format.

        Args:
            df: Source DataFrame
            dtypes: Dictionary mapping column names to numpy dtypes
            *args: Additional arguments passed to `Directory` constructor
            **kwargs: Additional keyword arguments passed to `Directory` constructor

        Returns:
            `Directory` with each column stored as a separate array

        Examples:
            ```python
            df = pd.DataFrame({'a': [1, 2, 3], 'b': ['x', 'y', 'z']})
            dtypes = {'a': np.int64, 'b': 'S'}

            # Store columns as separate arrays
            dir1 = Directory.from_df(df, dtypes)
            # Results in:
            # dir1/
            #   ├── a.h5  # array([1, 2, 3])
            #   └── b.h5  # array([b'x', b'y', b'z'])

            # Store as single CSV
            dir2 = Directory()
            dir2['data'] = df
            # Results in:
            # dir2/
            #   └── data.csv
            ```
        """
        directory = Directory.__new__(Directory, *args, **kwargs)
        directory.update({
            column: df[column].values.astype(dtypes[column]) for column in df.columns
        })
        return directory

    def update(self, other: Union[Dict, "Directory"], suffix: str = "") -> None:
        """Updates self with items of other and appends an optional suffix.

        Args:
            other: Dictionary or Directory to copy items from
            suffix: Optional string to append to copied keys
        """
        for key in other:
            if key + suffix not in self:
                self[key + suffix] = other[key]

    def move(self, dst: Union[str, Path]) -> "Directory":
        """Moves directory to new location.

        Args:
            dst: Destination path

        Returns:
            New `Directory` instance at the destination path
        """
        shutil.move(self.path, dst)
        return Directory(dst)

    def rmtree(self, y_n: Optional[str] = None) -> None:
        """Recursively deletes the directory after confirmation.

        Args:
            y_n: Optional pre-supplied confirmation ('y' or 'n'). If not provided,
                will prompt user interactively
        """
        reply = y_n or input(f"delete {self.path} recursively, y/n?")
        if reply.lower() == "y":
            shutil.rmtree(self.path, ignore_errors=True)

    def _rebuild(self, y_n: Optional[str] = None) -> None:
        """Rebuilds the directory by deleting and recreating it.

        Args:
            y_n: Optional pre-supplied confirmation ('y' or 'n'). If not provided,
                will prompt user interactively
        """
        self.rmtree(y_n)
        _build(self)

    def __truediv__(self, other: str) -> Any:
        """Implements path-like division operator for accessing entries.

        Args:
            other: Key to access

        Returns:
            Same as `self[other]`
        """
        return self.__getitem__(other)

    def __getitem__(self, key: str) -> Any:
        """Returns `ArrayFile`, `Path`, or `Directory` corresponding to `self.path/key`.

        HDF5 files are returned as `ArrayFile`s, other files as `Path`s, and
        directories and nonexistent entries as (possibly empty) `Directory`s.

        Args:
            key: Name of the entry to retrieve

        Returns:
            The requested entry as an appropriate type

        Note:
            Attribute access syntax is also supported, and occurrences of `__` in
            `key` are transformed into `.`, to support accessing encoded files as
            attributes (i.e. `Directory['name.ext']` is equivalent to
            `Directory.name__ext`).
        """

        try:
            # to catch cases where key is an index to a reference to an h5 file.
            # this will yield a TypeError because Path / slice does not work.
            path = self.path / key
        except TypeError as e:
            if not self.path.exists():
                # we wanted to index an H5Dataset but we tried to index a Directory
                # because the H5Dataset does not exist
                raise FileNotFoundError(
                    f"Indexing {self.path.name} at {key} not possible for"
                    f" Directory at {self.path.parent}. File "
                    f"{self.path.name}.h5 does not exist."
                ) from e
            raise e

        # Return an array.
        if path.with_suffix(".h5").is_file():
            return _read_h5(path.with_suffix(".h5"))

        # Return a csv
        if path.with_suffix(".csv").is_file():
            return pd.read_csv(path.with_suffix(".csv"))

        # Return the path to a file.
        elif path.is_file():
            return path

        # Return a subrecord
        else:
            return Directory(path)

    def __setitem__(self, key: str, val: object) -> None:
        """
        Writes an `ArrayFile`, `Path`, or `Directory` to `self.path/key`

        `np.ndarray`-like objects are written as `ArrayFiles`, `Path`-like
        objects are written as `Path`s, and string-keyed mappings are
        written as subDirectorys.

        Attribute access syntax is also supported, and occurrences of "__" in
        `key` are transformed into ".", to support accessing encoded files as
        attributes (i.e. `Directory['name.ext'] = val` is equivalent to
        `Directory.name__ext = val`).
        """

        path = self.path / key

        # Copy an existing file or directory.
        if isinstance(val, Path):
            if os.path.isfile(val):
                _copy_file(path, val)
            elif os.path.isdir(val):
                _copy_dir(path, val)

        # Write a Directory instance
        elif isinstance(val, Directory):
            assert path.suffix == ""
            # Create new directory with same type and config as source
            new_dir = type(val)(path, config=val.config)
            MutableMapping.update(new_dir, val)

        # Write a mapping as a new Directory
        elif isinstance(val, Mapping):
            assert path.suffix == ""
            MutableMapping.update(Directory(path), val)  # type: ignore

        # Write a dataframe.
        elif isinstance(val, pd.DataFrame):  # Use pd.DataFrame explicitly
            assert path.suffix == ""
            if not path.parent.exists():
                path.parent.mkdir(parents=True, exist_ok=True)
            val.to_csv(path.with_suffix(".csv"), index=False)

        # Write an array.
        else:
            assert path.suffix == ""
            if isinstance(val, H5Reader):
                val = val[()]
            try:
                _write_h5(path.with_suffix(".h5"), val)
            except TypeError as err:
                raise TypeError(
                    format_tb(err.__traceback__)[0]
                    + err.args[0]
                    + f"\nYou're trying to store {val} which cannot be converted to "
                    f"h5-file in {path}."
                    + "\nFor reference of supported types, see "
                    + "https://docs.h5py.org/en/stable/faq.html?highlight=types"
                    + "#numpy-object-types"
                    + "\nE.g. NumPy unicode strings must be converted to 'S' strings "
                    + "and back:"
                    + "\nfoo.bar = array.astype('S') to store and foo.bar[:]."
                    + "astype('U') "
                    + "to retrieve."
                ) from None

        if self.config is not None and self.status == "done":
            # Track if a Directory has been modified past __init__
            self._modified_past_init(True)

    def __delitem__(self, key: str) -> None:
        """
        Deletes the entry at `self.path/key`

        Attribute access syntax is also supported, and occurrences of "__" in
        `key` are transformed into ".", to support accessing encoded files as
        attributes (i.e. `del Directory['name.ext']` is equivalent to
        `del Directory.name__ext`).
        """
        path = self.path / key

        # Delete an array file.
        if path.with_suffix(".h5").is_file():
            path.with_suffix(".h5").unlink()

        # Delete a csv file.
        if path.with_suffix(".csv").is_file():
            path.with_suffix(".csv").unlink()

        # Delete a non-array file.
        elif path.is_file():
            path.unlink()

        # Delete a Directory.
        else:
            shutil.rmtree(path, ignore_errors=True)

    def __eq__(self, other: object) -> bool:
        """Returns True if `self` and `other` are equal.

        Two Directories are equal if they have the same keys and the same
        values for each key.

        Args:
            other: Object to compare against

        Returns:
            Whether the directories are equal

        Raises:
            ValueError: If comparing `Directory` with incompatible type
        """
        if not isinstance(other, Directory):
            raise ValueError(f"Cannot compare Directory to {type(other)}")

        if self.path == other.path:
            return True

        if self.path != other.path:
            diff = DirectoryDiff(self, other)
            return diff.equal(fail=False)

    def __neq__(self, other: object) -> bool:
        """Returns True if directories are not equal.

        Args:
            other: Object to compare against

        Returns:
            Whether the directories are not equal
        """
        return not self.__eq__(other)

    def diff(self, other: "Directory") -> Dict[str, List[str]]:
        """Returns a dictionary of differences between this directory and another.

        Args:
            other: Directory to compare against

        Returns:
            Dictionary with two keys - the name of self and other. Values are lists of
            strings describing differences between corresponding entries.
        """
        diff = DirectoryDiff(self, other)
        return diff.diff()

    def extend(self, key: str, val: object) -> None:
        """Extends an array, file or directory at the given key.

        Extending arrays performs concatenation along the first axis,
        extending files performs byte-level concatenation, and
        extending directories extends their fields.

        Args:
            key: Name of the entry to extend
            val: Value to append to the existing entry. Can be `np.ndarray`, `Path`,
                `Directory`, or `Mapping`

        Note:
            Files corresponding to `self[key]` are created if they do not already exist.
        """

        path = self.path / key

        # Append an existing file.
        if isinstance(val, Path):
            assert path.suffix != ""
            _extend_file(path, val)

        # Append a subDirectory.
        elif isinstance(val, (Mapping, Directory)):
            assert path.suffix == ""
            for k in val:
                Directory(path).extend(k, val[k])

        elif isinstance(val, pd.DataFrame):
            assert path.suffix == ""
            if path.with_suffix(".csv").is_file():
                old_df = pd.read_csv(path.with_suffix(".csv"))
                new_df = pd.concat([old_df, val], axis=0)
            else:
                new_df = val
            new_df.to_csv(path.with_suffix(".csv"), index=False)

        # Append an array.
        else:
            assert path.suffix == ""
            if isinstance(val, H5Reader):
                val = val[()]
            _extend_h5(path.with_suffix(".h5"), val)

        if self.config is not None and self.status == "done":
            # Track if a Directory has been modified past __init__
            self._modified_past_init(True)

    # --- Views ---

    def __repr__(self):
        if context.verbosity_level.get() == 1:
            string = tree(
                self.path,
                last_modified=True,
                level=2,
                length_limit=25,
                verbose=True,
                not_exists_message="empty",
            )
        elif context.verbosity_level.get() == 0:
            string = tree(
                self.path,
                last_modified=True,
                level=1,
                length_limit=0,
                verbose=False,
                not_exists_message="empty",
            )
        else:
            string = tree(
                self.path,
                level=-1,
                length_limit=None,
                last_modified=True,
                verbose=True,
                limit_to_directories=False,
            )
        return string

    def tree(
        self,
        level: int = -1,
        length_limit: Optional[int] = None,
        verbose: bool = True,
        last_modified: bool = True,
        limit_to_directories: bool = False,
    ) -> None:
        """Prints a tree representation of the directory structure.

        Args:
            level: Maximum depth to display (-1 for unlimited)
            length_limit: Maximum number of entries to show per directory
            verbose: Whether to show detailed information
            last_modified: Whether to show last modification times
            limit_to_directories: Whether to only show directories
        """
        print(
            tree(
                self.path,
                level=level,
                length_limit=length_limit,
                last_modified=last_modified,
                verbose=verbose,
                limit_to_directories=limit_to_directories,
            )
        )

    # -- Attribute-style element access --------------------

    def __getattr__(self, key: str) -> Any:
        if key.startswith("__") and key.endswith("__"):  # exclude dunder attributes
            return None
        return self.__getitem__(key.replace("__", "."))

    def __setattr__(self, key: str, value: object) -> None:
        # Fix autoreload related effect.
        if key.startswith("__") and key.endswith("__"):
            object.__setattr__(self, key, value)
            return
        # allow manual config writing
        if key == "config":
            self.__manual_config(value)
            return
        self.__setitem__(key.replace("__", "."), value)

    def __delattr__(self, key: str) -> None:
        self.__delitem__(key.replace("__", "."))

    # -- Attribute preemption, for REPL autocompletion -----

    def __getattribute__(self, key: str) -> Any:
        if key in object.__getattribute__(self, "_cached_keys"):
            try:
                object.__setattr__(self, key, self[key])
            except KeyError:
                object.__delattr__(self, key)
                object.__getattribute__(self, "_cached_keys").remove(key)
        return object.__getattribute__(self, key)

    def __dir__(self) -> List[str]:
        for key in self._cached_keys:
            object.__delattr__(self, key)
        self._cached_keys.clear()

        for key in set(self).difference(object.__dir__(self)):
            object.__setattr__(self, key, self[key])
            self._cached_keys.add(key)

        return cast(list, object.__dir__(self))

    # -- Convenience methods

    def __manual_config(self, config, status=None):
        """Overriding config stored in _meta.yaml.

        config (Dict): update for meta.config
        status (str): status if config did not exist before, i.e. _overrid_config
            is used to store a _meta.yaml for the first time instead of build.
        """
        meta_path = self.path / "_meta.yaml"

        current_config = self.config
        config = namespacify(config)
        if current_config is not None:
            with warnings.catch_warnings():
                warnings.simplefilter("always")
                warnings.warn(
                    (
                        f"Overriding config. Diff is:"
                        f'{config.diff(current_config, name1="passed", name2="stored")}'
                    ),
                    ConfigWarning,
                    stacklevel=2,
                )
            write_meta(path=meta_path, config=config, status="manually written")
        else:
            write_meta(path=meta_path, config=config, status=status or self.status)

    def _override_status(self, status: Literal["running", "done", "stopped"]) -> None:
        """Overrides the build status in metadata.

        Args:
            status: New status to set. Must be one of "running", "done", or "stopped"

        Warns:
            `ConfigWarning`: When overriding an existing status
        """
        meta_path = self.path / "_meta.yaml"

        current_status = self.status
        if current_status is not None:
            with warnings.catch_warnings():
                warnings.simplefilter("always")
                warnings.warn(
                    (f"Overriding status {current_status} to {status}"),
                    ConfigWarning,
                    stacklevel=2,
                )
        write_meta(path=meta_path, config=self.config, status=status)

    def _modified_past_init(self, is_modified: bool) -> None:
        """Tracks if a `Directory` has been modified after initialization.

        Updates the metadata file to record modification status.

        Args:
            is_modified: Whether the directory has been modified

        Note:
            This is used to warn users when attempting to reuse a modified directory.
        """
        meta_path = self.path / "_meta.yaml"

        if is_modified:
            write_meta(
                path=meta_path, config=self.config, status=self.status, modified=True
            )

    def check_size(
        self,
        warning_at: int = 20 * 1024**3,  # 20GB
        print_size: bool = False,
        *,
        raise_on_warning: bool = False,
    ) -> int:
        """Checks the total size of the directory.

        Args:
            warning_at: Size in bytes at which to issue a warning
            print_size: Whether to print the directory size
            raise_on_warning: Whether to raise exception instead of warning

        Returns:
            Total size in bytes

        Raises:
            ValueError: if directory size exceeds warning_at and raise_on_warning
                is True
        """
        size = check_size(self.path, warning_at, print_size)
        if raise_on_warning and size > warning_at:
            raise ValueError(f"Directory size {size} exceeds limit {warning_at}")
        return size

    def to_df(self, dtypes: Optional[Dict[str, Any]] = None) -> DataFrame:
        """Reconstruct a DataFrame from HDF5 column arrays in this directory.

        Combines all equal-length, single-dimensional HDF5 datasets into
        DataFrame columns. Results are cached to avoid expensive recomputation.

        Args:
            dtypes: Optional dictionary mapping column names to numpy dtypes

        Returns:
            `DataFrame` reconstructed from HDF5 column arrays

        Note:
            This is the complement to `from_df()`. While direct DataFrame assignment
            stores as CSV, `from_df()` splits columns into HDF5 arrays which can be
            recombined using this method.
        """
        try:
            return object.__getattribute__(self, "_as_df")
        except AttributeError:
            object.__setattr__(self, "_as_df", directory_to_df(self, dtypes))
            return self.to_df()

    def to_dict(self) -> DataFrame:
        """
        Returns a DataFrame from all equal length, single-dim .h5 datasets in self.path.
        """
        # to cache the dict that is expensive to create.
        try:
            return object.__getattribute__(self, "_as_dict")
        except AttributeError:
            object.__setattr__(self, "_as_dict", directory_to_dict(self))
            return self.to_dict()

    def mtime(self) -> datetime.datetime:
        """Returns the last modification time of the directory.

        Returns:
            Datetime object representing last modification time
        """
        return datetime.datetime.fromtimestamp(self.path.stat().st_mtime)

    @property
    def parent(self) -> "Directory":
        """The parent directory."""
        return Directory(self.path.absolute().parent)

    def _count(self) -> int:
        """Counts number of existing numbered subdirectories.

        Returns:
            Number of subdirectories matching pattern '[0-9a-f]{4}'
        """
        root = self.path
        count = 0
        for i in itertools.count():
            dst = root / f"{i:04x}"
            if dst.exists():
                count += 1
            else:
                return count
        return count

    def _next(self) -> "Directory":
        """Creates next available numbered subdirectory.

        Returns:
            New `Directory` instance for the next available numbered subdirectory

        Raises:
            AssertionError: If the next numbered directory already exists
        """
        root = self.path
        dst = root / f"{self._count():04x}"
        assert not dst.exists()
        return Directory(dst, self.config)

    def _clear_filetype(self, suffix: str) -> None:
        """Deletes all files with given suffix in the current directory.

        Args:
            suffix: File extension to match (e.g. '.h5', '.csv')
        """
        for file in self.path.iterdir():
            if file.is_file() and file.suffix == suffix:
                file.unlink()

meta property

meta

The metadata stored in {self.path}/_meta.yaml.

config property writable

config

The directory configuration.

status property

status

The build status from metadata.

size property

size

Total size of directory in bytes.

is_empty property

is_empty

Whether directory contains any files.

modified property

modified

Whether directory has been modified after initialization.

parent property

parent

The parent directory.

Config

Bases: Protocol

Protocol defining the configuration interface for Directory classes.

This protocol defines the structure of the config argument to the Directory constructor, and provides type hints for the config attribute of Directory instances.

Note

Subclasses should implement this protocol to define their configuration interface, or implement __init__ with typed parameters.

Source code in datamate/directory.py

class Config(Protocol):
    """Protocol defining the configuration interface for Directory classes.

    This protocol defines the structure of the `config` argument to the
    `Directory` constructor, and provides type hints for the `config`
    attribute of `Directory` instances.

    Note:
        Subclasses should implement this protocol to define their configuration
        interface, or implement `__init__` with typed parameters.
    """

    pass

__new__

__new__() -> T

__new__(path: Union[str, Path]) -> T

__new__(config: ConfigType) -> T

__new__(path: Union[str, Path], config: ConfigType) -> T

__new__(_type, *args, **kwargs)

Implementation of overloaded constructors.

Source code in datamate/directory.py

def __new__(_type: type[T], *args: object, **kwargs: object) -> T:
    """Implementation of overloaded constructors."""
    path, config = _parse_directory_args(args, kwargs)

    if path is not None and isinstance(path, Path) and path.exists():
        # case 1: path exists and global context is deleting if exists
        if context.delete_if_exists.get():
            shutil.rmtree(path)
        # case 2: path exists and local kwargs are deleting if exists
        if (
            config is not None
            and "delete_if_exists" in config
            and config["delete_if_exists"]
        ):
            shutil.rmtree(path)

        if config is not None and "delete_if_exists" in config:
            # always remove the deletion flag from the config
            config.pop("delete_if_exists")

    cls = _directory(_type)
    _check_implementation(cls)

    defaults = get_defaults(cls)

    if config is None and defaults:  # and _implements_init(cls):
        # to initialize from defaults if no config or path is provided
        if path is None or path is not None and not path.exists():
            config = defaults
        # if a non-empty path is provided, we cannot initialize from defaults
        else:
            pass
    # breakpoint()
    if path is not None and config is None:
        cls = _directory_from_path(cls, _resolve_path(path))
    elif path is None and config is not None:
        cls = _directory_from_config(cls, config)
    elif path is not None and config is not None:
        cls = _directory_from_path_and_config(cls, _resolve_path(path), config)
    elif path is None and config is None and _implements_init(cls):
        # raise ValueError("no configuration provided")
        pass

    if context.check_size_on_init.get():
        cls.check_size()

    return cls

__init__

__init__()

Implement to compile Directory from a configuration.

Note

Subclasses can either implement Config to determine the interface, types and defaults of config, or implement __init__ with keyword arguments. If both are implemented, the config is created from the joined interface as long as defaults are not conflicting.

Source code in datamate/directory.py

def __init__(self) -> None:
    """Implement to compile `Directory` from a configuration.

    Note:
        Subclasses can either implement `Config` to determine the interface,
        types and defaults of `config`, or implement `__init__` with keyword
        arguments. If both are implemented, the config is created from the joined
        interface as long as defaults are not conflicting.
    """
    pass

__init_subclass__

__init_subclass__(**kwargs)

Initializes a Directory subclass.

Automatically generates documentation for the subclass.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments passed to parent init_subclass	{}

Source code in datamate/directory.py

def __init_subclass__(cls, **kwargs) -> None:
    """Initializes a Directory subclass.

    Automatically generates documentation for the subclass.

    Args:
        **kwargs: Additional keyword arguments passed to parent __init_subclass__
    """
    super().__init_subclass__(**kwargs)
    cls.__doc__ = _auto_doc(cls)

__len__

__len__()

Returns the number of public files in self.path.

Non-public files (files whose names start with “_”) are not counted.

Returns:

Type	Description
int	Number of public files in the directory

Source code in datamate/directory.py

def __len__(self) -> int:
    """Returns the number of public files in `self.path`.

    Non-public files (files whose names start with "_") are not counted.

    Returns:
        Number of public files in the directory
    """
    return sum(1 for _ in self.path.glob("[!_]*"))

__iter__

__iter__()

Yields field names corresponding to the public files in self.path.

Entries it understands (subdirectories and HDF5 files) are yielded without extensions. Non-public files (files whose names start with “_”) are ignored.

Yields:

Type	Description
str	Field names for each public file

Source code in datamate/directory.py

def __iter__(self) -> Iterator[str]:
    """Yields field names corresponding to the public files in `self.path`.

    Entries it understands (subdirectories and HDF5 files) are yielded
    without extensions. Non-public files (files whose names start with "_")
    are ignored.

    Yields:
        Field names for each public file
    """
    for p in self.path.glob("[!_]*"):
        yield p.name.rpartition(".")[0] if p.suffix in [".h5", ".csv"] else p.name

__copy__

__copy__()

Creates a shallow copy of the directory.

Returns:

Type	Description
Directory	New Directory instance pointing to the same path

Source code in datamate/directory.py

def __copy__(self) -> "Directory":
    """Creates a shallow copy of the directory.

    Returns:
        New `Directory` instance pointing to the same path
    """
    return Directory(self.path)

keys

keys()

Returns an iterator over public file names in the directory.

Returns:

Type	Description
Iterator[str]	Iterator yielding public file names

Source code in datamate/directory.py

def keys(self) -> Iterator[str]:
    """Returns an iterator over public file names in the directory.

    Returns:
        Iterator yielding public file names
    """
    return self.__iter__()

items

items()

Returns an iterator over (key, value) pairs in the directory.

Returns:

Type	Description
Iterator[Tuple[str, ArrayFile]]	Iterator yielding tuples of (filename, file content)

Source code in datamate/directory.py

def items(self) -> Iterator[Tuple[str, ArrayFile]]:
    """Returns an iterator over (key, value) pairs in the directory.

    Returns:
        Iterator yielding tuples of (filename, file content)
    """
    for key in self.keys():
        yield (key, self[key])

from_df classmethod

from_df(df, dtypes, *args, **kwargs)

Create a Directory from a DataFrame by splitting into column arrays.

Each column is stored as a separate HDF5 array with specified dtype. This is different from storing the DataFrame directly, which uses CSV format.

Parameters:

Name	Type	Description	Default
df	DataFrame	Source DataFrame	required
dtypes	Dict[str, Any]	Dictionary mapping column names to numpy dtypes	required
*args		Additional arguments passed to Directory constructor	()
**kwargs		Additional keyword arguments passed to Directory constructor	{}

Returns:

Type	Description
T	Directory with each column stored as a separate array

Examples:

df = pd.DataFrame({'a': [1, 2, 3], 'b': ['x', 'y', 'z']})
dtypes = {'a': np.int64, 'b': 'S'}

# Store columns as separate arrays
dir1 = Directory.from_df(df, dtypes)
# Results in:
# dir1/
#   ├── a.h5  # array([1, 2, 3])
#   └── b.h5  # array([b'x', b'y', b'z'])

# Store as single CSV
dir2 = Directory()
dir2['data'] = df
# Results in:
# dir2/
#   └── data.csv

Source code in datamate/directory.py

@classmethod
def from_df(
    cls: type[T], df: DataFrame, dtypes: Dict[str, Any], *args, **kwargs
) -> T:
    """Create a Directory from a DataFrame by splitting into column arrays.

    Each column is stored as a separate HDF5 array with specified dtype.
    This is different from storing the DataFrame directly, which uses CSV format.

    Args:
        df: Source DataFrame
        dtypes: Dictionary mapping column names to numpy dtypes
        *args: Additional arguments passed to `Directory` constructor
        **kwargs: Additional keyword arguments passed to `Directory` constructor

    Returns:
        `Directory` with each column stored as a separate array

    Examples:
        ```python
        df = pd.DataFrame({'a': [1, 2, 3], 'b': ['x', 'y', 'z']})
        dtypes = {'a': np.int64, 'b': 'S'}

        # Store columns as separate arrays
        dir1 = Directory.from_df(df, dtypes)
        # Results in:
        # dir1/
        #   ├── a.h5  # array([1, 2, 3])
        #   └── b.h5  # array([b'x', b'y', b'z'])

        # Store as single CSV
        dir2 = Directory()
        dir2['data'] = df
        # Results in:
        # dir2/
        #   └── data.csv
        ```
    """
    directory = Directory.__new__(Directory, *args, **kwargs)
    directory.update({
        column: df[column].values.astype(dtypes[column]) for column in df.columns
    })
    return directory

update

update(other, suffix='')

Updates self with items of other and appends an optional suffix.

Parameters:

Name	Type	Description	Default
other	Union[Dict, Directory]	Dictionary or Directory to copy items from	required
suffix	str	Optional string to append to copied keys	''

Source code in datamate/directory.py

def update(self, other: Union[Dict, "Directory"], suffix: str = "") -> None:
    """Updates self with items of other and appends an optional suffix.

    Args:
        other: Dictionary or Directory to copy items from
        suffix: Optional string to append to copied keys
    """
    for key in other:
        if key + suffix not in self:
            self[key + suffix] = other[key]

move

move(dst)

Moves directory to new location.

Parameters:

Name	Type	Description	Default
dst	Union[str, Path]	Destination path	required

Returns:

Type	Description
Directory	New Directory instance at the destination path

Source code in datamate/directory.py

def move(self, dst: Union[str, Path]) -> "Directory":
    """Moves directory to new location.

    Args:
        dst: Destination path

    Returns:
        New `Directory` instance at the destination path
    """
    shutil.move(self.path, dst)
    return Directory(dst)

rmtree

rmtree(y_n=None)

Recursively deletes the directory after confirmation.

Parameters:

Name	Type	Description	Default
y_n	Optional[str]	Optional pre-supplied confirmation (‘y’ or ‘n’). If not provided, will prompt user interactively	None

Source code in datamate/directory.py

def rmtree(self, y_n: Optional[str] = None) -> None:
    """Recursively deletes the directory after confirmation.

    Args:
        y_n: Optional pre-supplied confirmation ('y' or 'n'). If not provided,
            will prompt user interactively
    """
    reply = y_n or input(f"delete {self.path} recursively, y/n?")
    if reply.lower() == "y":
        shutil.rmtree(self.path, ignore_errors=True)

__truediv__

__truediv__(other)

Implements path-like division operator for accessing entries.

Parameters:

Name	Type	Description	Default
other	str	Key to access	required

Returns:

Type	Description
Any	Same as self[other]

Source code in datamate/directory.py

def __truediv__(self, other: str) -> Any:
    """Implements path-like division operator for accessing entries.

    Args:
        other: Key to access

    Returns:
        Same as `self[other]`
    """
    return self.__getitem__(other)

__getitem__

__getitem__(key)

Returns ArrayFile, Path, or Directory corresponding to self.path/key.

HDF5 files are returned as ArrayFiles, other files as Paths, and directories and nonexistent entries as (possibly empty) Directorys.

Parameters:

Name	Type	Description	Default
key	str	Name of the entry to retrieve	required

Returns:

Type	Description
Any	The requested entry as an appropriate type

Note

Attribute access syntax is also supported, and occurrences of __ in key are transformed into ., to support accessing encoded files as attributes (i.e. Directory['name.ext'] is equivalent to Directory.name__ext).

Source code in datamate/directory.py

def __getitem__(self, key: str) -> Any:
    """Returns `ArrayFile`, `Path`, or `Directory` corresponding to `self.path/key`.

    HDF5 files are returned as `ArrayFile`s, other files as `Path`s, and
    directories and nonexistent entries as (possibly empty) `Directory`s.

    Args:
        key: Name of the entry to retrieve

    Returns:
        The requested entry as an appropriate type

    Note:
        Attribute access syntax is also supported, and occurrences of `__` in
        `key` are transformed into `.`, to support accessing encoded files as
        attributes (i.e. `Directory['name.ext']` is equivalent to
        `Directory.name__ext`).
    """

    try:
        # to catch cases where key is an index to a reference to an h5 file.
        # this will yield a TypeError because Path / slice does not work.
        path = self.path / key
    except TypeError as e:
        if not self.path.exists():
            # we wanted to index an H5Dataset but we tried to index a Directory
            # because the H5Dataset does not exist
            raise FileNotFoundError(
                f"Indexing {self.path.name} at {key} not possible for"
                f" Directory at {self.path.parent}. File "
                f"{self.path.name}.h5 does not exist."
            ) from e
        raise e

    # Return an array.
    if path.with_suffix(".h5").is_file():
        return _read_h5(path.with_suffix(".h5"))

    # Return a csv
    if path.with_suffix(".csv").is_file():
        return pd.read_csv(path.with_suffix(".csv"))

    # Return the path to a file.
    elif path.is_file():
        return path

    # Return a subrecord
    else:
        return Directory(path)

__setitem__

__setitem__(key, val)

Writes an ArrayFile, Path, or Directory to self.path/key

np.ndarray-like objects are written as ArrayFiles, Path-like objects are written as Paths, and string-keyed mappings are written as subDirectorys.

Attribute access syntax is also supported, and occurrences of “__” in key are transformed into “.”, to support accessing encoded files as attributes (i.e. Directory['name.ext'] = val is equivalent to Directory.name__ext = val).

Source code in datamate/directory.py

def __setitem__(self, key: str, val: object) -> None:
    """
    Writes an `ArrayFile`, `Path`, or `Directory` to `self.path/key`

    `np.ndarray`-like objects are written as `ArrayFiles`, `Path`-like
    objects are written as `Path`s, and string-keyed mappings are
    written as subDirectorys.

    Attribute access syntax is also supported, and occurrences of "__" in
    `key` are transformed into ".", to support accessing encoded files as
    attributes (i.e. `Directory['name.ext'] = val` is equivalent to
    `Directory.name__ext = val`).
    """

    path = self.path / key

    # Copy an existing file or directory.
    if isinstance(val, Path):
        if os.path.isfile(val):
            _copy_file(path, val)
        elif os.path.isdir(val):
            _copy_dir(path, val)

    # Write a Directory instance
    elif isinstance(val, Directory):
        assert path.suffix == ""
        # Create new directory with same type and config as source
        new_dir = type(val)(path, config=val.config)
        MutableMapping.update(new_dir, val)

    # Write a mapping as a new Directory
    elif isinstance(val, Mapping):
        assert path.suffix == ""
        MutableMapping.update(Directory(path), val)  # type: ignore

    # Write a dataframe.
    elif isinstance(val, pd.DataFrame):  # Use pd.DataFrame explicitly
        assert path.suffix == ""
        if not path.parent.exists():
            path.parent.mkdir(parents=True, exist_ok=True)
        val.to_csv(path.with_suffix(".csv"), index=False)

    # Write an array.
    else:
        assert path.suffix == ""
        if isinstance(val, H5Reader):
            val = val[()]
        try:
            _write_h5(path.with_suffix(".h5"), val)
        except TypeError as err:
            raise TypeError(
                format_tb(err.__traceback__)[0]
                + err.args[0]
                + f"\nYou're trying to store {val} which cannot be converted to "
                f"h5-file in {path}."
                + "\nFor reference of supported types, see "
                + "https://docs.h5py.org/en/stable/faq.html?highlight=types"
                + "#numpy-object-types"
                + "\nE.g. NumPy unicode strings must be converted to 'S' strings "
                + "and back:"
                + "\nfoo.bar = array.astype('S') to store and foo.bar[:]."
                + "astype('U') "
                + "to retrieve."
            ) from None

    if self.config is not None and self.status == "done":
        # Track if a Directory has been modified past __init__
        self._modified_past_init(True)

__delitem__

__delitem__(key)

Deletes the entry at self.path/key

Attribute access syntax is also supported, and occurrences of “__” in key are transformed into “.”, to support accessing encoded files as attributes (i.e. del Directory['name.ext'] is equivalent to del Directory.name__ext).

Source code in datamate/directory.py

def __delitem__(self, key: str) -> None:
    """
    Deletes the entry at `self.path/key`

    Attribute access syntax is also supported, and occurrences of "__" in
    `key` are transformed into ".", to support accessing encoded files as
    attributes (i.e. `del Directory['name.ext']` is equivalent to
    `del Directory.name__ext`).
    """
    path = self.path / key

    # Delete an array file.
    if path.with_suffix(".h5").is_file():
        path.with_suffix(".h5").unlink()

    # Delete a csv file.
    if path.with_suffix(".csv").is_file():
        path.with_suffix(".csv").unlink()

    # Delete a non-array file.
    elif path.is_file():
        path.unlink()

    # Delete a Directory.
    else:
        shutil.rmtree(path, ignore_errors=True)

__eq__

__eq__(other)

Returns True if self and other are equal.

Two Directories are equal if they have the same keys and the same values for each key.

Parameters:

Name	Type	Description	Default
other	object	Object to compare against	required

Returns:

Type	Description
bool	Whether the directories are equal

Raises:

Type	Description
ValueError	If comparing Directory with incompatible type

Source code in datamate/directory.py

def __eq__(self, other: object) -> bool:
    """Returns True if `self` and `other` are equal.

    Two Directories are equal if they have the same keys and the same
    values for each key.

    Args:
        other: Object to compare against

    Returns:
        Whether the directories are equal

    Raises:
        ValueError: If comparing `Directory` with incompatible type
    """
    if not isinstance(other, Directory):
        raise ValueError(f"Cannot compare Directory to {type(other)}")

    if self.path == other.path:
        return True

    if self.path != other.path:
        diff = DirectoryDiff(self, other)
        return diff.equal(fail=False)

__neq__

__neq__(other)

Returns True if directories are not equal.

Parameters:

Name	Type	Description	Default
other	object	Object to compare against	required

Returns:

Type	Description
bool	Whether the directories are not equal

Source code in datamate/directory.py

def __neq__(self, other: object) -> bool:
    """Returns True if directories are not equal.

    Args:
        other: Object to compare against

    Returns:
        Whether the directories are not equal
    """
    return not self.__eq__(other)

diff

diff(other)

Returns a dictionary of differences between this directory and another.

Parameters:

Name	Type	Description	Default
other	Directory	Directory to compare against	required

Returns:

Type	Description
Dict[str, List[str]]	Dictionary with two keys - the name of self and other. Values are lists of
Dict[str, List[str]]	strings describing differences between corresponding entries.

Source code in datamate/directory.py

def diff(self, other: "Directory") -> Dict[str, List[str]]:
    """Returns a dictionary of differences between this directory and another.

    Args:
        other: Directory to compare against

    Returns:
        Dictionary with two keys - the name of self and other. Values are lists of
        strings describing differences between corresponding entries.
    """
    diff = DirectoryDiff(self, other)
    return diff.diff()

extend

extend(key, val)

Extends an array, file or directory at the given key.

Extending arrays performs concatenation along the first axis, extending files performs byte-level concatenation, and extending directories extends their fields.

Parameters:

Name	Type	Description	Default
key	str	Name of the entry to extend	required
val	object	Value to append to the existing entry. Can be np.ndarray, Path, Directory, or Mapping	required

Note

Files corresponding to self[key] are created if they do not already exist.

Source code in datamate/directory.py

def extend(self, key: str, val: object) -> None:
    """Extends an array, file or directory at the given key.

    Extending arrays performs concatenation along the first axis,
    extending files performs byte-level concatenation, and
    extending directories extends their fields.

    Args:
        key: Name of the entry to extend
        val: Value to append to the existing entry. Can be `np.ndarray`, `Path`,
            `Directory`, or `Mapping`

    Note:
        Files corresponding to `self[key]` are created if they do not already exist.
    """

    path = self.path / key

    # Append an existing file.
    if isinstance(val, Path):
        assert path.suffix != ""
        _extend_file(path, val)

    # Append a subDirectory.
    elif isinstance(val, (Mapping, Directory)):
        assert path.suffix == ""
        for k in val:
            Directory(path).extend(k, val[k])

    elif isinstance(val, pd.DataFrame):
        assert path.suffix == ""
        if path.with_suffix(".csv").is_file():
            old_df = pd.read_csv(path.with_suffix(".csv"))
            new_df = pd.concat([old_df, val], axis=0)
        else:
            new_df = val
        new_df.to_csv(path.with_suffix(".csv"), index=False)

    # Append an array.
    else:
        assert path.suffix == ""
        if isinstance(val, H5Reader):
            val = val[()]
        _extend_h5(path.with_suffix(".h5"), val)

    if self.config is not None and self.status == "done":
        # Track if a Directory has been modified past __init__
        self._modified_past_init(True)

tree

tree(
    level=-1,
    length_limit=None,
    verbose=True,
    last_modified=True,
    limit_to_directories=False,
)

Prints a tree representation of the directory structure.

Parameters:

Name	Type	Description	Default
level	int	Maximum depth to display (-1 for unlimited)	-1
length_limit	Optional[int]	Maximum number of entries to show per directory	None
verbose	bool	Whether to show detailed information	True
last_modified	bool	Whether to show last modification times	True
limit_to_directories	bool	Whether to only show directories	False

Source code in datamate/directory.py

def tree(
    self,
    level: int = -1,
    length_limit: Optional[int] = None,
    verbose: bool = True,
    last_modified: bool = True,
    limit_to_directories: bool = False,
) -> None:
    """Prints a tree representation of the directory structure.

    Args:
        level: Maximum depth to display (-1 for unlimited)
        length_limit: Maximum number of entries to show per directory
        verbose: Whether to show detailed information
        last_modified: Whether to show last modification times
        limit_to_directories: Whether to only show directories
    """
    print(
        tree(
            self.path,
            level=level,
            length_limit=length_limit,
            last_modified=last_modified,
            verbose=verbose,
            limit_to_directories=limit_to_directories,
        )
    )

__manual_config

__manual_config(config, status=None)

Overriding config stored in _meta.yaml.

config (Dict): update for meta.config status (str): status if config did not exist before, i.e. _overrid_config is used to store a _meta.yaml for the first time instead of build.

Source code in datamate/directory.py

def __manual_config(self, config, status=None):
    """Overriding config stored in _meta.yaml.

    config (Dict): update for meta.config
    status (str): status if config did not exist before, i.e. _overrid_config
        is used to store a _meta.yaml for the first time instead of build.
    """
    meta_path = self.path / "_meta.yaml"

    current_config = self.config
    config = namespacify(config)
    if current_config is not None:
        with warnings.catch_warnings():
            warnings.simplefilter("always")
            warnings.warn(
                (
                    f"Overriding config. Diff is:"
                    f'{config.diff(current_config, name1="passed", name2="stored")}'
                ),
                ConfigWarning,
                stacklevel=2,
            )
        write_meta(path=meta_path, config=config, status="manually written")
    else:
        write_meta(path=meta_path, config=config, status=status or self.status)

check_size

check_size(
    warning_at=20 * 1024**3,
    print_size=False,
    *,
    raise_on_warning=False,
)

Checks the total size of the directory.

Parameters:

Name	Type	Description	Default
warning_at	int	Size in bytes at which to issue a warning	20 * 1024 ** 3
print_size	bool	Whether to print the directory size	False
raise_on_warning	bool	Whether to raise exception instead of warning	False

Returns:

Type	Description
int	Total size in bytes

Raises:

Type	Description
ValueError	if directory size exceeds warning_at and raise_on_warning is True

Source code in datamate/directory.py

def check_size(
    self,
    warning_at: int = 20 * 1024**3,  # 20GB
    print_size: bool = False,
    *,
    raise_on_warning: bool = False,
) -> int:
    """Checks the total size of the directory.

    Args:
        warning_at: Size in bytes at which to issue a warning
        print_size: Whether to print the directory size
        raise_on_warning: Whether to raise exception instead of warning

    Returns:
        Total size in bytes

    Raises:
        ValueError: if directory size exceeds warning_at and raise_on_warning
            is True
    """
    size = check_size(self.path, warning_at, print_size)
    if raise_on_warning and size > warning_at:
        raise ValueError(f"Directory size {size} exceeds limit {warning_at}")
    return size

to_df

to_df(dtypes=None)

Reconstruct a DataFrame from HDF5 column arrays in this directory.

Combines all equal-length, single-dimensional HDF5 datasets into DataFrame columns. Results are cached to avoid expensive recomputation.

Parameters:

Name	Type	Description	Default
dtypes	Optional[Dict[str, Any]]	Optional dictionary mapping column names to numpy dtypes	None

Returns:

Type	Description
DataFrame	DataFrame reconstructed from HDF5 column arrays

Note

This is the complement to from_df(). While direct DataFrame assignment stores as CSV, from_df() splits columns into HDF5 arrays which can be recombined using this method.

Source code in datamate/directory.py

def to_df(self, dtypes: Optional[Dict[str, Any]] = None) -> DataFrame:
    """Reconstruct a DataFrame from HDF5 column arrays in this directory.

    Combines all equal-length, single-dimensional HDF5 datasets into
    DataFrame columns. Results are cached to avoid expensive recomputation.

    Args:
        dtypes: Optional dictionary mapping column names to numpy dtypes

    Returns:
        `DataFrame` reconstructed from HDF5 column arrays

    Note:
        This is the complement to `from_df()`. While direct DataFrame assignment
        stores as CSV, `from_df()` splits columns into HDF5 arrays which can be
        recombined using this method.
    """
    try:
        return object.__getattribute__(self, "_as_df")
    except AttributeError:
        object.__setattr__(self, "_as_df", directory_to_df(self, dtypes))
        return self.to_df()

to_dict

to_dict()

Returns a DataFrame from all equal length, single-dim .h5 datasets in self.path.

Source code in datamate/directory.py

def to_dict(self) -> DataFrame:
    """
    Returns a DataFrame from all equal length, single-dim .h5 datasets in self.path.
    """
    # to cache the dict that is expensive to create.
    try:
        return object.__getattribute__(self, "_as_dict")
    except AttributeError:
        object.__setattr__(self, "_as_dict", directory_to_dict(self))
        return self.to_dict()

mtime

mtime()

Returns the last modification time of the directory.

Returns:

Type	Description
datetime	Datetime object representing last modification time

Source code in datamate/directory.py

def mtime(self) -> datetime.datetime:
    """Returns the last modification time of the directory.

    Returns:
        Datetime object representing last modification time
    """
    return datetime.datetime.fromtimestamp(self.path.stat().st_mtime)

NonExistingDirectory

Bases: type

Directory metaclass to allow create non-existing Directory instances.

Source code in datamate/directory.py

class NonExistingDirectory(type):
    """Directory metaclass to allow create non-existing Directory instances."""

    def __call__(cls, *args, **kwargs):
        return cls.__new__(cls, *args, **kwargs)