Skip to content

Commit

Permalink
Change to run wwatch3 concurrent with nowcast-green
Browse files Browse the repository at this point in the history 
Always launch make_ww3_*_file workers after NEMO forecast finishes instead of
using a config flag to control whether they launch then or after nowcast-green
runs. Concurrency is possible because we have changed to running wwatch3 on
different VMs to NEMO.
  • Loading branch information
@douglatornell
douglatornell committed Nov 14, 2023
1 parent c17e5a8 commit 3e332b2
Show file tree
Hide file tree
Showing 4 changed files with 33 additions and 172 deletions.
4 changes: 0 additions & 4 deletions config/nowcast.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -749,10 +749,6 @@ run:
wave forecasts:
# Compute host to run wave forecast on
host: arbutus.cloud-nowcast
# NEMO run to run wwatch3 forecast run after
# User 'after forecast' during storm surge season and 'after nowcast-green`
# the rest of the year
run when: after nowcast-green
# Directory on compute host where files (e.g. ww3_*.inp, mod_def.ww3) and
# directories (e,g. wind/ current/) necessary to prepare the wwatch3 runs
# are stored
Expand Down
80 changes: 26 additions & 54 deletions nowcast/next_workers.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -811,33 +811,32 @@ def after_watch_NEMO(msg, config, checklist):
]
)
if run_type == "forecast":
if wave_forecast_after == "forecast":
host_name = config["wave forecasts"]["host"]
next_workers[msg.type].extend(
[
NextWorker(
"nowcast.workers.make_ww3_wind_file",
args=[
host_name,
"forecast",
"--run-date",
msg.payload[run_type]["run date"],
],
host=host_name,
),
NextWorker(
"nowcast.workers.make_ww3_current_file",
args=[
host_name,
"forecast",
"--run-date",
msg.payload[run_type]["run date"],
],
host=host_name,
),
]
)
race_condition_workers = {"make_ww3_wind_file", "make_ww3_current_file"}
host_name = config["wave forecasts"]["host"]
next_workers[msg.type].extend(
[
NextWorker(
"nowcast.workers.make_ww3_wind_file",
args=[
host_name,
"forecast",
"--run-date",
msg.payload[run_type]["run date"],
],
host=host_name,
),
NextWorker(
"nowcast.workers.make_ww3_current_file",
args=[
host_name,
"forecast",
"--run-date",
msg.payload[run_type]["run date"],
],
host=host_name,
),
]
)
race_condition_workers = {"make_ww3_wind_file", "make_ww3_current_file"}
for host in config["run"]["enabled hosts"]:
if not config["run"]["enabled hosts"][host]["shared storage"]:
next_workers[msg.type].append(
Expand All @@ -852,33 +851,6 @@ def after_watch_NEMO(msg, config, checklist):
)
)
if run_type == "nowcast-green":
if wave_forecast_after == "nowcast-green":
host_name = config["wave forecasts"]["host"]
next_workers[msg.type].extend(
[
NextWorker(
"nowcast.workers.make_ww3_wind_file",
args=[
host_name,
"forecast",
"--run-date",
msg.payload[run_type]["run date"],
],
host=host_name,
),
NextWorker(
"nowcast.workers.make_ww3_current_file",
args=[
host_name,
"forecast",
"--run-date",
msg.payload[run_type]["run date"],
],
host=host_name,
),
]
)
race_condition_workers = {"make_ww3_wind_file", "make_ww3_current_file"}
for host in config["run"]["enabled hosts"]:
run_types = config["run"]["enabled hosts"][host]["run types"]
if "nowcast-dev" in run_types:
Expand Down
118 changes: 7 additions & 111 deletions tests/test_next_workers.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -127,14 +127,6 @@ def checklist():
return {}


class TestConfig:
"""Unit tests for production YAML config file elements related to next_workers."""

def test_wave_forecasts_run_when(self, prod_config):
wave_fcsts = prod_config["wave forecasts"]
assert wave_fcsts["run when"] == "after nowcast-green"


class TestAfterWorkerFunctions:
"""Unit test to confirm that all worker modules have a corresponding after_*() function
in the next_workers module.
Expand Down Expand Up @@ -1121,16 +1113,14 @@ def test_success_nowcast(self, config, checklist):
def test_success_forecast_launch_make_ww3_wind_file_forecast(
self, config, checklist, monkeypatch
):
"""storm surge season case of wwatch3 running after NEMO forecast"""
monkeypatch.setitem(config["wave forecasts"], "run when", "after forecast")
workers, race_condition_workers = next_workers.after_watch_NEMO(
Message(
"watch_NEMO",
"success forecast",
{
"forecast": {
"host": "arbutus.cloud",
"run date": "2023-03-16",
"run date": "2023-11-14",
"completed": True,
}
},
Expand All @@ -1140,7 +1130,7 @@ def test_success_forecast_launch_make_ww3_wind_file_forecast(
)
expected = NextWorker(
"nowcast.workers.make_ww3_wind_file",
args=["arbutus.cloud", "forecast", "--run-date", "2023-03-16"],
args=["arbutus.cloud", "forecast", "--run-date", "2023-11-14"],
host="arbutus.cloud",
)
assert workers[0] == expected
Expand All @@ -1149,16 +1139,14 @@ def test_success_forecast_launch_make_ww3_wind_file_forecast(
def test_success_forecast_launch_make_ww3_current_file_forecast(
self, config, checklist, monkeypatch
):
"""storm surge season case of wwatch3 running after NEMO forecast"""
monkeypatch.setitem(config["wave forecasts"], "run when", "after forecast")
workers, race_condition_workers = next_workers.after_watch_NEMO(
Message(
"watch_NEMO",
"success forecast",
{
"forecast": {
"host": "arbutus.cloud",
"run date": "2023-03-16",
"run date": "2023-11-14",
"completed": True,
}
},
Expand All @@ -1168,50 +1156,13 @@ def test_success_forecast_launch_make_ww3_current_file_forecast(
)
expected = NextWorker(
"nowcast.workers.make_ww3_current_file",
args=["arbutus.cloud", "forecast", "--run-date", "2023-03-16"],
args=["arbutus.cloud", "forecast", "--run-date", "2023-11-14"],
host="arbutus.cloud",
)
assert workers[1] == expected
assert race_condition_workers == {"make_ww3_wind_file", "make_ww3_current_file"}

def test_success_forecast_ww3_after_nowcast_green_launch_upload_forcing_turbidity(
self, config, checklist, monkeypatch
):
workers = next_workers.after_watch_NEMO(
Message(
"watch_NEMO",
"success forecast",
{
"forecast": {
"host": "arbutus.cloud",
"run date": "2023-07-15",
"completed": True,
}
},
),
config,
checklist,
)

expected = [
NextWorker(
"nowcast.workers.upload_forcing",
args=["arbutus.cloud", "turbidity", "--run-date", "2023-07-15"],
),
NextWorker(
"nowcast.workers.upload_forcing",
args=["orcinus", "turbidity", "--run-date", "2023-07-15"],
),
]
assert workers[0:2] == expected
not_expected = NextWorker(
"nowcast.workers.upload_forcing",
args=["salish-nowcast", "turbidity"],
host="localhost",
)
assert not_expected not in workers

def test_success_forecast_ww3_after_forecast_launch_upload_forcing_turbidity(
def test_success_forecast_launch_upload_forcing_turbidity(
self, config, checklist, monkeypatch
):
monkeypatch.setitem(config["wave forecasts"], "run when", "after forecast")
Expand Down Expand Up @@ -1325,64 +1276,10 @@ def test_success_forecast2_launch_make_ww3_current_file_forecast2(
assert workers[1] == expected
assert race_condition_workers == {"make_ww3_wind_file", "make_ww3_current_file"}

def test_success_nowcast_green_launch_make_ww3_wind_file_forecast(
self, config, checklist, monkeypatch
):
monkeypatch.setitem(config["wave forecasts"], "run when", "after nowcast-green")
workers, race_condition_workers = next_workers.after_watch_NEMO(
Message(
"watch_NEMO",
"success nowcast-green",
{
"nowcast-green": {
"host": "arbutus.cloud",
"run date": "2023-03-16",
"completed": True,
}
},
),
config,
checklist,
)
expected = NextWorker(
"nowcast.workers.make_ww3_wind_file",
args=["arbutus.cloud", "forecast", "--run-date", "2023-03-16"],
host="arbutus.cloud",
)
assert workers[0] == expected
assert race_condition_workers == {"make_ww3_wind_file", "make_ww3_current_file"}

def test_success_nowcast_green_launch_make_ww3_current_file_forecast(
self, config, checklist, monkeypatch
):
monkeypatch.setitem(config["wave forecasts"], "run when", "after nowcast-green")
workers, race_condition_workers = next_workers.after_watch_NEMO(
Message(
"watch_NEMO",
"success nowcast-green",
{
"nowcast-green": {
"host": "arbutus.cloud",
"run date": "2023-03-16",
"completed": True,
}
},
),
config,
checklist,
)
expected = NextWorker(
"nowcast.workers.make_ww3_current_file",
args=["arbutus.cloud", "forecast", "--run-date", "2023-03-16"],
host="arbutus.cloud",
)
assert workers[1] == expected
assert race_condition_workers == {"make_ww3_wind_file", "make_ww3_current_file"}

def test_success_nowcast_green_launch_make_forcing_links_nowcastp_shared_storage(
self, config, checklist
):
workers, race_condition_workers = next_workers.after_watch_NEMO(
workers = next_workers.after_watch_NEMO(
Message(
"watch_NEMO",
"success nowcast-green",
Expand All @@ -1402,8 +1299,7 @@ def test_success_nowcast_green_launch_make_forcing_links_nowcastp_shared_storage
args=["salish", "nowcast+", "--shared-storage"],
host="localhost",
)
assert expected in workers
assert race_condition_workers == {"make_ww3_wind_file", "make_ww3_current_file"}
assert workers[0] == expected

@pytest.mark.parametrize(
"msg",
Expand Down
3 changes: 0 additions & 3 deletions tests/workers/test_make_ww3_current_file.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -163,9 +163,6 @@ def test_wave_forecasts_section(self, prod_config):
)
assert wave_forecasts["current file template"] == "SoG_current_{yyyymmdd}.nc"

def test_run_when(self, prod_config):
assert prod_config["wave forecasts"]["run when"] == "after nowcast-green"

def test_run_types_section(self, prod_config):
run_types = prod_config["run types"]
assert run_types["nowcast"]["mesh mask"] == "mesh_mask201702.nc"
Expand Down

0 comments on commit 3e332b2

Please sign in to comment.