Processing

Link to code

The processing.json file captures the data processing and analysis steps that have been carried out – mostly for derived data assets. This tracks what code was used for each step, when it was run, what the input and outputs where, what parameters were set. This includes things like spike sorting, image alignment, cell segmentation. It also includes manual annotation, quality control, and data analysis.

The processing file should be appended to with each subsequent stage of processing or analysis.

Example

"""example processing"""

import argparse
from datetime import datetime, timezone

from aind_data_schema.components.identifiers import Code, DataAsset
from aind_data_schema.core.processing import (
    DataProcess,
    Processing,
    ProcessName,
    ProcessStage,
    ResourceTimestamped,
    ResourceUsage,
)
from aind_data_schema_models.units import MemoryUnit
from aind_data_schema_models.system_architecture import OperatingSystem, CPUArchitecture

# If a timezone isn't specified, the timezone of the computer running this
# script will be used as default
t = datetime(2022, 11, 22, 8, 43, 00, tzinfo=timezone.utc)


cpu_usage_list = [
    ResourceTimestamped(timestamp=datetime(2024, 9, 13, tzinfo=timezone.utc), usage=75.5),
    ResourceTimestamped(timestamp=datetime(2024, 9, 13, tzinfo=timezone.utc), usage=80.0),
]

gpu_usage_list = [
    ResourceTimestamped(timestamp=datetime(2024, 9, 13, tzinfo=timezone.utc), usage=60.0),
    ResourceTimestamped(timestamp=datetime(2024, 9, 13, tzinfo=timezone.utc), usage=65.5),
]

ram_usage_list = [
    ResourceTimestamped(timestamp=datetime(2024, 9, 13, tzinfo=timezone.utc), usage=70.0),
    ResourceTimestamped(timestamp=datetime(2024, 9, 13, tzinfo=timezone.utc), usage=72.5),
]

file_io_usage_list = [
    ResourceTimestamped(timestamp=datetime(2024, 9, 13, tzinfo=timezone.utc), usage=5.5),
    ResourceTimestamped(timestamp=datetime(2024, 9, 13, tzinfo=timezone.utc), usage=6.0),
]

example_code = Code(
    url="https://github.com/abcd",
    version="0.1",
    parameters={"size": 7},
)

p = Processing.create_with_sequential_process_graph(
    pipelines=[
        Code(
            name="Imaging processing pipeline",
            url="https://url/for/pipeline",
            version="0.1.1",
            input_data=[DataAsset(
                name="123456_2026-05-20_14-14-14",
            )]
        ),
    ],
    data_processes=[
        DataProcess(
            process_type=ProcessName.IMAGE_TILE_FUSING,
            experimenters=["Dr. Dan"],
            stage=ProcessStage.PROCESSING,
            start_date_time=t,
            end_date_time=t,
            output_path="path/to/outputs",
            pipeline_name="Imaging processing pipeline",
            code=example_code.model_copy(
                update=dict(
                    parameters={"size": 7},
                )
            ),
            resources=ResourceUsage(
                os=OperatingSystem.UBUNTU_20_04,
                architecture=CPUArchitecture.X86_64,
                cpu="Intel Core i7",
                cpu_cores=8,
                gpu="NVIDIA GeForce RTX 3080",
                system_memory=32.0,
                system_memory_unit=MemoryUnit.GB,
                ram=16.0,
                ram_unit=MemoryUnit.GB,
                cpu_usage=cpu_usage_list,
                gpu_usage=gpu_usage_list,
                ram_usage=ram_usage_list,
            ),
        ),
        DataProcess(
            process_type=ProcessName.FILE_FORMAT_CONVERSION,
            pipeline_name="Imaging processing pipeline",
            experimenters=["Dr. Dan"],
            stage=ProcessStage.PROCESSING,
            start_date_time=t,
            end_date_time=t,
            output_path="path/to/outputs",
            code=example_code.model_copy(
                update=dict(
                    parameters={"u": 7, "z": True},
                )
            ),
        ),
        DataProcess(
            process_type=ProcessName.IMAGE_DESTRIPING,
            pipeline_name="Imaging processing pipeline",
            experimenters=["Dr. Dan"],
            stage=ProcessStage.PROCESSING,
            start_date_time=t,
            end_date_time=t,
            output_path="path/to/output",
            code=example_code.model_copy(
                update=dict(
                    parameters={"a": 2, "b": -2},
                )
            ),
        ),
        DataProcess(
            name="Analysis 1",
            stage=ProcessStage.ANALYSIS,
            experimenters=["Some Analyzer"],
            process_type=ProcessName.ANALYSIS,
            start_date_time=t,
            end_date_time=t,
            output_path="path/to/outputs",
            code=example_code.model_copy(
                update=dict(
                    parameters={"size": 7},
                )
            ),
        ),
        DataProcess(
            name="Analysis 2",
            stage=ProcessStage.ANALYSIS,
            experimenters=["Some Analyzer"],
            process_type=ProcessName.ANALYSIS,
            start_date_time=t,
            end_date_time=t,
            output_path="path/to/outputs",
            code=example_code.model_copy(
                update=dict(
                    parameters={"u": 7, "z": True},
                )
            ),
        ),
    ],
)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-dir", default=None, help="Output directory for generated JSON file")
    args = parser.parse_args()

    serialized = p.model_dump_json()
    deserialized = Processing.model_validate_json(serialized)
    p.write_standard_file(output_directory=args.output_dir)

Core file

Processing

Description of all processes run on data

Field	Type	Title (Description)
data_processes	List[DataProcess]	Data processing
pipelines	Optional[List[Code]]	Pipelines (For processing done with pipelines, list the repositories here. Pipelines must use the name field ,and be referenced in the pipeline_name field of a DataProcess.)
notes	Optional[str]	Notes
dependency_graph	Optional[Dict[str, List[str]]]	Dependency graph (Directed graph of processing step dependencies. Each key is a process name, and the value is a list of process names that are inputs to that process.)

Model definitions

DataProcess

Description of a single processing step

Field	Type	Title (Description)
process_type	ProcessName	Process type
name	str	Name ((‘Unique name of the processing step.’, ‘ If not provided, the type will be used as the name.’))
stage	ProcessStage	Processing stage
code	Code	Code (Code used for processing)
experimenters	List[str]	Experimenters (People responsible for processing)
pipeline_name	Optional[str]	Pipeline name (Pipeline names must exist in Processing.pipelines)
start_date_time	datetime (timezone-aware)	Start date time
end_date_time	Optional[datetime (timezone-aware)]	End date time
output_path	Optional[AssetPath]	Output path (Path to processing outputs, if stored.)
output_parameters	Optional[dict]	Outputs (Output parameters)
notes	Optional[str]	Notes
resources	Optional[ResourceUsage]	Process resource usage

ProcessStage

Stages of processing

Name	Value
PROCESSING	Processing
ANALYSIS	Analysis

ResourceTimestamped

Description of resource usage at a moment in time

Field	Type	Title (Description)
timestamp	datetime (timezone-aware)	Timestamp
usage	float	Usage

ResourceUsage

Description of resources used by a process

Field	Type	Title (Description)
os	str	Operating system
architecture	str	Architecture
cpu	Optional[str]	CPU name
cpu_cores	Optional[int]	CPU cores
gpu	Optional[str]	GPU name
system_memory	Optional[float]	System memory
system_memory_unit	Optional[MemoryUnit]	System memory unit
ram	Optional[float]	System RAM
ram_unit	Optional[MemoryUnit]	Ram unit
cpu_usage	Optional[List[ResourceTimestamped]]	CPU usage
gpu_usage	Optional[List[ResourceTimestamped]]	GPU usage
ram_usage	Optional[List[ResourceTimestamped]]	RAM usage
usage_unit	str	Usage unit