import importlib.resources as imp_res
import json
[docs]class Component:
"""
The base class for housing all the tasks for a given component, such as
ocean, landice, or seaice
Attributes
----------
name : str
the name of the component
tasks : dict
A dictionary of tasks in the component with the subdirectories of the
tasks in the component as keys
steps : dict
A dictionary of steps in the component with the subdirectories of the
steps in the component as keys
cached_files : dict
A dictionary that maps from output file names in steps within tasks to
cached files in the ``polaris_cache`` database for the component. These
file mappings are read in from ``cached_files.json`` in the component.
"""
[docs] def __init__(self, name):
"""
Create a new container for the tasks for a given component
Parameters
----------
name : str
the name of the component
"""
self.name = name
# tasks are added with add_task()
self.tasks = dict()
# steps are added with add_step()
self.steps = dict()
# configs are added with add_config()
self.configs = dict()
self.cached_files = dict()
self._read_cached_files()
[docs] def add_task(self, task):
"""
Add a task to the component
Parameters
----------
task : polaris.Task
The task to add
"""
if task.subdir in self.tasks:
raise ValueError(f'A task has already been added with path '
f'{task.subdir}')
self.tasks[task.subdir] = task
[docs] def add_step(self, step):
"""
Add a step to the component
Parameters
----------
step : polaris.Step
The step to add
"""
if step.subdir in self.steps and self.steps[step.subdir] != step:
raise ValueError(f'A different step has already been added with '
f'path {step.subdir}')
self.steps[step.subdir] = step
[docs] def remove_step(self, step):
"""
Remove the given step from this component
Parameters
----------
step : polaris.Step
The step to add if adding by Step object, not subdirectory
"""
if step.subdir not in self.steps:
raise ValueError(f'step {step.name} at {step.subdir} not in the '
f'{self.name} component')
self.steps.pop(step.subdir)
[docs] def add_config(self, config):
"""
Add a shared config to the component
Parameters
----------
config : polaris.config.PolarisConfigParser
The config to add
"""
if config.filepath is None:
raise ValueError('The filepath attribute of a config must be set '
'before it can be added to a component')
if config.filepath in self.configs and \
self.configs[config.filepath] != config:
raise ValueError(f'A different shared config has already been '
f'added at {config.filepath}')
self.configs[config.filepath] = config
def configure(self, config):
"""
Configure the component
Parameters
----------
config : polaris.config.PolarisConfigParser
config options to modify
"""
pass
def _read_cached_files(self):
""" Read in the dictionary of cached files from cached_files.json """
package = f'polaris.{self.name}'
filename = 'cached_files.json'
try:
pkg_file = imp_res.files(package).joinpath(filename)
with pkg_file.open('r') as data_file:
self.cached_files = json.load(data_file)
except FileNotFoundError:
# no cached files for this core
pass