monkey_wrench.task.files module

Module to define Pydantic models for tasks related to product files.

class monkey_wrench.task.files.FilesTaskBase(*, verbose: bool = False, context: Literal[Context.product_files], action: Action, specifications: type[Model])

Bases: TaskBase

Pydantic base model for tasks related to product files.

context: Literal[Context.product_files]

class monkey_wrench.task.files.VerifyFilesSpecifications(*, number_of_processes: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 1, nominal_file_size: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] | None = None, file_size_relative_tolerance: ~typing.Annotated[float, ~annotated_types.Ge(ge=0)] = 0.01, filepath_transform_function: ~typing.Annotated[~typing.Callable[[...], ~monkey_wrench.generic.models._function.ReturnType], ~pydantic.functional_validators.BeforeValidator(func=~monkey_wrench.generic.models._function.validate_function_path, json_schema_input_type=PydanticUndefined)] | ~typing.Callable[[...], ~monkey_wrench.generic.models._function.ReturnType], filepaths: ~typing.Annotated[list[~monkey_wrench.generic._types.ElementType] | set[~monkey_wrench.generic._types.ElementType] | tuple[~monkey_wrench.generic._types.ElementType, ...] | ~monkey_wrench.input_output._models.Reader | ~monkey_wrench.input_output._models.DirectoryVisitor, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._models.validate_items)], reference: ~typing.Annotated[list[~monkey_wrench.generic._types.ElementType] | set[~monkey_wrench.generic._types.ElementType] | tuple[~monkey_wrench.generic._types.ElementType, ...] | ~monkey_wrench.input_output._models.Reader | ~monkey_wrench.input_output._models.DirectoryVisitor, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._models.validate_items)] | None = None, end_datetime: ~typing.Annotated[~pydantic.types.AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, start_datetime: ~typing.Annotated[~pydantic.types.AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, verbose: list[~typing.Literal['files', 'reference', 'corrupted', 'missing']] | bool = False)

Bases: DateTimePeriodStrict, FilesIntegrityValidator

Pydantic model for the specifications of a verification task.

filepath_transform_function: Annotated[Callable[[...], ReturnType], BeforeValidator(func=validate_function_path, json_schema_input_type=PydanticUndefined)] | Callable[[...], ReturnType]

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 to None which means no transformation is performed on the file paths and they will be used as they are.

verbose: list[Literal['files', 'reference', 'corrupted', 'missing']] | bool

Determines whether the given fields should be reported verbosely, i.e. the actual items will be dumped to the std output instead of only the number of items (for the non-verbose mode). It can be a list of field names or a single boolean value to change the behaviour for all fields at once.

classmethod validate_verbose(data: Any) → Any

Convert the verbose to a list of field names.

class monkey_wrench.task.files.FetchFilesSpecifications(*, 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), fsspec_cache: ~typing.Literal['filecache'] = 'filecache', 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, dataset_save_options: dict[str, bool | str | int] = {'include_lonlats': False, 'writer': 'cf'}, collection: ~monkey_wrench.query._types.EumetsatCollection | str, area: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)] | dict[str, ~typing.Any] | ~pyresample.geometry.AreaDefinition, radius_of_influence: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 20000, remove_file_if_exists: bool = True, number_of_processes: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 1, end_datetime: ~typing.Annotated[~pydantic.types.AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, start_datetime: ~typing.Annotated[~pydantic.types.AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None)

Bases: DateTimePeriodStrict, MultiProcess, Resampler, Reader

Pydantic model for the specifications of a fetch task.

class monkey_wrench.task.files.VerifyFiles(*, verbose: bool = False, context: Literal[Context.product_files], action: Literal[Action.verify], specifications: VerifyFilesSpecifications)

Bases: FilesTaskBase

Pydantic model for the verification task.

action: Literal[Action.verify]

specifications: VerifyFilesSpecifications

__verbose_or_not(field: list[ElementType] | set[ElementType], field_name: str) → Annotated[int, Ge(ge=0)] | list[ElementType] | set[ElementType]

perform() → dict[str, Annotated[int, Ge(ge=0)]]

Verify the product files using the reference.

class monkey_wrench.task.files.FetchFiles(*, verbose: bool = False, context: Literal[Context.product_files], action: Literal[Action.fetch], specifications: FetchFilesSpecifications)

Bases: FilesTaskBase

Pydantic model for the fetch task.

action: Literal[Action.fetch]

specifications: FetchFilesSpecifications

perform() → None

Fetch the product files.