Merge pull request #2246 from FCP-INDI/monitoring-datetime-edge-cases

sgiavasis · web-flow · commit 7c1275ca4046 · 2025-07-25T14:04:52.000-04:00
🥅 Catch more datetime edge cases in monitoring
diff --git a/CPAC/utils/monitoring/draw_gantt_chart.py b/CPAC/utils/monitoring/draw_gantt_chart.py
@@ -39,20 +39,19 @@
 
 # You should have received a copy of the GNU Lesser General Public
 # License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
-"""Module to draw an html gantt chart from logfile produced by
-``CPAC.utils.monitoring.log_nodes_cb()``.
+"""Module to draw an html gantt chart from logfile produced by `~CPAC.utils.monitoring.log_nodes_cb`.
 
 See https://nipype.readthedocs.io/en/latest/api/generated/nipype.utils.draw_gantt_chart.html
 """
 
 from collections import OrderedDict
-from datetime import datetime
+from datetime import datetime, timedelta
 import random
 from warnings import warn
 
 from nipype.utils.draw_gantt_chart import draw_lines, draw_resource_bar, log_to_dict
 
-from CPAC.utils.monitoring.monitoring import DatetimeWithSafeNone
+from CPAC.utils.monitoring.monitoring import _NoTime, DatetimeWithSafeNone
 
 
 def create_event_dict(start_time, nodes_list):
@@ -404,37 +403,38 @@ def generate_gantt_chart(
 
     for node in nodes_list:
         if "duration" not in node and (node["start"] and node["finish"]):
-            node["duration"] = (node["finish"] - node["start"]).total_seconds()
+            _duration = node["finish"] - node["start"]
+            assert isinstance(_duration, timedelta)
+            node["duration"] = _duration.total_seconds()
 
     # Create the header of the report with useful information
     start_node = nodes_list[0]
     last_node = nodes_list[-1]
+    start = DatetimeWithSafeNone(start_node["start"])
+    finish = DatetimeWithSafeNone(last_node["finish"])
+    if isinstance(start, _NoTime) or isinstance(finish, _NoTime):
+        return
+    start, finish = DatetimeWithSafeNone.sync_tz(start, finish)
     try:
-        duration = (last_node["finish"] - start_node["start"]).total_seconds()
+        duration = (finish - start).total_seconds()
     except TypeError:
         # no duration
         return
 
     # Get events based dictionary of node run stats
-    events = create_event_dict(start_node["start"], nodes_list)
+    events = create_event_dict(start, nodes_list)
 
     # Summary strings of workflow at top
-    html_string += (
-        "<p>Start: " + start_node["start"].strftime("%Y-%m-%d %H:%M:%S") + "</p>"
-    )
-    html_string += (
-        "<p>Finish: " + last_node["finish"].strftime("%Y-%m-%d %H:%M:%S") + "</p>"
-    )
+    html_string += "<p>Start: " + start.strftime("%Y-%m-%d %H:%M:%S") + "</p>"
+    html_string += "<p>Finish: " + finish.strftime("%Y-%m-%d %H:%M:%S") + "</p>"
     html_string += "<p>Duration: " + f"{duration / 60:.2f}" + " minutes</p>"
     html_string += "<p>Nodes: " + str(len(nodes_list)) + "</p>"
     html_string += "<p>Cores: " + str(cores) + "</p>"
     html_string += close_header
     # Draw nipype nodes Gantt chart and runtimes
-    html_string += draw_lines(
-        start_node["start"], duration, minute_scale, space_between_minutes
-    )
+    html_string += draw_lines(start, duration, minute_scale, space_between_minutes)
     html_string += draw_nodes(
-        start_node["start"],
+        start,
         nodes_list,
         cores,
         minute_scale,
@@ -448,8 +448,8 @@ def generate_gantt_chart(
     # Plot gantt chart
     resource_offset = 120 + 30 * cores
     html_string += draw_resource_bar(
-        start_node["start"],
-        last_node["finish"],
+        start,
+        finish,
         estimated_mem_ts,
         space_between_minutes,
         minute_scale,
@@ -458,8 +458,8 @@ def generate_gantt_chart(
         "Memory",
     )
     html_string += draw_resource_bar(
-        start_node["start"],
-        last_node["finish"],
+        start,
+        finish,
         runtime_mem_ts,
         space_between_minutes,
         minute_scale,
@@ -473,8 +473,8 @@ def generate_gantt_chart(
     runtime_threads_ts = calculate_resource_timeseries(events, "runtime_threads")
     # Plot gantt chart
     html_string += draw_resource_bar(
-        start_node["start"],
-        last_node["finish"],
+        start,
+        finish,
         estimated_threads_ts,
         space_between_minutes,
         minute_scale,
@@ -483,8 +483,8 @@ def generate_gantt_chart(
         "Threads",
     )
     html_string += draw_resource_bar(
-        start_node["start"],
-        last_node["finish"],
+        start,
+        finish,
         runtime_threads_ts,
         space_between_minutes,
         minute_scale,
diff --git a/CPAC/utils/monitoring/monitoring.py b/CPAC/utils/monitoring/monitoring.py
@@ -16,14 +16,16 @@
 # License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
 """Monitoring utilities for C-PAC."""
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import glob
 import json
 import math
 import os
 import socketserver
+import struct
 import threading
-from typing import Any, Optional, TypeAlias
+from typing import Any, Optional, overload, TypeAlias
+from zoneinfo import available_timezones, ZoneInfo
 
 import networkx as nx
 from traits.trait_base import Undefined
@@ -72,16 +74,104 @@ def __sub__(self, other: "DatetimeWithSafeNone | _NoTime") -> datetime | timedel
         """Subtract between None and a datetime or timedelta or None."""
         return _safe_none_diff(self, other)
 
+    def isoformat(self) -> str:
+        """Return an ISO 8601-like string of 0s for display."""
+        return "0000-00-00"
+
 
 NoTime = _NoTime()
 """A singleton None that can be used in place of a datetime object."""
 
 
 class DatetimeWithSafeNone(datetime, _NoTime):
-    """Time class that can be None or a time value."""
-
-    def __new__(cls, dt: "OptionalDatetime") -> "DatetimeWithSafeNone | _NoTime":
+    """Time class that can be None or a time value.
+
+    Examples
+    --------
+    >>> from datetime import datetime
+    >>> DatetimeWithSafeNone(datetime(2025, 6, 18, 21, 6, 43, 730004)).isoformat()
+    '2025-06-18T21:06:43.730004'
+    >>> DatetimeWithSafeNone("2025-06-18T21:06:43.730004").isoformat()
+    '2025-06-18T21:06:43.730004'
+    >>> DatetimeWithSafeNone(b"\\x07\\xe9\\x06\\x12\\x10\\x18\\x1c\\x88\\x6d\\x01").isoformat()
+    '2025-06-18T16:24:28.028040+00:00'
+    >>> DatetimeWithSafeNone(b'\\x07\\xe9\\x06\\x12\\x10\\x18\\x1c\\x88m\\x00').isoformat()
+    '2025-06-18T16:24:28.028040'
+    >>> DatetimeWithSafeNone(DatetimeWithSafeNone("2025-06-18")).isoformat()
+    '2025-06-18T00:00:00'
+    >>> DatetimeWithSafeNone(None)
+    NoTime
+    >>> DatetimeWithSafeNone(None).isoformat()
+    '0000-00-00'
+    """
+
+    @overload
+    def __new__(
+        cls,
+        year: "OptionalDatetime",
+        month: None = None,
+        day: None = None,
+        hour: None = None,
+        minute: None = None,
+        second: None = None,
+        microsecond: None = None,
+        tzinfo: None = None,
+        *,
+        fold: None = None,
+    ) -> "DatetimeWithSafeNone | _NoTime": ...
+    @overload
+    def __new__(
+        cls,
+        year: int,
+        month: Optional[int] = None,
+        day: Optional[int] = None,
+        hour: int = 0,
+        minute: int = 0,
+        second: int = 0,
+        microsecond: int = 0,
+        tzinfo: Optional[timezone | ZoneInfo] = None,
+        *,
+        fold: int = 0,
+    ) -> "DatetimeWithSafeNone": ...
+
+    def __new__(
+        cls,
+        year: "int | OptionalDatetime",
+        month: Optional[int] = None,
+        day: Optional[int] = None,
+        hour: Optional[int] = 0,
+        minute: Optional[int] = 0,
+        second: Optional[int] = 0,
+        microsecond: Optional[int] = 0,
+        tzinfo: Optional[timezone | ZoneInfo] = None,
+        *,
+        fold: Optional[int] = 0,
+    ) -> "DatetimeWithSafeNone | _NoTime":
         """Create a new instance of the class."""
+        if (
+            isinstance(year, int)
+            and isinstance(month, int)
+            and isinstance(day, int)
+            and isinstance(hour, int)
+            and isinstance(minute, int)
+            and isinstance(second, int)
+            and isinstance(microsecond, int)
+            and isinstance(fold, int)
+        ):
+            return datetime.__new__(
+                cls,
+                year,
+                month,
+                day,
+                hour,
+                minute,
+                second,
+                microsecond,
+                tzinfo,
+                fold=fold,
+            )
+        else:
+            dt = year
         if dt is None:
             return NoTime
         if isinstance(dt, datetime):
@@ -98,9 +188,43 @@ def __new__(cls, dt: "OptionalDatetime") -> "DatetimeWithSafeNone | _NoTime":
             )
         if isinstance(dt, bytes):
             try:
-                dt = dt.decode("utf-8")
+                tzflag: Optional[int]
+                year, month, day, hour, minute, second = struct.unpack(">H5B", dt[:7])
+                microsecond, tzflag = struct.unpack("<HB", dt[7:])
+                match tzflag:
+                    case 1:
+                        tzinfo = timezone.utc
+                    case 2:  # pragma: no cover
+                        try:
+                            tzinfo = ZoneInfo(
+                                next(
+                                    zone
+                                    for zone in available_timezones()
+                                    if "localtime" in zone
+                                )
+                            )
+                        except StopIteration:
+                            tzinfo = None
+                    case 0 | _:
+                        tzinfo = None
+                if (
+                    isinstance(year, int)
+                    and isinstance(month, int)
+                    and isinstance(day, int)
+                    and isinstance(hour, int)
+                    and isinstance(minute, int)
+                    and isinstance(second, int)
+                    and isinstance(microsecond, int)
+                ):
+                    return datetime.__new__(
+                        cls, year, month, day, hour, minute, second, microsecond, tzinfo
+                    )
+                else:
+                    msg = f"Unexpected type: {[type(part) for part in [year, month, day, hour, minute, second, microsecond]]}"
+                    raise TypeError(msg)
             except UnicodeDecodeError:
-                error = f"Cannot decode bytes to string: {dt}"
+                error = f"Cannot decode bytes to string: {dt!r}"
+                raise TypeError(error)
         if isinstance(dt, str):
             try:
                 return DatetimeWithSafeNone(datetime.fromisoformat(dt))
@@ -114,7 +238,7 @@ def __bool__(self) -> bool:
         """Return True if not NoTime."""
         return self is not NoTime
 
-    def __sub__(self, other: "DatetimeWithSafeNone | _NoTime") -> datetime | timedelta:
+    def __sub__(self, other: "DatetimeWithSafeNone | _NoTime") -> datetime | timedelta:  # type: ignore[reportIncompatibleMethodOverride]
         """Subtract between a datetime or timedelta or None."""
         return _safe_none_diff(self, other)
 
@@ -128,6 +252,17 @@ def __str__(self) -> str:
         """Return the string representation of the datetime or NoTime."""
         return super().__str__()
 
+    @staticmethod
+    def sync_tz(
+        one: "DatetimeWithSafeNone", two: "DatetimeWithSafeNone"
+    ) -> tuple[datetime, datetime]:
+        """Add timezone to other if one datetime is aware and other isn't ."""
+        if one.tzinfo is None and two.tzinfo is not None:
+            return one.replace(tzinfo=two.tzinfo), two
+        if one.tzinfo is not None and two.tzinfo is None:
+            return one, two.replace(tzinfo=one.tzinfo)
+        return one, two
+
 
 class DatetimeJSONEncoder(json.JSONEncoder):
     """JSON encoder that handles DatetimeWithSafeNone instances."""
@@ -146,7 +281,9 @@ def json_dumps(obj: Any, **kwargs) -> str:
     return json.dumps(obj, cls=DatetimeJSONEncoder, **kwargs)
 
 
-OptionalDatetime: TypeAlias = Optional[datetime | str | DatetimeWithSafeNone | _NoTime]
+OptionalDatetime: TypeAlias = Optional[
+    datetime | str | bytes | DatetimeWithSafeNone | _NoTime
+]
 """Type alias for a datetime, ISO-format string or None."""
 
 
diff --git a/CPAC/utils/tests/test_utils.py b/CPAC/utils/tests/test_utils.py
@@ -204,16 +204,39 @@ def check_expected_keys(
 
 
 @pytest.mark.parametrize(
-    "t1", [datetime.now(), datetime.isoformat(datetime.now()), None]
+    "t1",
+    [
+        datetime.now(),
+        datetime.now().astimezone(),
+        datetime.isoformat(datetime.now()),
+        None,
+    ],
 )
 @pytest.mark.parametrize(
-    "t2", [datetime.now(), datetime.isoformat(datetime.now()), None]
+    "t2",
+    [
+        datetime.now(),
+        datetime.now().astimezone(),
+        datetime.isoformat(datetime.now()),
+        None,
+    ],
 )
 def test_datetime_with_safe_none(t1: OptionalDatetime, t2: OptionalDatetime):
     """Test DatetimeWithSafeNone class works with datetime and None."""
+    originals = t1, t2
     t1 = DatetimeWithSafeNone(t1)
     t2 = DatetimeWithSafeNone(t2)
     if t1 and t2:
+        _tzinfos = [getattr(_, "tzinfo", None) for _ in originals]
+        if (
+            all(isinstance(_, datetime) for _ in originals)
+            and any(_tzinfos)
+            and not all(_tzinfos)
+        ):
+            with pytest.raises(TypeError):
+                originals[1] - originals[0]  # type: ignore[reportOperatorIssue]
+            _t1, _t2 = DatetimeWithSafeNone.sync_tz(*originals)  # type: ignore[reportArgumentType]
+            assert isinstance(_t2 - _t1, timedelta)
         assert isinstance(t2 - t1, timedelta)
     else:
         assert t2 - t1 == timedelta(0)
