extraction_methods.plugins.header.backends package

Submodules

extraction_methods.plugins.header.backends.cf module

extraction_methods.plugins.header.backends.ncml module

NCML Header Backend

class extraction_methods.plugins.header.backends.ncml.NcMLHeader(*args: Any, **kwargs: Any)

Bases: ExtractionMethod

Method: ncml

Description:

NcML backend for header method.

Configuration Options: .. list-table:

- ``input_term``:term for method to run on
- ``request_params``:params for request
- ``namespaces``:NcML namespaces
- ``attributes``:attributes to be extracted
- ``request_timeout``:request time out

Example configuration: .. code-block:: yaml

• method: ncml inputs:

  input_term: hello_world

get_ncml() → bytes

Get the NcML file description.

get_ncml_from_fs() → bytes

Return NcML file description using ncdump utility.

get_ncml_from_thredds() → bytes

Read NcML response from THREDDS server.

Returns

bytes NcML content

input_class

alias of NcMLHeaderInput

run(body: dict[str, Any]) → Any

Run the method.

Parameters:

body (dict) – current generated properties

Returns:

updated body dict

Return type:

dict

class extraction_methods.plugins.header.backends.ncml.NcMLHeaderInput(*, exists_key: str = '$', exists_delimiter: str = '.', input_term: str = '$uri', request_params: dict[str, Any] = {'catalog': None, 'dataset': None}, namespaces: dict[str, str] = {'ncml': 'http://www.unidata.ucar.edu/namespaces/netcdf/ncml-2.2'}, attributes: list[KeyOutputKey] = [], request_timeout: int = 15)

Bases: Input

Model for NcML Header Input.

attributes: list[KeyOutputKey]

input_term: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'attributes': FieldInfo(annotation=list[KeyOutputKey], required=False, default=[], description='attributes to be extracted.'), 'exists_delimiter': FieldInfo(annotation=str, required=False, default='.', description='Delimiter for nested exists terms.'), 'exists_key': FieldInfo(annotation=str, required=False, default='$', description='Key to signify a previously extracted terms.'), 'input_term': FieldInfo(annotation=str, required=False, default='$uri', description='term for method to run on.'), 'namespaces': FieldInfo(annotation=dict[str, str], required=False, default={'ncml': 'http://www.unidata.ucar.edu/namespaces/netcdf/ncml-2.2'}, description='NcML namespaces.'), 'request_params': FieldInfo(annotation=dict[str, Any], required=False, default={'catalog': None, 'dataset': None}, description='params for request.'), 'request_timeout': FieldInfo(annotation=int, required=False, default=15, description='request time out.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

namespaces: dict[str, str]

request_params: dict[str, Any]

request_timeout: int

extraction_methods.plugins.header.backends.xarray module

Xarray Header Backend

class extraction_methods.plugins.header.backends.xarray.XarrayHeader(*args: Any, **kwargs: Any)

Bases: ExtractionMethod

Method: xarray

Description:

Xarray backend for header method.

Configuration Options: .. list-table:

- ``input_term``:term for method to run on
- ``dataset_kwargs``:kwargs to open dataset
- ``attributes``:attributes to be extracted

Example configuration: .. code-block:: yaml

• method: xarray inputs:

  input_term: hello_world

input_class

alias of XarrayHeaderInput

run(body: dict[str, Any]) → Any

Run the method.

Parameters:

body (dict) – current generated properties

Returns:

updated body dict

Return type:

dict

class extraction_methods.plugins.header.backends.xarray.XarrayHeaderInput(*, exists_key: str = '$', exists_delimiter: str = '.', input_term: str = '$uri', dataset_kwargs: dict[str, Any] = {}, attributes: list[KeyOutputKey] = [])

Bases: Input

Model for Xarray Header Method Input.

attributes: list[KeyOutputKey]

dataset_kwargs: dict[str, Any]

input_term: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'attributes': FieldInfo(annotation=list[KeyOutputKey], required=False, default=[], description='attributes to be extracted.'), 'dataset_kwargs': FieldInfo(annotation=dict[str, Any], required=False, default={}, description='kwargs to open dataset.'), 'exists_delimiter': FieldInfo(annotation=str, required=False, default='.', description='Delimiter for nested exists terms.'), 'exists_key': FieldInfo(annotation=str, required=False, default='$', description='Key to signify a previously extracted terms.'), 'input_term': FieldInfo(annotation=str, required=False, default='$uri', description='term for method to run on.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

Module contents