import time
from compass.model import run_model
from compass.step import Step
[docs]
class Forward(Step):
"""
A step for performing forward MPAS-Ocean runs as part of buttermilk bay
test cases.
"""
[docs]
def __init__(self, test_case, resolution,
name,
coord_type='single_layer',
wetdry='standard'):
"""
Create a new test case
Parameters
----------
test_case : compass.TestCase
The test case this step belongs to
resolution : float
The resolution of the test case
name : str
The name of the test case
subdir : str, optional
The subdirectory for the step. The default is ``name``
coord_type : str, optional
Vertical coordinate configuration
"""
self.resolution = resolution
super().__init__(test_case=test_case, name=name)
self.add_namelist_file('compass.ocean.tests.buttermilk_bay',
'namelist.forward')
res_name = f'{resolution}m'
self.add_namelist_file('compass.ocean.tests.buttermilk_bay',
f'namelist.{coord_type}.forward')
if wetdry == 'subgrid':
self.add_namelist_file('compass.ocean.tests.buttermilk_bay',
'namelist.subgrid.forward')
self.add_streams_file('compass.ocean.tests.buttermilk_bay',
'streams.subgrid.forward')
else:
self.add_streams_file('compass.ocean.tests.buttermilk_bay',
'streams.forward')
input_path = f'../initial_state_{res_name}'
self.add_input_file(filename='mesh.nc',
target=f'{input_path}/culled_mesh.nc')
self.add_input_file(filename='init.nc',
target=f'{input_path}/ocean.nc')
self.add_input_file(filename='graph.info',
target=f'{input_path}/culled_graph.info')
self.add_input_file(filename='forcing.nc',
target=f'{input_path}/init_mode_forcing_data.nc')
self.add_model_as_input()
self.add_output_file(filename='output.nc')
[docs]
def setup(self):
"""
Set namelist options based on config options
"""
dt = self.get_dt()
self.add_namelist_options({'config_dt': dt})
self._get_resources()
def constrain_resources(self, available_cores):
"""
Update resources at runtime from config options
"""
self._get_resources()
super().constrain_resources(available_cores)
[docs]
def run(self):
"""
Run this step of the testcase
"""
# update dt in case the user has changed dt_per_m
dt = self.get_dt()
self.update_namelist_at_runtime(options={'config_dt': dt},
out_name='namelist.ocean')
run_model(self)
[docs]
def get_dt(self):
"""
Get the time step
Returns
-------
dt : str
the time step in HH:MM:SS
"""
config = self.config
# dt is proportional to resolution
dt_per_m = config.getfloat('buttermilk_bay', 'dt_per_m')
dt = dt_per_m * self.resolution
# https://stackoverflow.com/a/1384565/7728169
dt = time.strftime('%H:%M:%S', time.gmtime(dt))
return dt
def _get_resources(self):
""" get the these properties from the config options """
config = self.config
self.ntasks = config.getint('buttermilk_bay',
f'{self.resolution}m_ntasks')
self.min_tasks = config.getint('buttermilk_bay',
f'{self.resolution}m_min_tasks')
self.openmp_threads = 1