utils
sleap_io.io.utils
¶
Miscellaneous utilities for working with different I/O formats.
Functions:
| Name | Description |
|---|---|
is_file_accessible |
Check if a file is accessible. |
read_hdf5_attrs |
Read attributes from an HDF5 dataset. |
read_hdf5_dataset |
Read data from an HDF5 file. |
read_hdf5_group |
Read an entire group from an HDF5 file. |
sanitize_filename |
Sanitize a filename to a canonical posix-compatible format. |
write_hdf5_attrs |
Write attributes to an HDF5 dataset. |
write_hdf5_dataset |
Write data to an HDF5 file. |
write_hdf5_group |
Write an entire group to an HDF5 file. |
Attributes:
| Name | Type | Description |
|---|---|---|
__cached__ |
str(object='') -> str |
|
__doc__ |
str(object='') -> str |
|
__file__ |
str(object='') -> str |
|
__name__ |
str(object='') -> str |
|
__package__ |
str(object='') -> str |
__cached__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/__pycache__/utils.cpython-313.pyc'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__doc__ = 'Miscellaneous utilities for working with different I/O formats.'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__file__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/utils.py'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__name__ = 'sleap_io.io.utils'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__package__ = 'sleap_io.io'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
is_file_accessible(filename)
¶
Check if a file is accessible.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str | Path
|
Path to a file. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
Notes
This checks if the file readable by the current user by reading one byte from the file.
Source code in sleap_io/io/utils.py
def is_file_accessible(filename: str | Path) -> bool:
"""Check if a file is accessible.
Args:
filename: Path to a file.
Returns:
`True` if the file is accessible, `False` otherwise.
Notes:
This checks if the file readable by the current user by reading one byte from
the file.
"""
filename = Path(filename)
try:
with open(filename, "rb") as f:
f.read(1)
return True
except (FileNotFoundError, PermissionError, OSError, ValueError):
return False
read_hdf5_attrs(filename, dataset='/', attribute=None, *, _hdf5_file=None)
¶
Read attributes from an HDF5 dataset.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Path to an HDF5 file. |
required |
dataset
|
str
|
Path to a dataset or group from which attributes will be read. |
'/'
|
attribute
|
str | None
|
If specified, the attribute name to read. If |
None
|
_hdf5_file
|
File | None
|
An already-open |
None
|
Returns:
| Type | Description |
|---|---|
Any | dict[str, Any]
|
The attributes in a dictionary, or the attribute field if |
Source code in sleap_io/io/utils.py
def read_hdf5_attrs(
filename: str,
dataset: str = "/",
attribute: str | None = None,
*,
_hdf5_file: h5py.File | None = None,
) -> Any | dict[str, Any]:
"""Read attributes from an HDF5 dataset.
Args:
filename: Path to an HDF5 file.
dataset: Path to a dataset or group from which attributes will be read.
attribute: If specified, the attribute name to read. If `None` (the default),
all attributes for the dataset will be returned.
_hdf5_file: An already-open `h5py.File` handle to read from. If provided,
the data is read from this handle (which is left open for the caller
to close); otherwise `filename` is opened and closed internally. This
is a private argument used to thread a single open handle (e.g. an
fsspec-backed remote file) through multiple reads.
Returns:
The attributes in a dictionary, or the attribute field if `attribute` was
provided.
"""
if _hdf5_file is not None:
return _read_attrs_from_open_file(_hdf5_file, dataset, attribute)
with h5py.File(filename, "r") as f:
return _read_attrs_from_open_file(f, dataset, attribute)
read_hdf5_dataset(filename, dataset, *, _hdf5_file=None)
¶
Read data from an HDF5 file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Path to an HDF5 file. |
required |
dataset
|
str
|
Path to a dataset. |
required |
_hdf5_file
|
File | None
|
An already-open |
None
|
Returns:
| Type | Description |
|---|---|
ndarray
|
The data as an array. If the dataset is a flat 2D array with a
|
Source code in sleap_io/io/utils.py
def read_hdf5_dataset(
filename: str,
dataset: str,
*,
_hdf5_file: h5py.File | None = None,
) -> np.ndarray:
"""Read data from an HDF5 file.
Args:
filename: Path to an HDF5 file.
dataset: Path to a dataset.
_hdf5_file: An already-open `h5py.File` handle to read from. If provided,
the data is read from this handle (which is left open for the caller
to close); otherwise `filename` is opened and closed internally. This
is a private argument used to thread a single open handle (e.g. an
fsspec-backed remote file) through multiple reads.
Returns:
The data as an array. If the dataset is a flat 2D array with a
``"field_names"`` JSON attribute (as written by h5wasm), it will be
converted to a structured array so that field-name access works
identically to compound-dtype datasets written by h5py.
"""
if _hdf5_file is not None:
return _read_dataset_from_open_file(_hdf5_file, dataset)
with h5py.File(filename, "r") as f:
return _read_dataset_from_open_file(f, dataset)
read_hdf5_group(filename, group='/', *, _hdf5_file=None)
¶
Read an entire group from an HDF5 file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Path an HDF5 file. |
required |
group
|
str
|
Path to a group within the HDF5 file. Defaults to "/" (read the entire file). |
'/'
|
_hdf5_file
|
File | None
|
An already-open |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, ndarray]
|
A flat dictionary with keys corresponding to dataset paths and values corresponding to the datasets as arrays. |
Source code in sleap_io/io/utils.py
def read_hdf5_group(
filename: str,
group: str = "/",
*,
_hdf5_file: h5py.File | None = None,
) -> dict[str, np.ndarray]:
"""Read an entire group from an HDF5 file.
Args:
filename: Path an HDF5 file.
group: Path to a group within the HDF5 file. Defaults to "/" (read the entire
file).
_hdf5_file: An already-open `h5py.File` handle to read from. If provided,
the data is read from this handle (which is left open for the caller
to close); otherwise `filename` is opened and closed internally. This
is a private argument used to thread a single open handle (e.g. an
fsspec-backed remote file) through multiple reads.
Returns:
A flat dictionary with keys corresponding to dataset paths and values
corresponding to the datasets as arrays.
"""
if _hdf5_file is not None:
return _read_group_from_open_file(_hdf5_file, group)
with h5py.File(filename, "r") as f:
return _read_group_from_open_file(f, group)
sanitize_filename(filename)
¶
Sanitize a filename to a canonical posix-compatible format.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str | Path | list[str] | list[Path]
|
A string or |
required |
Returns:
| Type | Description |
|---|---|
str | list[str]
|
A sanitized filename as a string (or list of strings if a list was provided) with forward slashes and posix-formatted. |
Notes
URLs (e.g. https://..., s3://...) are returned unchanged. Passing a
URL through Path would collapse the // after the scheme (turning
https://host/x into https:/host/x), corrupting the URL.
Source code in sleap_io/io/utils.py
def sanitize_filename(
filename: str | Path | list[str] | list[Path],
) -> str | list[str]:
"""Sanitize a filename to a canonical posix-compatible format.
Args:
filename: A string or `Path` object or list of either to sanitize.
Returns:
A sanitized filename as a string (or list of strings if a list was provided)
with forward slashes and posix-formatted.
Notes:
URLs (e.g. ``https://...``, ``s3://...``) are returned unchanged. Passing a
URL through ``Path`` would collapse the ``//`` after the scheme (turning
``https://host/x`` into ``https:/host/x``), corrupting the URL.
"""
if isinstance(filename, list):
return [sanitize_filename(f) for f in filename]
from sleap_io.io._remote import _is_url
if isinstance(filename, str) and _is_url(filename):
return filename
return Path(filename).as_posix().replace("\\", "/")
write_hdf5_attrs(filename, dataset, attributes)
¶
Write attributes to an HDF5 dataset.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Path to an HDF5 file. |
required |
dataset
|
str
|
Path to a dataset or group to which attributes will be written. |
required |
attributes
|
dict[str, Any]
|
The attributes in a dictionary with the keys as the attribute names. |
required |
Source code in sleap_io/io/utils.py
def write_hdf5_attrs(filename: str, dataset: str, attributes: dict[str, Any]):
"""Write attributes to an HDF5 dataset.
Args:
filename: Path to an HDF5 file.
dataset: Path to a dataset or group to which attributes will be written.
attributes: The attributes in a dictionary with the keys as the attribute names.
"""
def _overwrite_hdf5_attr(
group_or_dataset: h5py.Group | h5py.Dataset, attr_name: str, data: Any
):
"""Overwrite attribute for group or dataset in HDF5 file.
Args:
group_or_dataset: Path to group or dataset in HDF5 file.
attr_name: Name of attribute.
data: Data to write to attribute.
"""
try:
del group_or_dataset.attrs[attr_name]
except KeyError:
pass
group_or_dataset.attrs.create(attr_name, data)
with h5py.File(filename, "a") as f: # "a": read/write if exists, create otherwise
ds = f[dataset]
for attr_name, attr_value in attributes.items():
_overwrite_hdf5_attr(ds, attr_name, attr_value)
write_hdf5_dataset(filename, dataset, data)
¶
Write data to an HDF5 file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Path to an HDF5 file. |
required |
dataset
|
str
|
Path to a dataset. |
required |
data
|
ndarray
|
Data to write to dataset. |
required |
Source code in sleap_io/io/utils.py
def write_hdf5_dataset(filename: str, dataset: str, data: np.ndarray):
"""Write data to an HDF5 file.
Args:
filename: Path to an HDF5 file.
dataset: Path to a dataset.
data: Data to write to dataset.
"""
with h5py.File(filename, "a") as f: # "a": read/write if exists, create otherwise
_overwrite_hdf5_dataset(f, dataset, data)
write_hdf5_group(filename, data)
¶
Write an entire group to an HDF5 file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str
|
Path an HDF5 file. |
required |
data
|
dict[str, ndarray]
|
A dictionary with keys corresponding to dataset/group paths and values corresponding to either sub group paths or the datasets as arrays. |
required |
Source code in sleap_io/io/utils.py
def write_hdf5_group(filename: str, data: dict[str, np.ndarray]):
"""Write an entire group to an HDF5 file.
Args:
filename: Path an HDF5 file.
data: A dictionary with keys corresponding to dataset/group paths and values
corresponding to either sub group paths or the datasets as arrays.
"""
def overwrite_hdf5_group(
file_or_group: h5py.File | h5py.Group, group_name: str
) -> h5py.Group:
"""Overwrite group in HDF5 file.
Args:
file_or_group: Path to an HDF5 file or parent group.
group_name: Path to a group.
Return:
group: (Sub-)group under specified file or parent group.
"""
try:
del file_or_group[group_name]
except KeyError:
pass
group = file_or_group.create_group(group_name)
return group
def write_group(parent_group, data_to_write):
for name, dataset_or_group in data_to_write.items():
if isinstance(dataset_or_group, dict):
# Create (sub-)group under parent group (top level being the file)
group = overwrite_hdf5_group(parent_group, name)
write_group(group, dataset_or_group) # Recall with new parent
else:
# Create dataset if dataset_or_group is a dataset
_overwrite_hdf5_dataset(
f=parent_group, dataset=name, data=dataset_or_group
)
with h5py.File(filename, "a") as f: # "a": read/write if exists, create otherwise
write_group(f, data)