monkey_wrench.input_output package
The package providing utilities for input and output operations.
- class monkey_wrench.input_output.DateTimeDirectory(*, parent_output_directory_path: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)], datetime_format_string: str = '%Y/%m/%d', reset_child_datetime_directory: bool = False)[source]
Bases:
ParentOutputDirectoryPydantic model for datetime directories needed to store products and the input/output of CHIMP.
- datetime_format_string: str
The format string to create subdirectories from the datetime object. Defaults to
"%Y/%m/%d".
- reset_child_datetime_directory: bool
Whether to remove the (child) directory first if it already exists. Defaults to
False.This might save us from issues regarding files being overwritten and corrupted.
- get_datetime_directory(datetime_object: datetime) Path[source]
Get the full path to the datetime directory (given the datetime object). This does not create the directory.
- Parameters:
datetime_object – The datetime object for which the full directory path will be returned.
- Returns:
The full path of the datetime directory.
Example
>>> path = DateTimeDirectory( ... datetime_format_string="%Y/%m/%d", ... parent_output_directory_path=Path.home() ... ).get_datetime_directory( ... datetime(2022, 3, 12) ... ) >>> expected_path = Path.home() / Path("2022/03/12") >>> expected_path == path True
- create_datetime_directory(datetime_object: datetime) Path[source]
Create a directory based on the datetime object.
- Parameters:
datetime_object – The datetime object to create the directory for.
- Returns:
The full path of the (created) directory.
Example
>>> path = DateTimeDirectory( ... datetime_format_string="%Y/%m/%d", ... parent_output_directory_path=Path.home() ... ).create_datetime_directory( ... datetime(2022, 3, 12) ... ) >>> expected_path = Path.home() / Path("2022/03/12") >>> expected_path.exists() True >>> expected_path == path True
- class monkey_wrench.input_output.DatasetSaveOptions(*, dataset_save_options: dict[str, bool | str | int] = {'include_lonlats': False, 'writer': 'cf'})[source]
Bases:
ModelPydantic model for the storage options using which the dataset is to be saved. This is dataset-dependent.
- dataset_save_options: dict[str, bool | str | int]
A dictionary which includes the actual storage options.
The default behaviour is to use
cfas the writer and exclude longitude and latitude values, i.e.dataset_save_options = dict(writer="cf", include_lonlats=False)
- class monkey_wrench.input_output.DirectoryVisitor(*, negate: bool = False, sub_strings: str | list[str] | None = None, case_sensitive: bool = True, match_all: bool = True, parent_input_directory_path: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)], visitor_writer: ~monkey_wrench.input_output._models.Writer | None = None, visitor_callback: ~typing.Annotated[~typing.Callable[[...], ~typing.Any], ~pydantic.functional_validators.BeforeValidator(func=~monkey_wrench.generic.models._function.validate_function_path, json_schema_input_type=PydanticUndefined)] | ~typing.Callable[[...], ~typing.Any] | None = None, reverse: bool = False, recursive: bool = True, post_visit_transform_function: ~typing.Annotated[~typing.Callable[[...], ~monkey_wrench.input_output._models.ReturnType], ~pydantic.functional_validators.BeforeValidator(func=~monkey_wrench.generic.models._function.validate_function_path, json_schema_input_type=PydanticUndefined)] | ~typing.Callable[[...], ~monkey_wrench.input_output._models.ReturnType] | None = None)[source]
Bases:
ParentInputDirectory,PatternPydantic model for visiting files in a directory tree.
- visitor_writer: Writer | None
If given, it will be used to write the list of visited files to a text file.
- visitor_callback: Annotated[Callable[[...], Any], BeforeValidator(func=validate_function_path, json_schema_input_type=PydanticUndefined)] | Callable[[...], Any] | None
A function that will be called every time a match is found for a file. Defaults to
None.
- reverse: bool
A boolean to determine whether to sort the files in reverse order.
Defaults to
False, which means sorting is in the alphabetical order.
- recursive: bool
Determines whether to recursively visit the directory tree. or just visit the top-level directory.
Defaults to
True.
- post_visit_transform_function: Annotated[Callable[[...], ReturnType], BeforeValidator(func=validate_function_path, json_schema_input_type=PydanticUndefined)] | Callable[[...], ReturnType] | None
The transform function that will be applied on filepaths after visiting them.
Defaults to
None, which means no transformation is applied.Note
If it is provided, the result of transformation will be returned instead of filepaths.
- __collect_files() list[Path]
- visit() list[ReturnType] | list[Path][source]
Visit all files in the directory, either recursively or just the top-level files.
- Returns:
A sorted flat list of all file paths in the given directory that match the given pattern and have been treated according to the
visitor_callbackfunction. If thepost_visit_transform_functionis provided , a list of transformed filepaths will be returned instead.
- class monkey_wrench.input_output.ExistingInputFile(*, input_filepath: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)])[source]
Bases:
ModelPydantic model for an input file which must exist.
Example
A text file which includes the list of product IDs which have been already fetched. This file will be used to fetch the product files.
- input_filepath: ensure_path_does_not_end_with_slash)]
- class monkey_wrench.input_output.FilesIntegrityValidator(*, number_of_processes: Annotated[int, Ge(ge=0)] = 1, nominal_file_size: Annotated[int, Ge(ge=0)] | None = None, file_size_relative_tolerance: Annotated[float, Ge(ge=0)] = 0.01, filepath_transform_function: Annotated[Callable[[...], ReturnType], BeforeValidator(func=validate_function_path, json_schema_input_type=PydanticUndefined)] | Callable[[...], ReturnType] | None = None, filepaths: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)], reference: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)] | None = None)[source]
Bases:
MultiProcessPydantic model to verify files integrity by checking their size as well as comparing them against a reference.
Note
This class does two main verifications, namely checking for corrupted and missing files as follows
1- Checking that the file sizes are within some threshold from a nominal file size. 2- Checking filepaths against a reference collection.
- nominal_file_size: Annotated[int, Ge(ge=0)] | None
The nominal size of files in bytes. This is used to check for corrupted files.
Defaults to
None, which means the search for corrupted files will not be performed.
- file_size_relative_tolerance: Annotated[float, Ge(ge=0)]
The maximum relative difference in the size of a file, before it can be marked as corrupted.
Defaults to
0.01, i.e. any file whose size differs by more than one percent from the nominal size, will be marked as corrupted.
- filepath_transform_function: Annotated[Callable[[...], ReturnType], BeforeValidator(func=validate_function_path, json_schema_input_type=PydanticUndefined)] | Callable[[...], ReturnType] | None
A function to transform the file paths into other types of objects before comparing them against the reference.
This can be e.g. a
parse()function to make datetime objects out of file paths. Defaults toNonewhich means no transformation is performed on the file paths and they will be used as they are.
- filepaths: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)]
The file paths to perform the validation on.
- reference: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)] | None
Reference items to compare against, used in finding the missing files.
Defaults to
Nonewhich means the search for missing files will not be performed.
- find_corrupted_files(filepaths: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)] | None = None) set[Path] | None[source]
- find_missing_files(filepaths: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)] | None = None, reference: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)] | None = None) set[Path] | None[source]
- verify_files(filepaths: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)] | None = None, reference: Annotated[list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Reader | DirectoryVisitor, AfterValidator(func=validate_items)] | None = None) tuple[set[InputType] | set[ReturnType] | None, set[Path] | None][source]
Check for missing and corrupted files.
- class monkey_wrench.input_output.FsSpecCache(*, fsspec_cache: Literal['filecache'] = 'filecache')[source]
Bases:
ModelPydantic model for the caching scheme of fsspec.
Note
See fsspec cache, to learn more about buffering and random access in fsspec.
- fsspec_cache: Literal['filecache']
How to buffer, at the moment the only supported option and the default is
"filecache".
- property fsspec_cache_str
Return the cache string with a leading
::.
- class monkey_wrench.input_output.ExistingInputDirectory(*, input_directory: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)])[source]
Bases:
ModelPydantic model for an input directory which must exist.
Note
This model is to be solely used for a flat structure. If you have a hierarchical tree structure, use
ParentInputDirectoryinstead to be more clear about the directory structure.- input_directory: <lambda>)]
- class monkey_wrench.input_output.InputFile(*, input_filepath: Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)] | ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=new), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)] | None = None)[source]
Bases:
ModelPydantic model for an input file which does not necessarily exist during the model validation.
- input_filepath: ensure_path_does_not_end_with_slash)] | None
- class monkey_wrench.input_output.ModelFile(*, model_filepath: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)])[source]
Bases:
ModelPydantic model for a model file which must exist.
Example
A
*.ptfile used by CHIMP, as the model, to perform a retrieval.- model_filepath: ensure_path_does_not_end_with_slash)]
- class monkey_wrench.input_output.NewOutputFile(*, output_filepath: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=new), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)])[source]
Bases:
ModelPydantic mode for an output file which must not already exist.
Example
A text file to store the result of visiting a directory, i.e. collected files that match the determined pattern.
- output_filepath: ensure_path_does_not_end_with_slash)]
- class monkey_wrench.input_output.ExistingOutputDirectory(*, output_directory: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)])[source]
Bases:
ModelPydantic model for an output directory which must exist.
Note
This model is to be solely used for a flat structure. If you have a hierarchical tree structure, use
ParentOutputDirectoryinstead to be more clear about the directory structure.- output_directory: <lambda>)]
- class monkey_wrench.input_output.OutputFile(*, output_filepath: Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)] | ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=new), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)] | None = None)[source]
Bases:
ModelPydantic model for an output file which does not necessarily exist during the model validation.
- output_filepath: ensure_path_does_not_end_with_slash)] | None
- class monkey_wrench.input_output.Reader(*, input_filepath: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)], post_reading_transformation: ~monkey_wrench.generic.models._pattern.StringTransformation = StringTransformation(trim=True, transform_function=None))[source]
Bases:
ExistingInputFilePydantic model for an ASCII file (text mode) reader.
- post_reading_transformation: StringTransformation
The transformation after reading items from the file and before returning them.
Defaults to
StringTransformation(), which means the items will be only trimmed.Note
The items will be first trimmed and then transformed according to
post_reading_transformation.transform_function.
- read() list[Any][source]
Read items from a text file, assuming each line corresponds to a single item.
Examples of items are product IDs.
Warning
This function does not check whether the items are valid or not. It is a simple convenience function for reading items from a text file.
- Returns:
A list of (transformed) items, where each item corresponds to a single line in the given file.
- class monkey_wrench.input_output.ParentInputDirectory(*, parent_input_directory_path: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)])[source]
Bases:
ModelPydantic model for the top-level directory where the child (input) directories reside. The directory must exist.
Example
A directory which includes all SEVIRI files that have to be reprocessed using CHIMP.
- parent_input_directory_path: <lambda>)]
- class monkey_wrench.input_output.ParentOutputDirectory(*, parent_output_directory_path: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)])[source]
Bases:
ModelPydantic model for the top-level directory where the child (output) directories reside. The directory must exist.
Example
A directory which the output of CHIMP will be saved.
- parent_output_directory_path: <lambda>)]
- class monkey_wrench.input_output.Writer(*, output_filepath: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)] | ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=new), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.ensure_path_does_not_end_with_slash)] | None = None, open_mode: ~typing.Literal['w', 'a'] = 'w', pre_writing_transformation: ~monkey_wrench.generic.models._pattern.StringTransformation = StringTransformation(trim=True, transform_function=None), on_write_catch_exceptions: tuple[type[Exception], ...] | None = None)[source]
Bases:
OutputFilePydantic model for an ASCII file (text mode) writer.
- open_mode: Literal['w', 'a']
The mode using which the text file will be opened. Defaults to
"w".
- pre_writing_transformation: StringTransformation
The transformation before writing items to the file.
Defaults to StringTransformation, which means the items will be only trimmed.
Note
The items will first be transformed according to the
pre_writing_transformation.transform_function, and then trimmed.
- on_write_catch_exceptions: tuple[type[Exception], ...] | None
Exceptions which will be caught and logged as warnings.
Defaults to
None, which means all exceptions will be caught. If it is a tuple, only the given exceptions will be caught. As a result, in the case of an empty tuple, no exceptions will be caught.
- __get_open_mode(open_mode: Literal['w', 'a'] | None = None)
Return the
open_modeif it is notNone. Otherwise return the value ofself.open_mode.
- prepare_output_file_for_writing(open_mode: Literal['w', 'a'] | None = None) None[source]
Prepare the output file depending on whether the file exists and the value of
open_mode.Note
If the file exists and
open_modeis"a", the file will be left as it is.If the file exists and
open_modeis"w", the file will be overwritten.If the file does not exist, the file will be created, regardless of
open_mode.
- Parameters:
open_mode – Defaults to
None, which means the value fromself.open_modewill be used.
- write(items: list[ElementType] | set[ElementType] | tuple[ElementType, ...] | Generator[Any, None, None], open_mode: Literal['w', 'a'] | None = None) Annotated[int, Ge(ge=0)][source]
Write items from a list/set/tuple or a generator to a text file, with one item per line.
Examples of items are product IDs.
This function opens a text file in the (over)write or append mode. It then writes each item from the provided iterable to the file. It catches specified exceptions during the writing process, and logs them as warnings.
- Parameters:
items – An iterable of items to be written to the file.
open_mode – Defaults to
None, which means the value fromself.open_modewill be used.
- Returns:
The number of items that are written to the file successfully.
- write_in_batches(batches: Generator[tuple[T, int], None, None], open_mode: Literal['w', 'a'] | None = None) Annotated[int, Ge(ge=0)][source]
Similar to Writer.write(), but assumes that the input is in batches.
- monkey_wrench.input_output.copy_files_between_directories(source_directory: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)], destination_directory: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)], pattern: ~monkey_wrench.generic.models._pattern.Pattern | None = None) list[Path][source]
Copy (top-level) files whose names include the pattern from one directory to another.
Warning
The copying is not performed recursively. Only the top-level files are copied.
- Parameters:
source_directory – The source directory to copy files from.
destination_directory – The destination directory to copy files to.
pattern – The pattern to filter the files.
- Returns:
The list of filepaths that have been copied.
- monkey_wrench.input_output.copy_single_file_to_directory(destination_directory: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)], filepath: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)]) None[source]
Copy a single file with the given path to another destination directory.
- Parameters:
destination_directory – The destination directory to copy the given file to.
filepath – The path of the file that needs to be copied.
- monkey_wrench.input_output.datetime_to_filename(prefix: str, datetime_object: datetime, extension: str = '.nc') Path[source]
Generate a CHIMP-compliant filename based on the datetime object and the given prefix.
- Parameters:
prefix – A string with which the filename will start.
datetime_object – The datetime object to retrieve the timestamp string from.
extension – The file extension, Defaults to
".nc".
- Returns:
A filename with the following format
"<prefix>_<year><month><day>_<hour>_<minute><extension>".
- monkey_wrench.input_output.output_filename_from_datetime(datetime_objects: datetime | list[datetime] | set[datetime] | tuple[datetime, ...], extension: str = '.nc') Path | list[Path] | set[Path] | tuple[Path, ...][source]
Generate (a) CHIMP-compliant output filename(s) based on (a) datetime object(s).
- Parameters:
datetime_objects – Either a single datetime object , or a list/set/tuple of datetime objects.
extension – The file extension, Defaults to
".nc".
- Returns:
Depending on the input, either a single filename, or a list/set/tuple of filenames. The type of the output matches the type of the input in case of a list/set.tuple, e.g. a tuple of strings as input will result in a tuple of paths.
Example
>>> output_filename_from_datetime(datetime(2020, 1, 1, 0, 12)) PosixPath('chimp_20200101_00_12.nc')
>>> output_filename_from_datetime( ... [datetime(2020, 1, 1, 0, 12), datetime(2020, 3, 4, 2, 42)] ... ) [PosixPath('chimp_20200101_00_12.nc'), PosixPath('chimp_20200304_02_42.nc')]
Subpackages
- monkey_wrench.input_output.fci package
- monkey_wrench.input_output.hrit package
- monkey_wrench.input_output.resampler package
- monkey_wrench.input_output.seviri package
Submodules
- monkey_wrench.input_output._common module
- monkey_wrench.input_output._models module
ExistingInputFileInputFileNewOutputFileOutputFileModelFileParentInputDirectoryParentOutputDirectoryExistingInputDirectoryExistingOutputDirectoryFsSpecCacheDatasetSaveOptionsReaderDirectoryVisitorvalidate_items()ItemsFilesIntegrityValidatorFilesIntegrityValidator.nominal_file_sizeFilesIntegrityValidator.file_size_relative_toleranceFilesIntegrityValidator.filepath_transform_functionFilesIntegrityValidator.filepathsFilesIntegrityValidator.referenceFilesIntegrityValidator.file_is_corrupted()FilesIntegrityValidator.find_corrupted_files()FilesIntegrityValidator.find_missing_files()FilesIntegrityValidator.verify_files()
DateTimeDirectory
- monkey_wrench.input_output._types module