Change to run grib_to_netcdf on skookum

grib_to_netcdf, after refactor to use HRDPS continental grid subdomain that
is cropped to just what we need to the SalishSeaCast domain, has a much
smaller memory demand. So, the dask cluster on salish is no longer required.

config/nowcast.yaml

@@ -1188,7 +1188,6 @@ zmq:
         "salish.eos.ubc.ca:5580", "salish.eos.ubc.ca:5581", "salish.eos.ubc.ca:5582",
         "salish.eos.ubc.ca:5583", "salish.eos.ubc.ca:5584", "salish.eos.ubc.ca:5585",
       ]
-      grib_to_netcdf: "salish.eos.ubc.ca:5562"
 
 
 message registry:

nowcast/next_workers.py

@@ -227,19 +227,11 @@ def after_crop_gribs(msg, config, checklist):
     }
     if msg.type == "success 06":
         next_workers["success 06"].append(
-            NextWorker(
-                "nowcast.workers.grib_to_netcdf",
-                args=["forecast2"],
-                host="salish-nowcast",
-            )
+            NextWorker("nowcast.workers.grib_to_netcdf", args=["forecast2"])
         )
     if msg.type == "success 12":
         next_workers["success 12"].append(
-            NextWorker(
-                "nowcast.workers.grib_to_netcdf",
-                args=["nowcast+"],
-                host="salish-nowcast",
-            )
+            NextWorker("nowcast.workers.grib_to_netcdf", args=["nowcast+"])
         )
     return next_workers[msg.type]
 

nowcast/workers/grib_to_netcdf.py

@@ -32,7 +32,6 @@
 from pathlib import Path
 
 import arrow
-import dask.distributed
 import numpy
 from nemo_nowcast import NowcastWorker
 import xarray
@@ -110,7 +109,6 @@ def grib_to_netcdf(parsed_args, config, *args):
         # So, we compensate for that here.
         acpc_index = var_names.index(["APCP_Sfc", "unknown", "precip"])
         var_names[acpc_index] = ["APCP_Sfc", "t", "precip"]
-    dask_client = dask.distributed.Client(config["weather"]["dask cluster"])
     match run_type:
         case "nowcast+":
             logger.info(
@@ -224,7 +222,6 @@ def grib_to_netcdf(parsed_args, config, *args):
                 fcst=True,
             )
             _update_checklist(nc_file, checklist, fcst=True)
-    dask_client.close()
     return checklist
 
 

tests/test_config.py

@@ -294,7 +294,6 @@ def test_log_aggregator_ports(self, prod_config):
                 "salish.eos.ubc.ca:5584",
                 "salish.eos.ubc.ca:5585",
             ],
-            "grib_to_netcdf": "salish.eos.ubc.ca:5562",
         }
 
 

tests/test_next_workers.py

@@ -443,11 +443,7 @@ def mock_now():
         workers = next_workers.after_crop_gribs(
             Message("crop_gribs", "success 06"), config, checklist
         )
-        expected = [
-            NextWorker(
-                "nowcast.workers.grib_to_netcdf", ["forecast2"], host="salish-nowcast"
-            ),
-        ]
+        expected = [NextWorker("nowcast.workers.grib_to_netcdf", ["forecast2"])]
         assert workers == expected
 
     def test_success_12(self, config, checklist, monkeypatch):
@@ -459,11 +455,7 @@ def mock_now():
         workers = next_workers.after_crop_gribs(
             Message("crop_gribs", "success 12"), config, checklist
         )
-        expected = [
-            NextWorker(
-                "nowcast.workers.grib_to_netcdf", ["nowcast+"], host="salish-nowcast"
-            ),
-        ]
+        expected = [NextWorker("nowcast.workers.grib_to_netcdf", ["nowcast+"])]
         assert workers == expected
 
 

tests/workers/test_grib_to_netcdf.py

@@ -69,7 +69,6 @@ def config(base_config):
 
                   ops dir: forcing/atmospheric/continental2.5/nemo_forcing/
                   file template: "hrdps_{:y%Ym%md%d}.nc"
-                  dask cluster: tcp://142.103.36.12:4386
                 """
             )
         )
@@ -193,7 +192,6 @@ def test_weather_section(self, prod_config):
             == "/results/forcing/atmospheric/continental2.5/nemo_forcing/"
         )
         assert weather["file template"] == "hrdps_{:y%Ym%md%d}.nc"
-        assert weather["dask cluster"] == "tcp://142.103.36.12:4386"
         assert (
             weather["monitoring image"]
             == "/results/nowcast-sys/figures/monitoring/wg.png"
@@ -239,20 +237,6 @@ def test_failure(self, run_type, caplog):
 class TestGribToNetcdf:
     """Unit test for grib_to_netcdf() function."""
 
-    @staticmethod
-    @pytest.fixture
-    def mock_dask_client(monkeypatch):
-        def _mock_dask_client(address):
-            class MockClient:
-                def close(self):
-                    pass
-
-            return MockClient()
-
-        monkeypatch.setattr(
-            grib_to_netcdf.dask.distributed, "Client", _mock_dask_client
-        )
-
     @staticmethod
     @pytest.fixture
     def mock_open_dataset(monkeypatch):
@@ -336,7 +320,6 @@ def test_log_messages(
         self,
         run_type,
         full_grid,
-        mock_dask_client,
         mock_open_dataset,
         mock_calc_nemo_var_ds,
         mock_combine_by_coords,
@@ -365,7 +348,6 @@ def test_log_messages(
     def test_nowcast_checklist(
         self,
         full_grid,
-        mock_dask_client,
         mock_open_dataset,
         mock_calc_nemo_var_ds,
         mock_combine_by_coords,
@@ -399,7 +381,6 @@ def test_nowcast_checklist(
     def test_forecast2_checklist(
         self,
         full_grid,
-        mock_dask_client,
         mock_open_dataset,
         mock_calc_nemo_var_ds,
         mock_combine_by_coords,