""" Configuration file handling
"""
from collections import Mapping
import logging
import json
import os
from pathlib import Path
import yaml
from .. import defaults, validation
from . import build
logger = logging.getLogger(__name__)
TEMPLATE_FILENAME = 'config.yaml.tmpl'
TEMPLATE_FILE = Path(__file__).parent.joinpath(TEMPLATE_FILENAME)
SCHEMA_FILENAME = 'schema.json'
SCHEMA_FILE = os.path.join(os.path.dirname(__file__), SCHEMA_FILENAME)
[docs]class Config(Mapping):
""" CEDAR configuration file
"""
def __init__(self, config, schema=None):
self._config = config.copy()
self.schema = schema or self._load_schema()
self.validate()
# Mapping methods
def __getitem__(self, key):
return self._config[key]
def __iter__(self):
for key in self._config:
yield key
def __len__(self):
return len(self._config)
# Class create/serialize methods
[docs] @classmethod
def from_yaml(cls, filename, schema=None):
""" Load from a YAML configuration file
"""
with open(filename) as f:
config = yaml.safe_load(f)
return cls(config, schema=schema)
[docs] @classmethod
def from_template(cls, schema=None):
""" Load from the included YAML configuration file template
"""
return cls.from_yaml(TEMPLATE_FILE, schema=schema)
[docs] def to_yaml(self, dest=None, indent=2, sort_keys=False, **kwds):
""" Write to a YAML (file, if ``dest`` is provided)
Parameters
----------
dest : str or Path
Filename to write to. If not provided, returns a str
Returns
-------
str
Either the filename written to, or a str containing the
YAML data if ``dest`` is ``None``
"""
if dest is not None:
with open(dest, 'w') as dst:
dmp = yaml.safe_dump(self._config, stream=dst,
indent=indent, sort_keys=sort_keys,
**kwds)
return dst
else:
return yaml.safe_dump(self._config,
indent=indent, sort_keys=sort_keys,
**kwds)
# Getter-s for various objects this config describes
@staticmethod
def _load_schema(filename=SCHEMA_FILE):
with open(filename) as src:
return json.load(src)
[docs] def validate(self):
""" Validate the configuration against schema
Raises
------
ValidationError
Raised if there's an issue
"""
validation.validate_with_defaults(self._config, schema=self.schema)
[docs] def get_tracker(self):
""" Get the Tracker described by this store
"""
# Copy tracker config
cfg = self['tracker'].copy()
# Create tile grid
tile_grid = self.get_tile_grid()
# Create store
service = cfg.pop('store').lower()
if service == 'gcs':
store = self.get_gcs_store()
elif service == 'gdrive':
store = self.get_gdrive_store()
else:
raise ValueError(f'Unknown `store_service` named "{service}"')
# Create tracker
tracker = build.build_tracker(tile_grid, store, **cfg)
return tracker
[docs] def get_tile_grid(self):
""" Return the Tile Grid described by this config
Returns
-------
stems.gis.grids.TileGrid
"""
cfg = self['tile_grid'].copy()
grid_name = cfg.pop('grid_name', None)
grid_filename = cfg.pop('grid_filename', None)
return build.build_tile_grid(grid_name, grid_filename, **cfg)
[docs] def get_gcs_store(self):
""" Return a GCSStore described by this config
"""
from cedar.stores.gcs import GCSStore
cfg = self.get('gcs', {})
store = GCS.from_credentials(**kwds)
return store
[docs] def get_gdrive_store(self):
""" Return a GDriveStore described by this config
"""
from cedar.stores.gdrive import GDriveStore
cfg = self.get('gdrive', {})
store = GDriveStore.from_credentials(**cfg)
return store