add .status property to check if storm is currently being updated in real-time (#46)

* add `.status` property to check if storm is currently being updated in real-time

* fix issues with backwards logic

* handle case where the end date of a historical storm is `NaT`

Author: Zach Burnett

stormevents/stormevent.py

@@ -1,4 +1,6 @@
 from datetime import datetime
+from datetime import timedelta
+from enum import Enum
 from functools import lru_cache
 from os import PathLike
 from typing import List
@@ -24,13 +26,17 @@
 from stormevents.nhc import VortexTrack
 from stormevents.nhc.atcf import ATCF_Advisory
 from stormevents.nhc.atcf import ATCF_FileDeck
-from stormevents.nhc.atcf import ATCF_Mode
 from stormevents.usgs import usgs_flood_storms
 from stormevents.usgs import USGS_StormEvent
 from stormevents.utilities import relative_to_time_interval
 from stormevents.utilities import subset_time_interval
 
 
+class StormStatus(Enum):
+    HISTORICAL = "historical"
+    REALTIME = "realtime"
+
+
 class StormEvent:
     """
     The ``StormEvent`` class can be used to retrieve data
@@ -250,6 +256,18 @@ def __data_end(self) -> datetime:
             data_end = VortexTrack.from_storm_name(self.name, self.year).end_date
         return data_end
 
+    @property
+    def status(self) -> StormStatus:
+        entry = self.__entry
+        age = datetime.today() - entry["end_date"]
+        if pandas.isna(entry["end_date"]) or age < timedelta(days=1):
+            if datetime.today() - entry["start_date"] > timedelta(days=30):
+                return StormStatus.HISTORICAL
+            else:
+                return StormStatus.REALTIME
+        else:
+            return StormStatus.HISTORICAL
+
     def track(
         self,
         start_date: datetime = None,

tests/test_stormevent.py

@@ -2,11 +2,14 @@
 from datetime import datetime
 from datetime import timedelta
 
+import pandas
 import pytest
 import shapely
 from shapely.geometry import box
 
+from stormevents.nhc import nhc_storms
 from stormevents.stormevent import StormEvent
+from stormevents.stormevent import StormStatus
 from tests import check_reference_directory
 from tests import OUTPUT_DIRECTORY
 from tests import REFERENCE_DIRECTORY
@@ -212,3 +215,27 @@ def test_storm_event_coops_product_within_region(florence2018):
 
     assert null_tidal_data["t"].sizes == {}
     assert east_coast_tidal_data.sizes == {"nos_id": 112, "t": 1}
+
+
+def test_status():
+    florence2018 = StormEvent("florence", 2018)
+    paine2016 = StormEvent.from_nhc_code("EP172016")
+    henri2021 = StormEvent.from_usgs_id(310)
+    ida2021 = StormEvent("ida", 2021)
+
+    assert florence2018.status == StormStatus.HISTORICAL
+    assert paine2016.status == StormStatus.HISTORICAL
+    assert henri2021.status == StormStatus.HISTORICAL
+    assert ida2021.status == StormStatus.HISTORICAL
+
+    storms = nhc_storms()
+    latest_storm_entry = storms.iloc[-1]
+    latest_storm = StormEvent.from_nhc_code(latest_storm_entry.name)
+    age = datetime.today() - latest_storm_entry["end_date"]
+    if pandas.isna(latest_storm_entry["end_date"]) or age < timedelta(days=1):
+        if datetime.today() - latest_storm_entry["start_date"] > timedelta(days=30):
+            assert latest_storm.status == StormStatus.HISTORICAL
+        else:
+            assert latest_storm.status == StormStatus.REALTIME
+    else:
+        assert latest_storm.status == StormStatus.HISTORICAL