import os
import uuid
from abc import abstractmethod
from collections.abc import Callable
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import List, Optional, Protocol
[docs]
@dataclass
class PathInfo:
"""
Information about where and how to write a file.
The bluesky event model splits the URI for a resource into two segments to aid in
different applications mounting filesystems at different mount points.
The portion of this path which is relevant only for the writer is the 'root',
while the path from an agreed upon mutual mounting is the resource_path.
The resource_dir is used with the filename to construct the resource_path.
:param root: Path of a root directory, relevant only for the file writer
:param resource_dir: Directory into which files should be written, relative to root
:param filename: Base filename to use generated by FilenameProvider, w/o extension
:param create_dir_depth: Optional depth of directories to create if they do not
exist
"""
root: Path
resource_dir: Path
filename: str
create_dir_depth: int = 0
[docs]
class FilenameProvider(Protocol):
[docs]
@abstractmethod
def __call__(self) -> str:
"""Get a filename to use for output data, w/o extension"""
[docs]
class PathProvider(Protocol):
_filename_provider: FilenameProvider
[docs]
@abstractmethod
def __call__(self, device_name: Optional[str] = None) -> PathInfo:
"""Get the current directory to write files into"""
[docs]
class StaticFilenameProvider(FilenameProvider):
def __init__(self, filename: str):
self._static_filename = filename
[docs]
def __call__(self) -> str:
return self._static_filename
[docs]
class UUIDFilenameProvider(FilenameProvider):
def __init__(
self,
uuid_call_func: Callable = uuid.uuid4,
uuid_call_args: Optional[List] = None,
):
self._uuid_call_func = uuid_call_func
self._uuid_call_args = uuid_call_args or []
[docs]
def __call__(self) -> str:
if (
self._uuid_call_func in [uuid.uuid3, uuid.uuid5]
and len(self._uuid_call_args) < 2
):
raise ValueError(
f"To use {self._uuid_call_func} to generate UUID filenames,"
" UUID namespace and name must be passed as args!"
)
uuid_str = self._uuid_call_func(*self._uuid_call_args)
return f"{uuid_str}"
[docs]
class AutoIncrementFilenameProvider(FilenameProvider):
def __init__(
self,
base_filename: str = "",
max_digits: int = 5,
starting_value: int = 0,
increment: int = 1,
inc_delimeter: str = "_",
):
self._base_filename = base_filename
self._max_digits = max_digits
self._current_value = starting_value
self._increment = increment
self._inc_delimeter = inc_delimeter
[docs]
def __call__(self):
if len(str(self._current_value)) > self._max_digits:
raise ValueError(
f"Auto incrementing filename counter \
exceeded maximum of {self._max_digits} digits!"
)
padded_counter = f"{self._current_value:0{self._max_digits}}"
filename = f"{self._base_filename}{self._inc_delimeter}{padded_counter}"
self._current_value += self._increment
return filename
[docs]
class StaticPathProvider(PathProvider):
def __init__(
self,
filename_provider: FilenameProvider,
directory_path: Path,
resource_dir: Path = Path("."),
create_dir_depth: int = 0,
) -> None:
self._filename_provider = filename_provider
self._directory_path = directory_path
self._resource_dir = resource_dir
self._create_dir_depth = create_dir_depth
[docs]
def __call__(self, device_name: Optional[str] = None) -> PathInfo:
filename = self._filename_provider()
return PathInfo(
root=self._directory_path,
resource_dir=self._resource_dir,
filename=filename,
create_dir_depth=self._create_dir_depth,
)
[docs]
class AutoIncrementingPathProvider(PathProvider):
def __init__(
self,
filename_provider: FilenameProvider,
directory_path: Path,
create_dir_depth: int = 0,
max_digits: int = 5,
starting_value: int = 0,
num_calls_per_inc: int = 1,
increment: int = 1,
inc_delimeter: str = "_",
base_name: str = None,
) -> None:
self._filename_provider = filename_provider
self._directory_path = directory_path
self._create_dir_depth = create_dir_depth
self._base_name = base_name
self._starting_value = starting_value
self._current_value = starting_value
self._num_calls_per_inc = num_calls_per_inc
self._inc_counter = 0
self._max_digits = max_digits
self._increment = increment
self._inc_delimeter = inc_delimeter
[docs]
def __call__(self, device_name: Optional[str] = None) -> PathInfo:
filename = self._filename_provider()
padded_counter = f"{self._current_value:0{self._max_digits}}"
resource_dir = str(padded_counter)
if self._base_name is not None:
resource_dir = f"{self._base_name}{self._inc_delimeter}{padded_counter}"
elif device_name is not None:
resource_dir = f"{device_name}{self._inc_delimeter}{padded_counter}"
self._inc_counter += 1
if self._inc_counter == self._num_calls_per_inc:
self._inc_counter = 0
self._current_value += self._increment
return PathInfo(
root=self._directory_path,
resource_dir=resource_dir,
filename=filename,
create_dir_depth=self._create_dir_depth,
)
[docs]
class YMDPathProvider(PathProvider):
def __init__(
self,
filename_provider: FilenameProvider,
directory_path: Path,
create_dir_depth: int = -3, # Default to -3 to create YMD dirs
device_name_as_base_dir: bool = False,
) -> None:
self._filename_provider = filename_provider
self._directory_path = Path(directory_path)
self._create_dir_depth = create_dir_depth
self._device_name_as_base_dir = device_name_as_base_dir
[docs]
def __call__(self, device_name: Optional[str] = None) -> PathInfo:
sep = os.path.sep
current_date = date.today().strftime(f"%Y{sep}%m{sep}%d")
if device_name is None:
resource_dir = current_date
elif self._device_name_as_base_dir:
resource_dir = os.path.join(
current_date,
device_name,
)
else:
resource_dir = os.path.join(
device_name,
current_date,
)
filename = self._filename_provider()
return PathInfo(
root=self._directory_path,
resource_dir=resource_dir,
filename=filename,
create_dir_depth=self._create_dir_depth,
)
[docs]
class NameProvider(Protocol):
[docs]
@abstractmethod
def __call__(self) -> str:
"""Get the name to be used as a data_key in the descriptor document"""
[docs]
class ShapeProvider(Protocol):
[docs]
@abstractmethod
async def __call__(self) -> tuple:
"""Get the shape of the data collection"""
