CHAP.foxden package

PipelineItems unique to communicating with FOXDEN

This module contains all the PipelineItems (Processors, Readers and Writers) that are unique to communicate with FOXDEN. Any of these PipelineItems can be used as items in a CHAP Pipeline or instantiated from a user Python script.

Note

Using the FOXDEN pipeline items in a CHAP Pipeline and running it, requires a conda environent for the appropriate workflow or access to the appropriate CHAP executable.

Submodules summary

models: Pydantic model configuration classes unique to the FOXDEN pipeline items.
processor: Processors unique to the FOXDEN pipeline items.
reader: Readers unique to the FOXDEN pipeline items.
writer: Writers unique to the FOXDEN pipeline items.

Submodules

CHAP.foxden.models module

Pydantic model configuration classes unique to the the FOXDEN integration.

class FoxdenRequestConfig(*, did: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=True, strict=True, min_length=None, max_length=None, pattern=None, ascii_only=None)] | None = None, idx: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None] | None = 0, limit: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None = 10, query: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=True, strict=True, min_length=None, max_length=None, pattern=None, ascii_only=None)] | None = None, url: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern=None, ascii_only=None)] | None = None, verbose: bool | None = 'False')[source]

Bases: CHAPBaseModel

FOXDEN HTTP request base configuration class.

Variables:

did (string, optional) – FOXDEN dataset identifier (DID).
idx (int, optional) – Index of the first record in the list of records to be retured, defaults to 0.
limit (int, optional) – Maximum number of returned records, defaults to 10.
query (string, optional) – FOXDEN query.
url (str) – URL of service.
verbose (bool, optional) – Verbose output flag, defaults to False.

create_http_request_payload(reader)[source]

Create the payload for a HTTP request.

Parameters:: reader (FoxdenDataDiscoveryReader or FoxdenMetadataReader or FoxdenProvenanceReader) – Any of the FOXDEN readers in reader.
Returns:: JSON string of the HTTP request.
Return type:: str

did: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=True, strict=True, min_length=None, max_length=None, pattern=None, ascii_only=None)] | None

idx: Annotated[int, None, Interval(gt=None, ge=0, lt=None, le=None), None] | None

limit: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=True, strict=True, min_length=None, max_length=None, pattern=None, ascii_only=None)] | None

url: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern=None, ascii_only=None)] | None

verbose: bool | None

CHAP.foxden.processor module

Module for Processors unique to the FOXDEN integration with CHAP.

Add discription of FOXDEN

class ProvenanceFileProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')[source]

Bases: Processor

A Processor that retrieves a FOXDEN provenance record from the pipeline and returns the content of the in or output file listed in the record.

Variables:

file_type (Literal['input', 'output'], optional.) – The ‘file_type’ in the provenance record, defaults to ‘output’.
nxmemory (int, optional) – Maximum memory usage when reading NeXus files, ignore for any other file type.

file_type: Literal['input', 'output'] | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

nxmemory: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

process(data)[source]

Return the content of in or output files listed in the provenance record.

Returns:: The file content.
Return type:: Any

CHAP.foxden.reader module

Module for Readers unique to the FOXDEN integration with CHAP.

class FoxdenDataDiscoveryReader(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')[source]

Bases: PipelineItem

Reader for the FOXDEN Data Discovery service.

Variables:: config (dict, optional) – Initialization parameters for an instance of FoxdenRequestConfig.

config: FoxdenRequestConfig | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

pipeline_fields: dict

read()[source]

Read records from the FOXDEN Data Discovery service based on DID (Dataset Identifier) or an arbitrary query.

Returns:: Discovered data records.
Return type:: list

validate_foxdendatadiscoveryreader_after()[source]

Validate the FoxdenDataDiscoveryReader configuration.

Returns:: Validated model configuration
Return type:: FoxdenDataDiscoveryReader

class FoxdenMetadataReader(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')[source]

Bases: PipelineItem

Reader for the FOXDEN Metadata service.

Variables:: config (dict, optional) – Initialization parameters for an instance of FoxdenRequestConfig.

config: FoxdenRequestConfig | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

pipeline_fields: dict

read()[source]

Read a record from the FOXDEN Metadata service based on DID (Dataset Identifier) or an arbitrary query.

Returns:: Metadata record.
Return type:: dict

validate_foxdenmetadatareader_after()[source]

Validate the FoxdenMetadataReader configuration.

Returns:: Validated model configuration
Return type:: FoxdenMetadataReader

class FoxdenProvenanceReader(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')[source]

Bases: PipelineItem

Reader for the FOXDEN Provenance service.

Variables:: config (dict, optional) – Initialization parameters for an instance of FoxdenRequestConfig.

config: FoxdenRequestConfig | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

pipeline_fields: dict

read()[source]

Read records from the FOXDEN Provenance service based on did or an arbitrary query.

Returns:: Provenance input and output file records.
Return type:: list

validate_foxdenprovenancereader_after()[source]

Validate the FoxdenProvenanceReader configuration.

Returns:: Validated model configuration
Return type:: FoxdenProvenanceReader

class FoxdenSpecScansReader(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')[source]

Bases: PipelineItem

Reader for FOXDEN SpecScans data from a specific FOXDEN SpecScans service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

read(url, data, *, did='', query='', spec=None, method='POST', verbose=False)[source]

Read and return data from a specific FOXDEN SpecScans service.

Parameters:

url (str) – URL of service.
data (list[PipelineData]) – Input data.
did (string, optional) – FOXDEN dataset identifier (DID).
query (string, optional) – FOXDEN query.
spec (dictionary, optional) – FOXDEN spec.
method (str, optional) – HTTP method to use, ‘POST’ for creation or ‘PUT’ for update, defaults to ‘POST’.
verbose (bool, optional) – Verbose output flag, defaults to False.

Returns:

Contents of the input data.

Return type:

list

CHAP.foxden.utils module

FOXDEN utils module.

HTTP_request(url, payload, *, method='POST', headers=None, scope='read', timeout=10, dry_run=False)[source]

Submit a HTTP request to a FOXDEN service.

Parameters:

url (str) – URL of service.
payload (str) – HTTP request payload.
method (str, optional) – HTTP method to use, defaults to ‘POST’.
headers (dictionary, optional) – HTTP headers to use.
scope (string, optional) – FOXDEN scope: ‘read’ or ‘write’, defaults to ‘read’.
timeout (str, optional) – Timeout of HTTP request, defaults to 10.
dry_run (bool, optional) – dry_run option to verify HTTP workflow, defaults to False.

Returns:

HTTP response.

Return type:

requests.models.Response

read_foxden_token(scope)[source]

Obtain a FOXDEN token.

Parameters:: scope (string) – FOXDEN scope: ‘read’ or ‘write’.
Returns:: The FOXDEN token.
Return type:: str

CHAP.foxden.writer module

Module for Readers unique to the FOXDEN integration with CHAP.

class FoxdenDoiWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')[source]

Bases: PipelineItem

Writer for saving info to the FOXDEN DOI service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

write(url, data, *, provider='Datacite', description='', draft=True, publishMetadata=True, verbose=False)[source]

Write data to the FOXDEN DOI service.

Parameters:

data (list[PipelineData]) – Input data.
url (str) – URL of service.
provider (str, optional) – DOI provider name, e.g. ‘Zenodo’, ‘Datacite’, or ‘Materialcommons’, defaults to ‘Datacite’.
description (str, optional) – Dataset description.
draft (bool, optional) – Draft DOI flag, defaults to True.
publishMetadata (bool, optional) – Publish metadata with DOI, defaults to True.
verbose (bool, optional) – Verbose output flag, defaults to False.

Returns:

HTTP response from FOXDEN DOI service.

Return type:

list[dict]

class FoxdenMetadataWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')[source]

Bases: PipelineItem

Writer for saving data to the FOXDEN Metadata service.

Variables:

url (str) – URL of service.
verbose (bool, optional) – Verbose output flag, defaults to False.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

url: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern=None, ascii_only=None)]

verbose: bool | None

write(data)[source]

Write data to the FOXDEN Metadata service.

Parameters:: data (list[PipelineData]) – Input data.
Returns:: HTTP response from FOXDEN Metadata service.
Return type:: dict

class FoxdenProvenanceWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO')[source]

Bases: PipelineItem

Writer for saving data to the FOXDEN Provenance service.

Variables:

url (str) – URL of service.
verbose (bool, optional) – Verbose output flag, defaults to False.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

url: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern=None, ascii_only=None)]

verbose: bool | None

write(data)[source]

Write data to the FOXDEN Provenance service.

Parameters:: data (list[PipelineData]) – Input data.
Returns:: HTTP response from FOXDEN Provenance serviceand the updated provenance record.
Return type:: PipelineData, PipelineData