daggum

Author: Eoghan-Perkins

README.md

@@ -30,16 +30,31 @@ NFGDA Service directory structure:
 - /nfgda_service contains the NFGDA service logic
 - responsible for all NFGDA execution, output processing, and file management
 
-# Todo (before MSU handoff)
-
-- Guard against short jobs that run forever for some reason
-- Figure out zoom level / blank frame issue on frontend
-- Switching to a new station view pauses slide deck playthrough
-- Convert geotiff output to cloud-optimized-geotiffs
-- Code cleanup / add comments where necessary
+And then there's a redis instance living at port 6379 where all the job status and asset information is stored.
+
+# Todo before MSU handoff
+
+- [ ] Figure out zoom level / blank frame issue on frontend
+- [ ] Switching to a new station view pauses slide deck playthrough
+- [ ] Can we pretty up the landing page? Put a title on it somewhere before the research celebration?
+- [ ] Set opacity slider on frontend
+- [ ] Enhance resolution of output on frontend
+- [ ] Add a "clear" button to the map that clears all job assets from the map
+- [ ] Deliver frame time-stamps to the frontend
+- [ ] Switch to cloud-optimized geotiffs
+- [ ] Make some stuff environment variables instead of random variables everywhere
+- [ ] Discuss pixel-width of gust fronts written to output file next team meeting
+- [ ] Diff the NFGDA code used in nfgda_service with the original NFGDA code, see if there are any useful features we're missing out on or bugs we introduced
+- [ ] Backend code cleanup / add comments where necessary
 
 # "Nice to have" features
 
+- There a should probably be a warning that shows up for small numbers of assets per job (2 frames produced or less). Maybe if not enough assets are produced, the job request could automatically re-run with a larger time window?
 - Average time to job completion estimator (small addition: new counter in redis, average out)
-- Serve tiles instead of individual GeoTIFFs (big refactor)
+- Serve tiles instead of individual GeoTIFFs (big refactor, honestly might not be worth at as Cloud-optimized-geotiffs are kinda the future anyway)
 - Hash job IDs to make them unguessable, so resources can't be directly accessed via URL (little development effort, likely med/large refactor effort)
+
+# Todo after MSU handoff (futures devs read this pls)
+
+- Check that automatic asset deletion occurs within the timeframe specified (should be 24 hours)
+- Familiarize with the .env file and environment variables, and what they do

frontend/vite.config.js

@@ -7,7 +7,7 @@ export default defineConfig({
   plugins: [react(), tailwindcss()],
   server: {
     proxy: {
-      "/APIs": "http://backend:8001", //backend to work with docker, localhost w/o
+      "/apis": "http://backend:8001", //backend to work with docker, localhost w/o
     },
   },
 });

nfgda_service/nfgda_runner.py

@@ -6,14 +6,12 @@
 
 logger = logging.getLogger(__name__)
 
-MAX_NO_DATA_POLLS = int(os.getenv("MAX_NO_DATA_POLLS", "10"))
-
 class NfgdaRunner:
-    """Executes the NFGDA algorithm for a given run request."""
+    """ executes the NFGDA algorithm for a given run request. """
 
     def __init__(self, station_id: str, start_utc: str, end_utc: str, job_id: str, out_dir: str) -> None:
         """
-        Initialize the NfgdaRunner with the given parameters.
+        initialize the NfgdaRunner with the given parameters.
 
         Args:
             station_id (str): The station code.
@@ -42,20 +40,25 @@ async def run(self):
         Returns:
             bool: True if the NFGDA process completed successfully, False otherwise.
         """
-        
+
+        # get the number of consecutive no data polls to allow before killing the process
+        no_data_polls = int(os.getenv("MAX_NO_DATA_POLLS", "10"))
+
         logger.info(f"timebox parameters set to start_utc: {self.start_utc}, end_utc: {self.end_utc}")
 
+        # create a temporary config file for the algorithm
         config_path = self.create_temp_config(self.out_dir)
         logger.info("setting environment variable NFGDA_CONFIG_PATH to %s", config_path)
 
+        # if the config file couldn't be created for some reason return False
         if config_path is None:
             return False, "Failed to create config file"
 
         logger.info("running algorithm for job %s", self.job_id)
         state = {"no_data_count": 0, "fatal_error_count": 0}
 
         try:
-            # Build an env dict with the per-job config path
+            # build an env dict with the per-job config path
             env = os.environ.copy()
             env["NFGDA_CONFIG_PATH"] = config_path
 
@@ -67,41 +70,41 @@ async def run(self):
                 stderr=asyncio.subprocess.PIPE,
             )
 
-            # Stream stdout and stderr from algorithm subprocesses line-by-line in real time
+            # stream stdout and stderr from algorithm subprocesses line-by-line in real time
             stream_tasks = [
                 asyncio.create_task(self.stream_pipe(proc.stdout, "stdout")),
                 asyncio.create_task(self.monitored_stream(proc.stderr, "stderr", proc, state)),
             ]
 
-            # Wait for the algorithm to complete, with a timeout
+            # wait for the algorithm to complete, with a timeout
             try:
                 await asyncio.wait_for(proc.wait(), timeout=self.algo_timeout_seconds)
             except asyncio.TimeoutError:
-                logger.error("NFGDA algorithm timed out — killing process")
+                logger.error("NFGDA algorithm timed out. Killing process")
                 proc.kill()
                 await proc.wait()
                 return False, "NFGDA algorithm timed out"
             finally:
                 # flush remaining buffered output
                 await asyncio.gather(*stream_tasks)
 
-            # Check if the algorithm was killed due to a data gap
-            if state["no_data_count"] >= MAX_NO_DATA_POLLS:
+            # check if the algorithm was killed due to a lack of data in the nexrad S3 bucket
+            if state["no_data_count"] >= no_data_polls:
                 logger.error(
-                    "NFGDA process killed due to data gap — no scans found after %d consecutive polls",
-                    MAX_NO_DATA_POLLS,
+                    "NFGDA process killed due to data gap. No scans found after %d consecutive polls",
+                    no_data_polls,
                 )
-                return False, f"No radar data found after {MAX_NO_DATA_POLLS} polls"
+                return False, f"No radar data found after {no_data_polls} polls"
 
-            # Check if the algorithm exited with a non-zero return code
+            # check if the algorithm exited with a non-zero return code
             if proc.returncode != 0:
                 logger.error(
                     "NFGDA algorithm exited with code %d",
                     proc.returncode,
                 )
                 return False, f"an error occurred processing the algorithm. Error code: {proc.returncode}"
 
-            # Check if fatal errors were logged during processing
+            # check if fatal errors were logged during processing
             if state["fatal_error_count"] > 0:
                 logger.error(
                     "NFGDA algorithm reported %d fatal error(s) during processing",
@@ -118,15 +121,19 @@ async def run(self):
 
     @staticmethod
     def iso_to_csv_time(iso_str: str) -> str:
-        """Convert an ISO 8601 timestamp (e.g. '2024-07-07T01:22:24Z') to the
+        """ convert an ISO 8601 timestamp (e.g. '2024-07-07T01:22:24Z') to the
         comma-separated format that NFGDA config asks for (e.g. 'year,month,day,hour,minute,second').
+
+        Why tf did they format their timestamps like this???
         """
-        dt = datetime.strptime(iso_str, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
-        return f"{dt.year},{dt.month},{dt.day},{dt.hour},{dt.minute},{dt.second}"
+        # convert the iso string to a datetime object
+        date_time = datetime.strptime(iso_str, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
+        return f"{date_time.year},{date_time.month},{date_time.day},{date_time.hour},{date_time.minute},{date_time.second}"
 
     def create_temp_config(self, out_dir: str) -> str:
-        """Create a temporary NFGDA config file. Returns the path to the file."""
+        """ create a temporary NFGDA config file. Returns the path to the file."""
 
+        # brief check against the config file shipped w/ the original algo for SnG
         if not os.path.exists("/app/scripts/NFGDA.ini"):
             logger.warning("NFGDA.ini default config not found (proceeding anyway)")
 
@@ -135,6 +142,7 @@ def create_temp_config(self, out_dir: str) -> str:
         logger.info("config times: start=%s -> %s, end=%s -> %s",
                      self.start_utc, csv_start, self.end_utc, csv_end)
 
+        # write the config file out
         with tempfile.NamedTemporaryFile("w", delete=False, suffix=".ini", prefix="nfgda_") as f:
             f.write(f"""[Settings]
                 radar_id = {self.station_id}
@@ -165,13 +173,15 @@ async def stream_pipe(stream, label: str):
 
     @staticmethod
     async def monitored_stream(stream, label: str, proc, state: dict):
-        """Read lines and, for stderr, count consecutive no-data polls.
+        """ read logs, monitor for no data and fatal errors. Kill wonky processes
 
         Args:
             stream: asyncio subprocess stream (stdout or stderr).
-            MAX_NO_DATA_POLLS: Kill the process after this many consecutive
+            no_data_polls: Kill the process after this many consecutive
                                "no new scans found" messages.
         """
+
+        # read the log stream lines and check for patterns that indicate error
         while True:
             line = await stream.readline()
             if not line:
@@ -180,11 +190,11 @@ async def monitored_stream(stream, label: str, proc, state: dict):
             logger.info("[NFGDA_Host %s] %s", label, text)
 
             if label == "stderr":
-                if "no new scans found" in text:
+                if "no new scans found" in text:        
                     state["no_data_count"] += 1
-                    if state["no_data_count"] >= MAX_NO_DATA_POLLS:
+                    if state["no_data_count"] >= no_data_polls:
                         logger.error(
-                            "no data found after %d consecutive polls — killing process",
+                            "no data found after %d consecutive polls. Killing process",
                             state["no_data_count"],
                         )
                         proc.kill()

nfgda_service/nfgda_service.py

@@ -6,8 +6,7 @@
 
 
 class NfgdaService:
-    """High-level service that orchestrates a single NFGDA run, including
-    job lifecycle updates in Redis."""
+    """ high-level service that orchestrates a single NFGDA run, including job lifecycle updates in Redis. """
 
     def __init__(self, redis_client, job_id: str, job_fields: dict, out_dir: str) -> None:
         self.redis_client = redis_client
@@ -17,7 +16,7 @@ def __init__(self, redis_client, job_id: str, job_fields: dict, out_dir: str) ->
         self.out_dir = out_dir
 
     async def run(self) -> None:
-        """Execute the NFGDA algorithm and update job status in Redis."""
+        """ execute the NFGDA algorithm and update job status in Redis. """
         try:
             self.redis_client.hset(self.job_key, mapping={"status": "PROCESSING"})
 
@@ -34,13 +33,15 @@ async def run(self) -> None:
             )
             success, message = await runner.run()
 
+            # update job status in redis
             if success:
-                logger.info("Algorithm processing for job %s completed successfully", self.job_id)
+                logger.info("algorithm processing for job %s completed successfully", self.job_id)
             else:
+                # no, this is patrick
                 self.redis_client.hset(self.job_key, mapping={"status": "FAILED", "error_message": message})
-                logger.warning("Job %s failed (runner returned falsy)", self.job_id)
-                logger.warning("Error message: %s", message)
+                logger.warning("job %s failed (runner returned falsy)", self.job_id)
+                logger.warning("error message: %s", message)
 
         except Exception as e:
             self.redis_client.hset(self.job_key, mapping={"status": "FAILED"})
-            logger.exception("Job %s failed with exception: %s", self.job_id, e)
+            logger.exception("job %s failed with exception: %s", self.job_id, e)

nfgda_service/process_output.py

@@ -252,13 +252,13 @@ def project_data(npz_path: str, radar_lat: float, radar_lon: float, out_dir: str
     )
 
     # ---------------------------
-    # Write final GeoTIFF
+    # Write final Cloud-Optimized GeoTIFF
     # ---------------------------
-    driver = gdal.GetDriverByName("GTiff")
+    driver = gdal.GetDriverByName("COG")
     driver.CreateCopy(
         final_tif,
         warped_ds,
-        options=["COMPRESS=DEFLATE", "TILED=YES"]
+        options=["COMPRESS=DEFLATE", "OVERVIEWS=IGNORE_EXISTING"]
     )
 
     warped_ds = None