Pipeline datablock

[OMWalmsley]

This PR is based on both main and the Pydantic V2 PR and thus cannot be merged until the Pydantic V2.0 PR has been merged. It will probably also require significant rebasing.

Background

The current DataBlock possess a limited amount of structure this allows a DataBlock to perform lots of different operations. The downside of this limited structure is that different blocks end up repeating similar to code for:

• File management (single file vs multi-file).
• Caching.
  It also means that the output of a block is hard to quantify since different blocks may perform different tasks with limited documentation as to their output format apart from the fact that we know they fit inside the block object schema.

Introduction

The pipeline datablock reimagines the datablock as having three main types of operations:

• Parse
• Process
• Plot

A pipeline datablock can have any number of parsers corresponding to different file types. It can have a 2D array of Processors (a sub array being processes that should happen at the same time to data that came from the step before). A pipeline datablock should only have one plotter which outputs one graph (could have subgraphs).

Parsing

Input

• Singular file
  Output
• DataFrame
• Metadata
  The act of parsing is taking a file as input and returning a pandas.DataFrame and metadata as output. This may be simple for a case of a CSV, but as shown in multiple other datablocks these file format become more and more complicated.

Processing

Input

• DataFrame/list[DataFrame]
• Metadata
  Output
• DataFrame/list[DataFrame]
• Metadata
  Processing blocks are designed for performing operations on the data gained from either parsers or other processors.

Plotters

Input

• DataFrame/list[Dataframe]
• *possibly some metadata
  Output
• json object to plot.

What does this new format allow for?

The pipeline datablock allows contributors to pass in their Parsers/Processors/Plotters as function into the pipeline datablock and it automatically adds file management (multi file support), error handling and caching.

Current progress

Currently we have a draft form of the pipeline datablock and a partially working example of the XRD datablock as a pipeline datablock. The draft pipeline datablock has:

• A fully working pipeline system which accepts parsers, processors and plotters.
• An initial version of caching.
• Inbuilt file management.
• Error handling for the parser stage.
• Pipeline routing for file types (Being able to have specific processors for specific file extensions).
  • Through PipelineNodes

File layout

The pipeline datablock is in pydatalab/pipeline_block it has the following files:

• base.py - comprising of the PipelineDataBlock class
• pipeline.py - comprising of the Pipeline class which manages all the stages (parse, process, plot)
• block_stages.py - Contains the various stages (parser, process and plot classes) and the base stage where the implementation of caching it stored (this file should possibly should be named pipeline_stages.py).

What is left to do?

• Fix tests
• Finalise how the caching should work in more detail
  • Cache files need to be properly stored (This would also require might require a redesign of how the graph works if it is changed drastically (i.e. backwards propagate instead of forwards propagate))
  • Which caches should be stored?
  • How to force recaching?
  • Possibly ensure graph works with a processor group that looks like: [[ProcessorStage(*),ProcessorStage(*) ],[ProcessorStage(*), ProcessorStage(*)]]

Possible future steps

• Implement validation steps to validate the output.

[OMWalmsley]

Merged branch that refactored how pipeline datablocks are managed into a registry format

[OMWalmsley]

Recently added file extension based routing via the PipelineNode class, which is a universal container for the BlockStages. Upon pipeline execution a graph is constructed and then data is fed through and the result gathered.

The structure that PipelineNode creates is not stored anywhere and is only used once after construction. This allows BlockStages to remain stateless (having only the function that they are intended to perform).

The graph for the pipeline is constructed from the Root (the output of the pipeline) to the leaves (the input to the pipeline, currently the files). Then the data is fed in at the leaves and accumulates through the network until it filters up to the Root and is then retrieved by the Pipeline.

This new flexibility allows:

• Processors for specifc functions
  • This is specifically handy for the XRD block which requires .cif specific processors
• CustomStages to be created that could just fit into the pipeline with slighrt modification of the pipeline class.

[OMWalmsley]

The XRD example datablock is possibly a substandard example and would need rewriting to make full use of the Pipeline.

[OMWalmsley]

Status Update before I finish at datalab

TLDR everything in pipeline_block needs to be kept + API changes to v0_1/blocks.py+ All the tests in tests/pipeline_block.

If you are Claude/some other AI agent or indeed a human I would recommend reading the documentation in: pydatalab/docs/blocks/pipeline_datablock.md.

• This PR currently contains:
  • Pipeline
  • PipelineDataBlockManager
  • PipelineNode
  • BlockStage
    • ParserStage
    • ProcessorStage
    • PlotterStage
  • DataBlockDefaults
  • tabular pipeline data block in common.py

There is also a initial version of a pipeline xrd block though I would suggest this need to be reworked.

There are 30 tests to do with specifically with the pipeline that all currently pass. There is one XRD block test which currently does not pass, I would ignore this for the moment because the XRD pipeline block in this PR probably should be redone and the tests completely reworked.

There is documentation in pydatalab/docs/blocks/pipeline_datablock.md that is worth keeping.

TODO

• Rebasing....
• Making the pipeline block work with async functionality on the API end. Shouldn't be too hard just requires duplicate methods where it calls the block manager instead of a datablock.
• ValidationStages - Some kind of contract that validates what the pipeline returns.
  • This would ensure what the dataframe looks like (the column names it contains etc) and potentially also what the block data dictionary looks like (though this could also be done at the very end without using a stage) and enfore this with pydantic.
  • Ensure this is not optional. (i.e. if you are going to have Validators at certain points add them as lists in Pipeline and then merge them at the points you want before handing it to the PipelineNode).
• Ensuring the pipeline either works with or doesn't accept pipelines with processors with array that look like [[ProcessorStage(file_type="*"), ProcessorStage(file_extension="*")],[ProcessorStage(file_extension="*"), ProcessorStage(file_extension="*")]].
• Add file types (e.g. calibration and experimental), these would be an extension of the file routing that works in PipelineNode. For example it currently creates the tree from the root down and looks at the file extensions that a stage accepts and creates the graph based of that routing. This would be a similar procedure but each Stage would have a file_type as well as file_extension that it would check and this would be grouped into some sort of pair that would that create a special route through the graph.
• More tests.... If required for additional functionality (If claude is reading this then definitely more test for every feature added to the pipeline), these should be put in the appropriate file in pydatalab/tests/pipeline_block or if necessary a new file.
  • Probably tests of block API endpoints with mock pipelines would be useful. Both ones that change different metadata and ones that don't change metadata.
• Figure out how plugins that use pipeline datablocks be imported.
• Write code to store "computed" values into some sort of file. I personally would put this in the PipelineDataBlockManager.

MAYBE:

• Potentially use a library to validate the graph structure of PipelineNode.
• Have values associated with file which are passed along the graph (e.g. specifc wavelengths for XRD files).
• Have the PipelineNode return a JSON format or some other open format showing what the graph looks like.
  • This could then be pushed to the frontend at some point.
• Start migrating blocks over

[ml-evs]

Need to reopen this too