Vocabularies

Controlled vocabularies for envlib metadata fields.

Each vocabulary is a bundled JSON file in this package directory. A user-level overlay directory (~/.envlib/vocabularies/) takes precedence per-file when present — :func:refresh writes there, never into the installed package.

Vocabulary semantics:

• Upstream-sourced lists (variable membership, standard_name) are refreshable from their APIs via :func:refresh.
• The envlib-curated (variable, feature) -> CF standard_name mapping inside variable.json is hand-maintained and NEVER regenerated by refresh — refresh only reports new/removed upstream terms for manual curation.
• envlib-defined vocabularies (feature, method, processing_level, aggregation_statistic, frequency_interval, license) have no upstream source; refresh does not touch them.

list(field)

Return the valid canonical values for a vocabulary field.

Parameters:

Name	Type	Description	Default
field	str	One of the envlib vocabulary fields (e.g. 'variable', 'feature').	required

Returns:

Type	Description
list[str]	List of canonical values. For frequency_interval this is the canonical
list[str]	codes only — input aliases ('24h', '60min') are not included.

Source code in envlib/vocabularies/__init__.py

def list(field: str) -> builtins.list[str]:  # noqa: A001 - public API name set by the design plan
    """Return the valid canonical values for a vocabulary field.

    Args:
        field: One of the envlib vocabulary fields (e.g. ``'variable'``, ``'feature'``).

    Returns:
        List of canonical values. For ``frequency_interval`` this is the canonical
        codes only — input aliases (``'24h'``, ``'60min'``) are not included.
    """
    return _names(field)

canonical(field, value)

Resolve user input to the canonical vocabulary value.

Matching is case-insensitive after whitespace stripping; frequency_interval additionally resolves the closed input-alias table ('24h' -> 'day', '60min' -> '1h'). Raises ValueError when the value is not in the vocabulary.

Source code in envlib/vocabularies/__init__.py

def canonical(field: str, value: str) -> str:
    """Resolve user input to the canonical vocabulary value.

    Matching is case-insensitive after whitespace stripping; ``frequency_interval``
    additionally resolves the closed input-alias table (``'24h'`` -> ``'day'``,
    ``'60min'`` -> ``'1h'``). Raises ``ValueError`` when the value is not in the
    vocabulary.
    """
    _check_field(field)
    if not isinstance(value, str):
        msg = f'{field} value must be a str, not {type(value).__name__}.'
        raise TypeError(msg)
    key = value.strip().lower()
    table = _lookup(field)
    if key not in table:
        msg = f'{value!r} is not a valid {field}; see envlib.vocabularies.list({field!r}).'
        raise ValueError(msg)
    return table[key]

is_valid(field, value)

Whether value is acceptable input for the vocabulary field (aliases included).

Source code in envlib/vocabularies/__init__.py

def is_valid(field: str, value: str) -> bool:
    """Whether ``value`` is acceptable input for the vocabulary field (aliases included)."""
    try:
        canonical(field, value)
    except (ValueError, TypeError):
        return False
    return True

get_cf_standard_names(variable, feature)

CF standard_name candidates envlib considers for a (variable, feature) pair.

Parameters:

Name	Type	Description	Default
variable	str	envlib variable (CV member; raises ValueError if unknown).	required
feature	str	envlib feature (CV member; raises ValueError if unknown).	required

Returns:

Type	Description
list[str] | None	Ordered candidate list — the FIRST entry is the curated default that
list[str] | None	envlib auto-populates at validate/register time; an empty list means
list[str] | None	"curated: no applicable standard name" (common for freshwater
list[str] | None	water-quality variables); None means the pair is not yet curated
list[str] | None	(envlib warns but does not block).

Source code in envlib/vocabularies/__init__.py

def get_cf_standard_names(variable: str, feature: str) -> builtins.list[str] | None:
    """CF standard_name candidates envlib considers for a (variable, feature) pair.

    Args:
        variable: envlib variable (CV member; raises ``ValueError`` if unknown).
        feature: envlib feature (CV member; raises ``ValueError`` if unknown).

    Returns:
        Ordered candidate list — the FIRST entry is the curated default that
        envlib auto-populates at validate/register time; an empty list means
        "curated: no applicable standard name" (common for freshwater
        water-quality variables); ``None`` means the pair is not yet curated
        (envlib warns but does not block).
    """
    var = canonical('variable', variable)
    feat = canonical('feature', feature)
    entry = _load('variable')['entries'][var]
    cf = entry.get('cf')
    if cf is None:
        return None
    candidates = cf.get(feat)
    if candidates is None:
        return None
    return [c['standard_name'] for c in candidates]

frequency_entry(code)

Return the frequency_interval table entry for a canonical or alias code.

The entry dict has keys name (canonical code), kind ('fixed' | 'calendar'), seconds (int, or None for calendar codes), aliases.

Source code in envlib/vocabularies/__init__.py

def frequency_entry(code: str) -> dict:
    """Return the frequency_interval table entry for a canonical or alias code.

    The entry dict has keys ``name`` (canonical code), ``kind`` (``'fixed'`` |
    ``'calendar'``), ``seconds`` (int, or None for calendar codes), ``aliases``.
    """
    canon = canonical('frequency_interval', code)
    for entry in _load('frequency_interval')['entries']:
        if entry['name'] == canon:
            return entry
    msg = f'frequency_interval table entry missing for {canon!r} (corrupt vocabulary file?).'
    raise RuntimeError(msg)

refresh(field=None, _target_dir=None)

Refresh upstream-sourced vocabularies from their APIs.

Fetches the ODM2 variablename list and/or the current CF standard-name table and writes updated vocabulary files to the user overlay directory (~/.envlib/vocabularies/) — never into the installed package. For variable, only ODM2-sourced membership is updated (additively — upstream removals are reported, never deleted); curated cf mappings and envlib extensions are never touched.

Parameters:

Name	Type	Description	Default
field	str | None	'variable', 'standard_name', or None for both.	None
_target_dir	str | Path | None	Internal/dev hook — write somewhere other than the user overlay dir (used to generate the bundled files).	None

Returns:

Type	Description
dict	Report dict keyed by refreshed field (counts, added/removed terms,
dict	snake-case collisions, extension conflicts).

Source code in envlib/vocabularies/__init__.py

def refresh(field: str | None = None, _target_dir: str | pathlib.Path | None = None) -> dict:
    """Refresh upstream-sourced vocabularies from their APIs.

    Fetches the ODM2 variablename list and/or the current CF standard-name table
    and writes updated vocabulary files to the user overlay directory
    (``~/.envlib/vocabularies/``) — never into the installed package. For
    ``variable``, only ODM2-sourced membership is updated (additively — upstream
    removals are reported, never deleted); curated ``cf`` mappings and envlib
    extensions are never touched.

    Args:
        field: ``'variable'``, ``'standard_name'``, or None for both.
        _target_dir: Internal/dev hook — write somewhere other than the user
            overlay dir (used to generate the bundled files).

    Returns:
        Report dict keyed by refreshed field (counts, added/removed terms,
        snake-case collisions, extension conflicts).
    """
    refreshable = ('standard_name', 'variable')
    fields: tuple[str, ...]
    if field is None:
        fields = refreshable
    elif field in refreshable:
        fields = (field,)
    else:
        _check_field(field)
        msg = f'{field!r} has no upstream source to refresh from.'
        raise ValueError(msg)

    target_dir = pathlib.Path(_target_dir).expanduser() if _target_dir is not None else USER_DIR
    target_dir.mkdir(parents=True, exist_ok=True)

    report = {}
    for f in fields:
        if f == 'standard_name':
            report[f] = _refresh_standard_name(target_dir)
        else:
            report[f] = _refresh_variable(target_dir)
    clear_cache()
    return report

clear_cache()

Clear the in-memory vocabulary caches (called automatically by :func:refresh).

Source code in envlib/vocabularies/__init__.py

def clear_cache():
    """Clear the in-memory vocabulary caches (called automatically by :func:`refresh`)."""
    _cache.clear()
    _lookup_cache.clear()