def get_namespaces():
def csv_splitter(input_string):
    def __init__(self, split_func=None):
    def validator_func(self, input_value):
    def validate(self, input_name, input_value):
    def validate_type(self, input_name, input_value):
    def validator_func(self, input_value):
    def validator_func(self, input_value):
    def validator_func(self, input_value):
    def __init__(self, fpath):
    def filepath(self):
    def setup(self, ds):
    def __init__(self, options=None):
    def get_test_ctx(self, severity, name, variable=None):
    def __del__(self):
    def std_check_in(cls, dataset, name, allowed_vals):
    def std_check(cls, dataset, name):
    def __init__(
        self,
        weight=BaseCheck.MEDIUM,
        value=None,
        name=None,
        msgs=None,
        children=None,
        checker=None,
        check_method=None,
        variable_name=None,
    ):
    def __repr__(self):
    def serialize(self):
    def __eq__(self, other):
    def __init__(
        self,
        category=None,
        description="",
        out_of=0,
        score=0,
        messages=None,
        variable=None,
    ):
    def to_result(self):
    def assert_true(self, test, message):
    def add_failure(self, message):
    def add_pass(self):
def std_check_in(base_context, name, allowed_vals):
def std_check(dataset, name):
def xpath_check(tree, xpath):
def maybe_get_global_attr(attr_name, ds):
def attr_check(kvp, ds, priority, ret_val, gname=None, var_name=None):
def check_has(priority=BaseCheck.HIGH, gname=None):
    def _inner(func):
        def _dec(s, ds):
def fix_return_value(v, method_name, method=None, checker=None):
def ratable_result(value, name, msgs, variable_name=None):
def score_group(group_name=None):
    def _inner(func):
        def _dec(s, ds):
            def dogroup(r):
class ValidationObject(object):
class EmailValidator(ValidationObject):
class RegexValidator(ValidationObject):
class UrlValidator(ValidationObject):
class GenericFile(object):
class BaseCheck(object):
class BaseNCCheck(object):
class BaseSOSGCCheck(object):
class BaseSOSDSCheck(object):
class Result(object):
class TestCtx(object):
    HIGH = 3
    MEDIUM = 2
    LOW = 1 class ACDD1_3Check(ACDDNCCheck):

    _cc_spec_version = "1.3"
    _cc_description = "Attribute Conventions for Dataset Discovery (ACDD) 1.3"
    register_checker = True

    def __init__(self):
        super(ACDD1_3Check, self).__init__()
        self.high_rec_atts.extend([("Conventions", self.verify_convention_version)])

        self.rec_atts.extend(
            [
                "geospatial_vertical_positive",
                "geospatial_bounds_crs",
                "geospatial_bounds_vertical_crs",
                "publisher_name",  # publisher,dataCenter
                "publisher_url",  # publisher
                "publisher_email",  # publisher
                "source",
            ]
        )

        self.sug_atts.extend(
            [
                ("creator_type", ["person", "group", "institution", "position"]),
                "creator_institution",
                "platform",
                "platform_vocabulary",
                "keywords_vocabulary",
                "instrument",
                "metadata_link",
                "product_version",
                "references",
                ("publisher_type", ["person", "group", "institution", "position"]),
                "instrument_vocabulary",
                "date_metadata_modified",
                "program",
                "publisher_institution",
            ]
        )

        # override the ISO date checks in
        def _check_attr_is_iso_date(attr, ds):
            result_name = "{}_is_iso".format(attr)
            if not hasattr(ds, attr):
                return ratable_result(
                    (0, 2), result_name, ["Attr {} is not present".format(attr)]
                )
            else:
                iso_check, msgs = datetime_is_iso(getattr(ds, attr))
                return ratable_result((1 + iso_check, 2), result_name, msgs)

        # run ISO 8601 date checks against the date_created, date_issued,
        # date_modified, and date_metadata_modified global attributes
        self.rec_atts = kvp_convert(self.rec_atts)
        self.rec_atts["date_created"] = partial(_check_attr_is_iso_date, "date_created")
        self.sug_atts = kvp_convert(self.sug_atts)
        for k in (
            "date_{}".format(suffix)
            for suffix in ("issued", "modified", "metadata_modified")
        ):

            self.sug_atts[k] = partial(_check_attr_is_iso_date, k)

    def check_metadata_link(self, ds):
        """
        Checks if metadata link is formed in a rational manner

        :param netCDF4.Dataset ds: An open netCDF dataset
        """
        if not hasattr(ds, u"metadata_link"):
            return
        msgs = []
        meta_link = getattr(ds, "metadata_link")
        if "http" not in meta_link:
            msgs.append("Metadata URL should include http:// or https://")
        valid_link = len(msgs) == 0
        return Result(BaseCheck.LOW, valid_link, "metadata_link_valid", msgs)

    def check_id_has_no_blanks(self, ds):
        """
        Check if there are blanks in the id field

        :param netCDF4.Dataset ds: An open netCDF dataset
        """
        if not hasattr(ds, u"id"):
            return
        if " " in getattr(ds, u"id"):
            return Result(
                BaseCheck.MEDIUM,
                False,
                "no_blanks_in_id",
                msgs=[u"There should be no blanks in the id field"],
            )
        else:
            return Result(BaseCheck.MEDIUM, True, "no_blanks_in_id", msgs=[])

    def check_var_coverage_content_type(self, ds):
        """
        Check coverage content type against valid ISO-19115-1 codes

        :param netCDF4.Dataset ds: An open netCDF dataset
        """
        results = []
        for variable in cfutil.get_geophysical_variables(ds):
            msgs = []
            ctype = getattr(ds.variables[variable], "coverage_content_type", None)
            check = ctype is not None
            if not check:
                msgs.append("coverage_content_type")
                results.append(
                    Result(
                        BaseCheck.HIGH, check, self._var_header.format(variable), msgs
                    )
                )
                continue

            # ISO 19115-1 codes
            valid_ctypes = {
                "image",
                "thematicClassification",
                "physicalMeasurement",
                "auxiliaryInformation",
                "qualityInformation",
                "referenceInformation",
                "modelResult",
                "coordinate",
            }
            if ctype not in valid_ctypes:
                msgs.append(
                    'coverage_content_type in "%s"' % (variable, sorted(valid_ctypes))
                )
                results.append(
                    Result(
                        BaseCheck.HIGH,
                        check,  # append to list
                        self._var_header.format(variable),
                        msgs,
                    )
                )

        return results class BaseCheck(object):
    HIGH = 3
    MEDIUM = 2
    LOW = 1

    _cc_checker_version = __version__
    _cc_display_headers = {3: "High Priority", 2: "Medium Priority", 1: "Low Priority"}

    supported_ds = []

    def setup(self, ds):
        """
        Common setup method for a Checker.

        Automatically run when running a CheckSuite. Define this method in your Checker class.
        """
        pass

    def __init__(self, options=None):
        self._defined_results = defaultdict(lambda: defaultdict(dict))
        if options is None:
            self.options = set()
        else:
            self.options = options

    def get_test_ctx(self, severity, name, variable=None):
        """
        Creates an existing TestCtx object in _defined_results dict if it does
        not exist for the current checker instance, or an returns the existing
        TestCtx for modification. Takes a severity level and name and uses the
        two element tuple formed by the arguments as a key into the dict.

        :param int severity: A BaseCheck severity level
        :param str name: The name of the check
        :rtype compliance_checker.base.TestCtx:
        :returns: A new or or existing `TestCtx` instance taken from this
                  instance's _defined_results dict
        """
        # Is it necessary to key out by severity?  Is severity level unique
        # per check?  If so, it could be eliminated from key hierarchy
        if severity not in self._defined_results[name][variable]:
            self._defined_results[name][variable][severity] = TestCtx(
                severity, name, variable=variable
            )
        return self._defined_results[name][variable][severity]

    def __del__(self):
        """
        Finalizer. Ensure any caches shared by multiple checkers
        are cleared before the next checker uses it. Some caches were
        inadvertently mutated by other functions.
        """

        cfutil.get_geophysical_variables.cache_clear()
        cfutil.get_time_variables.cache_clear() class BaseNCCheck(object):
    """
    Base Class for NetCDF Dataset supporting Check Suites.
    """

    supported_ds = {Dataset, MemoizedDataset}

    @classmethod
    def std_check_in(cls, dataset, name, allowed_vals):
        """
        Returns 0 if attr not present, 1 if present but not in correct value, 2 if good
        """
        if name not in dataset.ncattrs():
            return 0

        ret_val = 1
        if dataset.getncattr(name) in allowed_vals:
            ret_val += 1

        return ret_val

    @classmethod
    def std_check(cls, dataset, name):
        return name in dataset.ncattrs() class BaseSOSDSCheck(object):
    """
    Base class for SOS-DescribeSensor supporting Check Suites.
    """

    supported_ds = [SensorML] class BaseSOSGCCheck(object):
    """
    Base class for SOS-GetCapabilities supporting Check Suites.
    """

    supported_ds = [SensorObservationService_1_0_0] class Result(object):
    """
    Holds the result of a check method.

    Stores such information as the check's value (True, False, a 2-tuple of (pass, total) or None for a skip),
    weight of the check, any granular messages, or a hierarchy of results. If given value is not a tuple, it
    is cast as a boolean using the bool() function.

    Stores the checker instance and the check method that produced this result.
    """

    def __init__(
        self,
        weight=BaseCheck.MEDIUM,
        value=None,
        name=None,
        msgs=None,
        children=None,
        checker=None,
        check_method=None,
        variable_name=None,
    ):

        self.weight = weight

        if value is None:
            self.value = None
        elif isinstance(value, tuple):
            assert len(value) == 2, "Result value must be 2-tuple or boolean!"
            self.value = value
        else:
            self.value = bool(value)
        self.name = name
        self.msgs = msgs or []

        self.children = children or []

        self.checker = checker
        self.check_method = check_method
        self.variable_name = variable_name

    def __repr__(self):
        ret = "{} (*{}): {}".format(self.name, self.weight, self.value)

        if len(self.msgs):
            if len(self.msgs) == 1:
                ret += " ({})".format(self.msgs[0])
            else:
                ret += " ({!s} msgs)".format(len(self.msgs))

        if len(self.children):
            ret += " ({!s} children)".format(len(self.children))
            ret += "\n" + pprint.pformat(self.children)

        return ret

    def serialize(self):
        """
        Returns a serializable dictionary that represents the result object
        """
        return {
            "name": self.name,
            "weight": self.weight,
            "value": self.value,
            "msgs": self.msgs,
            "children": [i.serialize() for i in self.children],
        }

    def __eq__(self, other):
        return self.serialize() == other.serialize() class TestCtx(object):
    """
    Simple struct object that holds score values and messages to compile into a result
    """

    def __init__(
        self,
        category=None,
        description="",
        out_of=0,
        score=0,
        messages=None,
        variable=None,
    ):
        self.category = category or BaseCheck.LOW
        self.out_of = out_of
        self.score = score
        self.messages = messages or []
        self.description = description or ""
        self.variable = variable

    def to_result(self):
        return Result(
            self.category,
            (self.score, self.out_of),
            self.description,
            self.messages,
            variable_name=self.variable,
        )

    def assert_true(self, test, message):
        """
        Increments score if test is true otherwise appends a message
        :rtype: bool
        :return: Boolean indicating whether test condition passed or not
        """
        self.out_of += 1

        if test:
            self.score += 1
        else:
            self.messages.append(message)

        return test

    def add_failure(self, message):
        """
        Adds a failure along with a message
        :rtype: None
        """
        self.assert_true(False, message)

    def add_pass(self):
        """
        Adds a pass condition
        :rtype: None
        """
        self.assert_true(True, None) def attr_check(kvp, ds, priority, ret_val, gname=None, var_name=None):
    """
    Handles attribute checks for simple presence of an attribute, presence of
    one of several attributes, and passing a validation function.  Returns a
    status along with an error message in the event of a failure.  Mutates
    ret_val parameter

    :param tuple(str, func) or str l: the attribute being checked
    :param netCDF4 dataset ds       : dataset being checked
    :param int priority             : priority level of check
    :param list ret_val             : result to be returned
    :param str or None gname        : group name assigned to a group of attribute Results
    :param str or None var_name     : name of the variable which contains this attribute
    """

    msgs = []
    name, other = kvp
    if var_name is not None:
        display_name = "attribute {} in variable {}".format(name, var_name)
        base_context = ds.variables[var_name]
    else:
        display_name = name
        base_context = ds
    if other is None:
        res = std_check(ds, name)
        if not res:
            msgs = ["{} not present".format(display_name)]
        else:
            try:
                # see if this attribute is a string, try stripping
                # whitespace, and return an error if empty
                att_strip = base_context.getncattr(name).strip()
                if not att_strip:
                    res = False
                    msgs = ["{} is empty or completely whitespace".format(display_name)]
            # if not a string/has no strip method we should be OK
            except AttributeError:
                pass

        # gname arg allows the global attrs to be grouped together
        ret_val.append(
            Result(
                priority,
                value=res,
                name=gname if gname else name,
                msgs=msgs,
                variable_name=var_name,
            )
        )
    elif hasattr(other, "__iter__"):
        # redundant, we could easily do this with a hasattr
        # check instead
        res = std_check_in(base_context, name, other)
        if res == 0:
            msgs.append("{} not present".format(display_name))
        elif res == 1:
            msgs.append(
                "{} present, but not in expected value list ({})".format(
                    display_name, sorted(other)
                )
            )

        ret_val.append(
            Result(
                priority,
                (res, 2),
                gname if gname else name,  # groups Globals if supplied
                msgs,
                variable_name=var_name,
            )
        )
    # if we have an XPath expression, call it on the document
    elif type(other) is etree.XPath:
        # TODO: store tree instead of creating it each time?
        # no execution path for variable
        res = xpath_check(ds._root, other)
        if not res:
            msgs = ["XPath for {} not found".format(display_name)]
        ret_val.append(
            Result(
                priority, res, gname if gname else name, msgs, variable_name=var_name
            )
        )
    # check if this is a subclass of ValidationObject
    elif isinstance(other, ValidationObject):
        attr_result = maybe_get_global_attr(name, ds)
        if not attr_result[0]:
            res_tup = attr_result
        else:
            check_val = attr_result[1]
            res_tup = other.validate(name, check_val)

        msgs = [] if res_tup[1] is None else res_tup[1]

        ret_val.append(Result(priority, res_tup[0], name, msgs))
    elif isinstance(other, re_pattern_type):
        attr_result = maybe_get_global_attr(name, ds)
        if not attr_result[0]:
            return attr_result
        else:
            check_val = attr_result[1]
        if not isinstance(check_val, str):
            res = False
            msgs = ["{} must be a string".format(name)]
        elif not other.search(check_val):
            res = False
            msgs = ["{} must match regular expression {}".format(name, other)]
        else:
            res = True
            msgs = []

        ret_val.append(
            Result(priority, value=res, name=gname if gname else name, msgs=msgs)
        )

    # if the attribute is a function, call it
    # right now only supports single attribute
    # important note: current magic approach uses all functions
    # starting with "check".  Avoid naming check functions
    # starting with check if you want to pass them in with
    # a tuple to avoid them being checked more than once
    elif hasattr(other, "__call__"):
        # check that the attribute is actually present.
        # This reduces boilerplate in functions by not needing
        # to check whether the attribute is present every time
        # and instead focuses on the core functionality of the
        # test

        res = other(base_context)  # call the method on the dataset
        if not res:
            msgs = ["{} not present".format(display_name)]
            ret_val.append(
                Result(
                    priority,
                    res,
                    gname if gname else name,
                    msgs,
                    variable_name=var_name,
                )
            )
        else:
            ret_val.append(res(priority))
    # unsupported second type in second
    else:
        raise TypeError(
            "Second arg in tuple has unsupported type: {}".format(type(other))
        )

    return ret_val def check_has(priority=BaseCheck.HIGH, gname=None):
    """Decorator to wrap a function to check if a dataset has given attributes.
    :param function func: function to wrap"""

    def _inner(func):
        def _dec(s, ds):
            attr_process = kvp_convert(func(s, ds))

            ret_val = []
            # could potentially run tests in parallel if we eliminated side
            # effects on `ret_val`
            for kvp in attr_process.items():
                # function mutates ret_val
                attr_check(kvp, ds, priority, ret_val, gname)
            return ret_val

        return wraps(func)(_dec)

    return _inner class DotDict(dict):
class VariableReferenceError(Exception):
class NCGraph(object):
class StandardNameTable(object):
    class NameEntry(object):
    def __dir__(self):
    def __getattr__(self, key):
    def __setattr__(self, key, value):
    def copy(self):
    def get_safe(self, qual_key, default=None):
    def fromkeys(cls, seq, value=None):
def get_safe(dict_instance, keypath, default=None):
    def __init__(self, name: str, dataset: Dataset = None):
    def __str__(self):
    def __init__(
        self, ds, name, nc_object, self_reference_variables, reference_map=None
    ):
    def get_references(self):
    def get_dimension(self, dim):
    def get_coordinate(self, coord):
    def get_grid_mapping(self, gm):
    def __getattr__(self, key):
        def __init__(self, entrynode):
        def _get(self, entrynode, attrname, required=False):
    def __init__(self, cached_location=None):
    def __len__(self):
    def __getitem__(self, key):
    def get(self, key, default=None):
    def __contains__(self, key):
    def __iter__(self):
def download_cf_standard_name_table(version, location=None):
def create_cached_data_dir():
def units_known(units):
def units_convertible(units1, units2, reftimeistime=True):
def units_temporal(units):
def map_axes(dim_vars, reverse_map=False):
def find_coord_vars(ncds):
def is_time_variable(varname, var):
def is_vertical_coordinate(var_name, var):
def compare_unit_types(specified, reference):
def string_from_var_type(variable):
def reference_attr_variables(
    dataset: Dataset, attributes_string: str, split_by: str = None
): def get_coordinate_variables(ds):
    """
    Returns a list of variable names that identify as coordinate variables.

    A coordinate variable is a netCDF variable with exactly one dimension. The
    name of this dimension must be equivalent to the variable name.

    From CF §1.2 Terminology

    It is a one-dimensional variable with the same name as its dimension [e.g.,
    time(time) ], and it is defined as a numeric data type with values that are
    ordered monotonically. Missing values are not allowed in coordinate
    variables.

    :param netCDF4.Dataset ds: An open netCDF dataset
    """
    coord_vars = []
    for dimension in ds.dimensions:
        if dimension in ds.variables:
            if ds.variables[dimension].dimensions == (dimension,):
                coord_vars.append(dimension)
    return coord_vars @lru_cache(128)
def get_geophysical_variables(ds):
    """
    Returns a list of variable names for the variables detected as geophysical
    variables.

    :param netCDF4.Dataset nc: An open netCDF dataset
    """

    parameters = []
    for variable in ds.variables:
        if is_geophysical(ds, variable):
            parameters.append(variable)
    return parameters def get_instrument_variables(ds):
    """
    Returns a list of instrument variables

    :param netCDF4.Dataset ds: An open netCDF4 Dataset
    """
    candidates = []
    for variable in ds.variables:
        instrument = getattr(ds.variables[variable], "instrument", "")
        if instrument and instrument in ds.variables:
            if instrument not in candidates:
                candidates.append(instrument)

    instrument = getattr(ds, "instrument", "")
    if instrument and instrument in ds.variables:
        if instrument not in candidates:
            candidates.append(instrument)
    return candidates def get_z_variables(nc):
    """
    Returns a list of all variables matching definitions for Z

    :param netcdf4.dataset nc: an open netcdf dataset object
    """
    z_variables = []
    # Vertical coordinates will be identifiable by units of pressure or the
    # presence of the positive attribute with a value of up/down

    # optionally, the vertical type may be indicated by providing the
    # standard_name attribute or axis='Z'

    total_coords = get_coordinate_variables(nc) + get_auxiliary_coordinate_variables(nc)
    for coord_name in total_coords:
        if coord_name in z_variables:
            continue
        coord_var = nc.variables[coord_name]
        units = getattr(coord_var, "units", None)
        positive = getattr(coord_var, "positive", None)
        standard_name = getattr(coord_var, "standard_name", None)
        axis = getattr(coord_var, "axis", None)
        # If there are no units, we can't identify it as a vertical coordinate
        # by checking pressure or positive
        if units is not None:
            if units_convertible(units, "bar"):
                z_variables.append(coord_name)
            elif isinstance(positive, str):
                if positive.lower() in ["up", "down"]:
                    z_variables.append(coord_name)
        # if axis='Z' we're good
        if coord_name not in z_variables and axis == "Z":
            z_variables.append(coord_name)
        if coord_name not in z_variables and standard_name in (
            "depth",
            "height",
            "altitude",
        ):
            z_variables.append(coord_name)
        if (
            coord_name not in z_variables
            and standard_name in DIMENSIONLESS_VERTICAL_COORDINATES
        ):
            z_variables.append(coord_name)

    return z_variables
import re
import validators
from numbers import Number
from cf_units import Unit
from lxml.etree import XPath
from owslib.namespaces import Namespaces
from compliance_checker import base
from compliance_checker.acdd import ACDD1_3Check
from compliance_checker.base import (
    BaseCheck,
    BaseNCCheck,
    BaseSOSDSCheck,
    BaseSOSGCCheck,
    Result,
    TestCtx,
    attr_check,
    check_has,
)
from compliance_checker.cf import util as cf_util  # not to be confused with cfutil.py
from compliance_checker.cf.cf import CF1_6Check, CF1_7Check
from compliance_checker.cfutil import (
    get_coordinate_variables,
    get_geophysical_variables,
    get_instrument_variables,
    get_z_variables,
)
                    ["contributor_role should be present"],
                )
            )

        vocb_results = []
        if vocb:
            try:
                _vocbs = base.csv_splitter(vocb)
                for _vocb in _vocbs:
                    vocb_val = _vocb in self.valid_contributor_role_vocabs
                    vocb_results.append(
                        Result(
                            BaseCheck.MEDIUM,
                            vocb_val,
                            "contributor_role_vocabulary",
                            None if vocb_val else [vocb_msg.format(_vocb)],
                        )
                    )
            except TypeError as e:
                vocb_results.append(
                    Result(
                        BaseCheck.MEDIUM,
                        False,
                        "contributor_role_vocabulary",
                        [
                            "contributor_role_vocabulary '{}' must be of type 'string'".format(
                                vocb
                            )
                        ],
                    )
                )
        else:
            vocb_results.append(
                Result(
                    BaseCheck.MEDIUM,
                    False,
                    "contributor_role_vocabulary",
                    ["contributor_role_vocabulary should be present"],
                )
            )

        return role_results + vocb_results

    def check_geophysical_vars_have_attrs(self, ds):
        """
        All geophysical variables must have certain attributes.

        Parameters
        ----------
        ds: netCDF4.Dataset

        Returns
        -------
        list: list of Result objects
        """

        # get geophysical variables
        geophys_vars = get_geophysical_variables(ds)  # list of str
        results = self._check_vars_have_attrs(  # list
            ds, geophys_vars, self.geophys_check_var_attrs
        )

        return results

    def check_accuracy(self, ds):
        """
        Special check for accuracy when in the salinity context.
        https://github.com/ioos/compliance-checker/issues/839

        Parameters
        ----------
        ds: netCDF4.Dataset

        Returns
        -------
        list of Results objects
        """

        results = []
        msg = (
            "Variable '{v}' attribute 'accuracy' should have the " "same units as '{v}'"
        )
        for v in get_geophysical_variables(ds):
            _v = ds.variables[v]
            std_name = getattr(_v, "standard_name", None)
            gts_ingest = getattr(_v, "gts_ingest", None)
            if (std_name == "sea_water_practical_salinity") and (gts_ingest == "true"):
                msg = (
                    "Variable '{v}' should have an 'accuracy' attribute "
                    "that is numeric and of the same units as '{v}'"
                )
                r = isinstance(getattr(_v, "accuracy", None), Number)
            else:  # only test if exists
                r = getattr(_v, "accuracy", None) is not None

            results.append(
                Result(
                    BaseCheck.MEDIUM,
                    r,
                    "geophysical_variable:accuracy",
                    [msg.format(v=v)],
                )
            )

        return results

    def check_geospatial_vars_have_attrs(self, ds):
        """
        All geospatial variables must have certain attributes.

        Parameters
        ----------
        ds: netCDF4.Dataset

        Returns
        -------
        list: list of Result objects
        """

        return self._check_vars_have_attrs(

            ds, get_coordinate_variables(ds), self.geospat_check_var_attrs