Skip to content

Ncml

NcMLHeader

Bases: ExtractionMethod

NcML backend for header method.

Method name: ncml

Example configuration

.. code-block:: yaml

  • method: ncml inputs: input_term: hello_world
Source code in extraction_methods/plugins/header/backends/ncml.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class NcMLHeader(ExtractionMethod):
    """
    NcML backend for header method.

    **Method name:** ``ncml``

    Example configuration:
        .. code-block:: yaml

        - method: ncml
          inputs:
            input_term: hello_world
    """

    input_class = NcMLHeaderInput

    def get_ncml(self) -> bytes:
        """Get the NcML file description."""

        parse_result = urlparse(self.input.input_term)

        if parse_result.netloc:
            return self.get_ncml_from_thredds()

        return self.get_ncml_from_fs()

    def get_ncml_from_thredds(self) -> bytes:
        """Read NcML response from THREDDS server.

        Returns
        -------
        bytes
        NcML content
        """

        r = httpx.get(
            self.input.input_term,
            params=self.input.request_params,
            timeout=self.input.request_timeout,
        )
        r.raise_for_status()
        return r.content

    def get_ncml_from_fs(self) -> bytes:
        """Return NcML file description using `ncdump` utility."""

        cmd = ["ncdump", "-hx", self.input.input_term]
        proc = subprocess.Popen(
            cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE
        )  # nosec B603
        if proc.stdout:
            return proc.stdout.read()
        else:
            return b""

    @update_input
    def run(self, body: dict[str, Any]) -> dict[str, Any]:
        # Convert response to an XML etree.Element
        content = self.get_ncml()
        elemement = fromstring(
            content, parser=XMLParser(encoding="UTF-8")
        )  # nosec B320

        for attribute in self.input.attributes:

            # Execute xpath expression
            value = elemement.xpath(attribute.key, namespaces=self.input.namespaces)

            if value:
                body[attribute.output_key] = value[0]

        return body

get_ncml()

Get the NcML file description.

Source code in extraction_methods/plugins/header/backends/ncml.py 
67
68
69
70
71
72
73
74
75
def get_ncml(self) -> bytes:
    """Get the NcML file description."""

    parse_result = urlparse(self.input.input_term)

    if parse_result.netloc:
        return self.get_ncml_from_thredds()

    return self.get_ncml_from_fs()

get_ncml_from_fs()

Return NcML file description using ncdump utility.

Source code in extraction_methods/plugins/header/backends/ncml.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_ncml_from_fs(self) -> bytes:
    """Return NcML file description using `ncdump` utility."""

    cmd = ["ncdump", "-hx", self.input.input_term]
    proc = subprocess.Popen(
        cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE
    )  # nosec B603
    if proc.stdout:
        return proc.stdout.read()
    else:
        return b""

get_ncml_from_thredds()

Read NcML response from THREDDS server.

Returns

bytes NcML content

Source code in extraction_methods/plugins/header/backends/ncml.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_ncml_from_thredds(self) -> bytes:
    """Read NcML response from THREDDS server.

    Returns
    -------
    bytes
    NcML content
    """

    r = httpx.get(
        self.input.input_term,
        params=self.input.request_params,
        timeout=self.input.request_timeout,
    )
    r.raise_for_status()
    return r.content

NcMLHeaderInput

Bases: Input

Model for NcML Header Input.

Parameters:

Name Type Description Default
input_term str

term for method to run on.

 '$uri'
request_params dict[str, Any]

params for request.

 {'catalog': None, 'dataset': None}
namespaces dict[str, str]

NcML namespaces.

 {'ncml': 'http://www.unidata.ucar.edu/namespaces/netcdf/ncml-2.2'}
attributes list[KeyOutputKey]

attributes to be extracted.

 []
request_timeout int

request time out.

 15
Source code in extraction_methods/plugins/header/backends/ncml.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class NcMLHeaderInput(Input):
    """
    Model for NcML Header Input.
    """

    input_term: str = Field(
        default="$uri",
        description="term for method to run on.",
    )
    request_params: dict[str, Any] = Field(
        default={"catalog": None, "dataset": None},
        description="params for request.",
    )
    namespaces: dict[str, str] = Field(
        default={"ncml": "http://www.unidata.ucar.edu/namespaces/netcdf/ncml-2.2"},
        description="NcML namespaces.",
    )
    attributes: list[KeyOutputKey] = Field(
        default=[],
        description="attributes to be extracted.",
    )
    request_timeout: int = Field(
        default=15,
        description="request time out.",
    )