process.blockwise
process.blockwise
Block processing framework for xarray datasets with lazy evaluation support.
This module provides a framework for processing large xarray datasets in chunks using lazy evaluation, leveraging xarray’s map_blocks functionality and dask’s chunked arrays.
Key components: - BlockProcessor: Abstract base class for defining individual processing operations that modify or create dataset variables within data blocks. - CompoundProcessor: Class for chaining multiple processors into optimized pipelines with flexible output selection
The framework supports: - Variable metadata management (dimensions, dtypes, attributes, flags) - Inputs and outputs description, with automatic validation - Performance optimization through processor chaining
Example
Create a simple processor that adds two variables:
class AddProcessor(BlockProcessor): … def input_vars(self): … return [Var(‘a’), Var(‘b’)] … … def created_vars(self): … return [Var(‘sum’, ‘float64’, (‘x’, ‘y’))] … … def process_block(self, block): … block[‘sum’] = block[‘a’] + block[‘b’] … result = AddProcessor().map_blocks(dataset)
Chain multiple processors:
compound = CompoundProcessor([processor1, processor2]) result = compound.map_blocks(dataset)
Classes
| Name | Description |
|---|---|
| BlockProcessor | Abstract base class for block processing operations. Processors are intended to be |
| CompoundProcessor | Combines multiple BlockProcessor instances into a single processor. |
BlockProcessor
process.blockwise.BlockProcessor()Abstract base class for block processing operations. Processors are intended to be executed by their method map_blocks. Use CompoundProcessor to combine multiple processors into a single processing pipeline and minimize dask graph management overhead.
Each processor defines: - process_block: Method that performs the actual processing on a data block, by adding or modifying Dataset variables in place. - input_vars: List of variable names required as input (optional) - modified_vars: List of Var objects describing the variables to be modified (optional) - created_vars: List of Var objects describing newly created variables (optional) - output_vars: List of Var objects describing the variables to be output by the map_blocks method (optional) - created_dims: Dictionary of new dimensions to be created by this processor (optional) - global_attrs: Dictionary of the global attributes set by this processor (optional) - auto_template: Method that returns whether automatic templating should be activated (optional) - check: Method to validate the input dataset before processing (optional)
Examples
>>> class AddProcessor(BlockProcessor):
... def input_vars(self):
... return [Var('a'), Var('b')]
...
... def created_vars(self):
... return [Var('sum', 'float64', ('x', 'y'))]
...
... def process_block(self, block, **kwargs):
... block['sum'] = block['a'] + block['b']
...
>>> result = AddProcessor().map_blocks(ds)Methods
| Name | Description |
|---|---|
| auto_template | Returns whether “automatic templating” should be activated |
| build_template | Modifies a template dataset for xr.map_blocks operations. |
| check | Validate the input dataset for this processor. |
| created_dims | Describes new dimensions to be created by this processor. |
| created_vars | Variables that this processor will create. |
| describe | Print a description of this processor for logging/debugging. |
| get_flag_mappings | Extract flag bit mappings (var -> flags) from modified_vars() and |
| global_attrs | Return dictionary of global attributes to be set by this processor. |
| input_vars | Input variables required by this processor. |
| map_blocks | Apply this processor to a dataset using xarray’s map_blocks. |
| modified_vars | Variables to be modified in-place. |
| output_vars | Defines the list of variables to be returned by map_blocks |
| process_and_validate | Process a block and validate the created output variables. |
| process_block | Process a single block of data (in place). |
| raiseflag | Apply a flag to a variable in the block where condition is True. |
| set_attributes | Set attributes defined in created/modified vars in ds |
| validate | Validate that all advertised created output variables exist with correct specs |
auto_template
process.blockwise.BlockProcessor.auto_template()Returns whether “automatic templating” should be activated
Automatic templating calls process_block on mocked up (empty) data to determine the output variables types and dimensions.
auto_template is disactivated by default.
build_template
process.blockwise.BlockProcessor.build_template(ds_template)Modifies a template dataset for xr.map_blocks operations.
This method builds a template dataset that defines the structure and metadata of the output dataset after processing. The template includes: - All output variables (created and modified) - Appropriate data types and dimensions for created variables - Flag metadata for variables with defined flags - Global attributes from the processor
If some dimensions are not fully described, the method runs the processing code on mockup data to assess the data types and dimensions of created variables.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| ds_template | xr.Dataset | Input template, modified in place. | required |
Returns
| Name | Type | Description |
|---|---|---|
| None | The input dataset is modified in place. |
check
process.blockwise.BlockProcessor.check(ds)Validate the input dataset for this processor.
This method can be overridden by subclasses to perform custom validation on the input dataset before processing. By default, it does nothing.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| ds | xr.Dataset | The input dataset to validate. | required |
created_dims
process.blockwise.BlockProcessor.created_dims()Describes new dimensions to be created by this processor.
Each dimension is defined either by: - An int: specifying the dimension size - An array-like: providing coordinates for the dimension
Returns
| Name | Type | Description |
|---|---|---|
| Dict[str, int or array - like] | Dictionary mapping dimension names to their size (int) or coordinates (array-like) |
Examples
>>> def created_dims(self):
... return {'z': 10, 'wavelength': [400, 500, 600, 700]}created_vars
process.blockwise.BlockProcessor.created_vars()Variables that this processor will create.
Provide Var definition: name, dtype, dims or dims_like (plus optionally flags or other attributes). Declare any new dims in created_dims(). If auto_template() returns True and dtype or dims/dims_like are not specified, they will be assessed by running process_block on mockup data.
Use dims_like to specify that the dimensions should match those of an existing variable in the dataset (e.g., dims_like='input_var').
Example: >>> def created_vars(self): … return [Var(‘sum’, ‘float64’, (‘x’, ‘y’))]
describe
process.blockwise.BlockProcessor.describe()Print a description of this processor for logging/debugging.
get_flag_mappings
process.blockwise.BlockProcessor.get_flag_mappings()Extract flag bit mappings (var -> flags) from modified_vars() and created_vars(). Results are cached.
global_attrs
process.blockwise.BlockProcessor.global_attrs()Return dictionary of global attributes to be set by this processor. For variable attributes, use the attrs argument of the created/modified variables.
input_vars
process.blockwise.BlockProcessor.input_vars()Input variables required by this processor.
Example: >>> def input_vars(self): … return [Var(‘a’), Var(‘b’), Var(‘flags’)]
map_blocks
process.blockwise.BlockProcessor.map_blocks(ds)Apply this processor to a dataset using xarray’s map_blocks.
This method validates that all required input and modified variables are present in the dataset, creates a subset containing only the necessary variables, builds a template dataset defining the output structure, and applies the processor’s process_block method to each chunk using xr.map_blocks.
The output includes only the variables specified by output_vars() (by default, created and modified variables).
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| ds | xr.Dataset | Input dataset to process. Must contain all variables required by input_vars() and modified_vars(). | required |
Returns
| Name | Type | Description |
|---|---|---|
| xr.Dataset | Processed dataset containing the output variables as defined by output_vars(). |
modified_vars
process.blockwise.BlockProcessor.modified_vars()Variables to be modified in-place.
Use Var with only name (leave dims/dtype as None). Optionally provide other attributes or flags to register flag metadata used by raiseflag().
Example: >>> def modified_vars(self): … return [Var(‘flags’, flags={‘CLOUD’: 1})]
Note: flags are raised like so: >>> def process_block(self, block, **kwargs): … self.raiseflag(block, ‘flags’, ‘CLOUD’, block[‘rho_toa’] > 0.2)
output_vars
process.blockwise.BlockProcessor.output_vars()Defines the list of variables to be returned by map_blocks
By default, the created and modified variables are included as output_vars.
This method can be overridden to adjust this behaviour.
process_and_validate
process.blockwise.BlockProcessor.process_and_validate(block)Process a block and validate the created output variables.
This method wraps the call to process_block, and is called by xr.map_map_blocks
process_block
process.blockwise.BlockProcessor.process_block(block)Process a single block of data (in place).
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| block | xr.Dataset | Input data block containing the required input variables. This block is modified in place. | required |
raiseflag
process.blockwise.BlockProcessor.raiseflag(
block,
var_name,
flag_name,
condition,
)Apply a flag to a variable in the block where condition is True.
This is a convenience wrapper around core.tools.raiseflag that automatically looks up the flag bit position from the processor’s flag definitions.
Note that the flags must be registered in the Var definition for the variable in modified_vars() or created_vars().
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| block | xr.Dataset | Dataset block containing the flag variable | required |
| var_name | str | Name of the flag variable (e.g., ‘flags’, ‘quality_flags’) | required |
| flag_name | str | Name of the flag to raise (e.g., ‘CLOUDY’, ‘BAD_DATA’) | required |
| condition | xr.DataArray | Boolean array indicating where to raise the flag | required |
Raises
| Name | Type | Description |
|---|---|---|
| ValueError | If the variable has no flags defined or the flag name is not defined |
Examples
>>> def modified_vars(self):
... return [Var('flags', flags={'CLOUDY': 1})]
>>> def process_block(self, block, **kwargs):
... cloud_mask = block['reflectance'] > 0.8
... self.raiseflag(block, 'flags', 'CLOUDY', cloud_mask)set_attributes
process.blockwise.BlockProcessor.set_attributes(ds)Set attributes defined in created/modified vars in ds
This method is called: - During the template creation - After each execution of process_block (in process_and_validate)
validate
process.blockwise.BlockProcessor.validate(block)Validate that all advertised created output variables exist with correct specs
CompoundProcessor
process.blockwise.CompoundProcessor(
list_processors,
outputs='created_modified',
outputs_tags=None,
outputs_names=None,
preserve_attrs=True,
)Combines multiple BlockProcessor instances into a single processor.
This class allows chaining multiple processors together, automatically filtering out processors that are not required to produce the specified outputs. It provides flexible output selection based on variable roles (created/modified/input) or custom criteria like tags or specific names.
The compound processor optimizes execution by only running processors that contribute to the final output, reducing unnecessary computation in processing pipelines.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| list_processors | list[BlockProcessor] | List of processors to combine in execution order | required |
| outputs | Literal['created_modified', 'all', 'tags', 'named'] | How to select output variables: - “created_modified”: output variables that are created or modified by the processors - “all”: output all variables (created, modified, and input variables) - “tags”: output variables that have tags matching outputs_tags - “named”: output variables with names matching outputs_names | "created_modified" |
| outputs_tags | list[str] | None | Required when outputs=“tags”. List of tags that variables must have to be included in the output. Variables are included if they have at least one matching tag. | None |
| outputs_names | list[str] | None | Required when outputs=“named”. List of variable names to include in the output. All specified names must exist in the processor chain. | None |
| preserve_attrs | bool | Whether to preserve global attributes from the input dataset in the output. | True |
Examples
Combine two processors with default output (created and modified variables):
>>> p1 = CloudMaskProcessor()
>>> p2 = QualityFlagProcessor()
>>> compound = CompoundProcessor([p1, p2])
>>> result = compound.map_blocks(dataset)Output only variables with specific tags:
>>> compound = CompoundProcessor([p1, p2],
... outputs="tags",
... outputs_tags=["quality"])
>>> result = compound.map_blocks(dataset)Output specific named variables:
>>> compound = CompoundProcessor([p1, p2], outputs="named",
... outputs_names=["cloud_mask", "quality_flags"])
>>> result = compound.map_blocks(dataset)Methods
| Name | Description |
|---|---|
| build_template | Modifies a template dataset for xr.map_blocks operations for compound processors. |
| created_vars | Return the list of variables created by the compound processor. |
| describe | Print a detailed description of the compound processor. |
| get_required_processors | Filter the processor chain to keep only those required to produce allowed_vars. |
| global_attrs | Safe merge of the global attributes from all processors |
| input_vars | Return the list of input variables for the compound processor. |
| map_blocks | Apply processors to dataset blocks with optional chaining. |
| modified_vars | Return the list of variables modified by the compound processor. |
| process_and_validate | Process a single block of data by applying all required processors, hereby |
build_template
process.blockwise.CompoundProcessor.build_template(ds_template)Modifies a template dataset for xr.map_blocks operations for compound processors.
This method builds a template dataset that defines the structure and metadata of the output dataset after processing by all processors in the compound. The template includes: - All output variables (created and modified) from all processors - Appropriate data types and dimensions for created variables - Flag metadata for variables with defined flags - Global attributes from all processors
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| ds_template | xr.Dataset | Input dataset, modified in place. | required |
Returns
| Name | Type | Description |
|---|---|---|
| None | The input dataset is modified in place. |
created_vars
process.blockwise.CompoundProcessor.created_vars()Return the list of variables created by the compound processor.
The compound created vars is the union of all created vars from sub-processors.
Returns
| Name | Type | Description |
|---|---|---|
| list[Var] | List of variables created by the compound processor |
describe
process.blockwise.CompoundProcessor.describe()Print a detailed description of the compound processor.
get_required_processors
process.blockwise.CompoundProcessor.get_required_processors()Filter the processor chain to keep only those required to produce allowed_vars.
A processor is kept if: - It creates any variables in allowed_vars, OR - It modifies any variables in allowed_vars, OR - It creates variables needed by a later kept processor
This avoids running unnecessary processors that don’t contribute to the output.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| processors | list[BlockProcessor] | Full list of processors | required |
Returns
| Name | Type | Description |
|---|---|---|
| list[BlockProcessor] | Filtered list containing only required processors |
global_attrs
process.blockwise.CompoundProcessor.global_attrs()Safe merge of the global attributes from all processors
input_vars
process.blockwise.CompoundProcessor.input_vars()Return the list of input variables for the compound processor.
The compound input vars are the variables being input in at least one processor, but not modified nor created by any processor.
Returns
| Name | Type | Description |
|---|---|---|
| list[Var] | List of input variables required by the compound processor |
map_blocks
process.blockwise.CompoundProcessor.map_blocks(ds, chained=True)Apply processors to dataset blocks with optional chaining.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| ds | xr.Dataset | Input xarray Dataset to process. | required |
| chained | bool | If True, use processor chaining for efficient computation. If False, don’t chain processors; this may involve additional dask graph overhead. | True |
Returns
| Name | Type | Description |
|---|---|---|
| xr.Dataset | Processed dataset containing only the output variables. |
modified_vars
process.blockwise.CompoundProcessor.modified_vars()Return the list of variables modified by the compound processor.
The compound modified vars are the variables that are modified vars for all processors (not created, not input).
Returns
| Name | Type | Description |
|---|---|---|
| list[Var] | List of variables modified by the compound processor |
process_and_validate
process.blockwise.CompoundProcessor.process_and_validate(block)Process a single block of data by applying all required processors, hereby implementing the compound processor.
This method iterates through the list of required processors obtained from get_required_processors() and calls their process_block method on the provided block, modifying it in place.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| block | xr.Dataset | The xarray Dataset representing the block to be processed. | required |