monkey_wrench.task.files module

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

class monkey_wrench.task.files.FilesTaskBase(*, 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.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, reference_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, reference: list[~monkey_wrench.input_output._models.InputType] | set[~monkey_wrench.input_output._models.InputType] | tuple[~monkey_wrench.input_output._models.InputType, ...] | ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=file), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)] | ~monkey_wrench.input_output._models.DirectoryVisitor | None = None, 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, 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, DirectoryVisitor, FilesIntegrityValidator

Pydantic model for the specifications of a verification task.

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.

classmethod validate_filepath_transform_function(data: Any) → Any

Ensure that the filepath transform function is set to a default value if it is not given explicitly.

class monkey_wrench.task.files.FetchFilesSpecifications(*, temp_directory_path: ~typing.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=dir), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.input_output._types.<lambda>)], 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', 'blockcache'] | None = None, 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'}, 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, output_filename_generator: ~typing.Annotated[~typing.Callable[[...], ~pathlib.Path], ~pydantic.functional_validators.BeforeValidator(func=~monkey_wrench.generic.models._function.validate_function_path, json_schema_input_type=PydanticUndefined)] | ~typing.Callable[[str], ~pathlib.Path] = <function input_filename_from_product_id>, channel_names: list[str] = ['VIS006', 'VIS008', 'IR_016', 'IR_039', 'WV_062', 'WV_073', 'IR_087', 'IR_097', 'IR_108', 'IR_120', 'IR_134', 'HRV'], 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, TempDirectory

Pydantic model for the specifications of a fetch task.

class monkey_wrench.task.files.VerifyFiles(*, 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(*, 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.