You are viewing the latest unreleased documentation 3.10.0.dev41. You can switch to a stable version.

Source code for iris.config

# Copyright Iris contributors
# This file is part of Iris and is released under the BSD license.
# See LICENSE in the root of the repository for full licensing details.
"""Provides access to Iris-specific configuration values.

The default configuration values can be overridden by creating the file
``iris/etc/site.cfg``. If it exists, this file must conform to the format
defined by :mod:`configparser`.


.. py:data:: iris.config.TEST_DATA_DIR

    Local directory where test data exists.  Defaults to "test_data"
    sub-directory of the Iris package install directory. The test data
    directory supports the subset of Iris unit tests that require data.
    Directory contents accessed via :func:`iris.tests.get_data_path`.

.. py:data:: iris.config.PALETTE_PATH

    The full path to the Iris palette configuration directory



import configparser
import contextlib
import logging
import os.path
import warnings

import iris.warnings

[docs] def get_logger(name, datefmt=None, fmt=None, level=None, propagate=None, handler=True): """Create a custom class for logging. Create a :class:`logging.Logger` with a :class:`logging.StreamHandler` and custom :class:`logging.Formatter`. Parameters ---------- name : The name of the logger. Typically this is the module filename that owns the logger. datefmt : optional The date format string of the :class:`logging.Formatter`. Defaults to ``%d-%m-%Y %H:%M:%S``. fmt : optional The additional format string of the :class:`logging.Formatter`. This is appended to the default format string ``%(asctime)s %(name)s %(levelname)s - %(message)s``. level : optional The threshold level of the logger. Defaults to ``INFO``. propagate : optional Sets the ``propagate`` attribute of the :class:`logging.Logger`, which determines whether events logged to this logger will be passed to the handlers of higher level loggers. Defaults to ``False``. handler : bool, default=True Create and attach a :class:`logging.StreamHandler` to the logger. Defaults to ``True``. Returns ------- :class:`logging.Logger`. """ if level is None: # Default logging level. level = "INFO" if propagate is None: # Default logging propagate behaviour. propagate = False # Create the named logger. logger = logging.getLogger(name) logger.setLevel(level) logger.propagate = propagate # Create and add the handler to the logger, if required. if handler: if datefmt is None: # Default date format string. datefmt = "%d-%m-%Y %H:%M:%S" # Default format string. _fmt = "%(asctime)s %(name)s %(levelname)s - %(message)s" # Append additional format string, if appropriate. fmt = _fmt if fmt is None else f"{_fmt} {fmt}" # Create a formatter. formatter = logging.Formatter(fmt=fmt, datefmt=datefmt) # Create a logging handler. handler = logging.StreamHandler() handler.setFormatter(formatter) logger.addHandler(handler) return logger
# Returns simple string options
[docs] def get_option(section, option, default=None): """Return the option value for the given section. Returns the option value for the given section, or the default value if the section/option is not present. """ value = default if config.has_option(section, option): value = config.get(section, option) return value
# Returns directory path options
[docs] def get_dir_option(section, option, default=None): """Return the directory path from the given option and section. Returns the directory path from the given option and section, or returns the given default value if the section/option is not present or does not represent a valid directory. """ path = default if config.has_option(section, option): c_path = config.get(section, option) if os.path.isdir(c_path): path = c_path else: msg = ( "Ignoring config item {!r}:{!r} (section:option) as {!r}" " is not a valid directory path." ) warnings.warn( msg.format(section, option, c_path), category=iris.warnings.IrisIgnoringWarning, ) return path
# Figure out the full path to the "iris" package. ROOT_PATH = os.path.abspath(os.path.dirname(__file__)) # The full path to the configuration directory of the active Iris instance. CONFIG_PATH = os.path.join(ROOT_PATH, "etc") # Load the optional "site.cfg" file if it exists. config = configparser.ConfigParser()[os.path.join(CONFIG_PATH, "site.cfg")]) ################## # Resource options _RESOURCE_SECTION = "Resources" TEST_DATA_DIR = get_dir_option( _RESOURCE_SECTION, "test_data_dir", default=os.path.join(os.path.dirname(__file__), "test_data"), ) # Override the data repository if the appropriate environment variable # has been set. override = os.environ.get("OVERRIDE_TEST_DATA_REPOSITORY") if override: TEST_DATA_DIR = None if os.path.isdir(os.path.expanduser(override)): TEST_DATA_DIR = os.path.abspath(override) PALETTE_PATH = get_dir_option( _RESOURCE_SECTION, "palette_path", os.path.join(CONFIG_PATH, "palette") ) # Runtime options
[docs] class NetCDF: """Control Iris NetCDF options.""" def __init__(self, conventions_override=None): """Set up NetCDF processing options for Iris. Parameters ---------- conventions_override : bool, optional Define whether the CF Conventions version (e.g. `CF-1.6`) set when saving a cube to a NetCDF file should be defined by Iris (the default) or the cube being saved. If `False` (the default), specifies that Iris should set the CF Conventions version when saving cubes as NetCDF files. If `True`, specifies that the cubes being saved to NetCDF should set the CF Conventions version for the saved NetCDF files. Examples -------- * Specify, for the lifetime of the session, that we want all cubes written to NetCDF to define their own CF Conventions versions:: iris.config.netcdf.conventions_override = True'my_cube', '')'my_second_cube', '') * Specify, with a context manager, that we want a cube written to NetCDF to define its own CF Conventions version:: with iris.config.netcdf.context(conventions_override=True):'my_cube', '') """ # Define allowed `__dict__` keys first. self.__dict__["conventions_override"] = None # Now set specific values. setattr(self, "conventions_override", conventions_override) def __repr__(self): msg = "NetCDF options: {}." # Automatically populate with all currently accepted kwargs. options = ["{}={}".format(k, v) for k, v in self.__dict__.items()] joined = ", ".join(options) return msg.format(joined) def __setattr__(self, name, value): if name not in self.__dict__: # Can't add new names. msg = "Cannot set option {!r} for {} configuration." raise AttributeError(msg.format(name, self.__class__.__name__)) if value is None: # Set an unset value to the name's default. value = self._defaults_dict[name]["default"] if self._defaults_dict[name]["options"] is not None: # Replace a bad value with a good one if there is a defined set of # specified good values. If there isn't, we can assume that # anything goes. if value not in self._defaults_dict[name]["options"]: good_value = self._defaults_dict[name]["default"] wmsg = ( "Attempting to set invalid value {!r} for " "attribute {!r}. Defaulting to {!r}." ) warnings.warn( wmsg.format(value, name, good_value), category=iris.warnings.IrisDefaultingWarning, ) value = good_value self.__dict__[name] = value @property def _defaults_dict(self): # Set this as a property so that it isn't added to `self.__dict__`. return { "conventions_override": { "default": False, "options": [True, False], }, }
[docs] @contextlib.contextmanager def context(self, **kwargs): """Allow temporary modification of the options via a context manager. Accepted kwargs are the same as can be supplied to the Option. """ # Snapshot the starting state for restoration at the end of the # contextmanager block. starting_state = self.__dict__.copy() # Update the state to reflect the requested changes. for name, value in kwargs.items(): setattr(self, name, value) try: yield finally: # Return the state to the starting state. self.__dict__.clear() self.__dict__.update(starting_state)
netcdf = NetCDF()