Resample beliefs about instantaneous sensors

timely_beliefs/beliefs/classes.py

@@ -73,7 +73,7 @@ class TimedBelief(object):
     - a cumulative probability (the likelihood of the value being equal or lower than stated)*
 
     * The default assumption is that the mean value is given (cp=0.5), but if no beliefs about possible other outcomes
-    are given, then this will be treated as a deterministic belief (cp=1). As an alternative to specifying an cumulative
+    are given, then this will be treated as a deterministic belief (cp=1). As an alternative to specifying a cumulative
     probability explicitly, you can specify an integer number of standard deviations which is translated
     into a cumulative probability assuming a normal distribution (e.g. sigma=-1 becomes cp=0.1587).
     """
@@ -667,6 +667,15 @@ def __repr__(self):
         """Add the sensor and event resolution to the string representation of the BeliefsSeries."""
         return super().__repr__() + "\n" + meta_repr(self)
 
+    @property
+    def event_frequency(self) -> Optional[timedelta]:
+        """Duration between observations of events.
+
+        :returns: a timedelta for regularly spaced observations
+                  None for irregularly spaced observations
+        """
+        return pd.Timedelta(pd.infer_freq(self.index.unique("event_start")))
+
 
 class BeliefsDataFrame(pd.DataFrame):
     """Beliefs about a sensor.
@@ -1023,6 +1032,15 @@ def set_event_value_from_source(
             )
         )
 
+    @property
+    def event_frequency(self) -> Optional[timedelta]:
+        """Duration between observations of events.
+
+        :returns: a timedelta for regularly spaced observations
+                  None for irregularly spaced observations
+        """
+        return pd.Timedelta(pd.infer_freq(self.index.unique("event_start")))
+
     @property
     def knowledge_times(self) -> pd.DatetimeIndex:
         return pd.DatetimeIndex(
@@ -1386,6 +1404,13 @@ def resample_events(
             return self
         df = self
 
+        # Resample instantaneous sensors
+        # The event resolution stays zero, but the event frequency is updated
+        if df.event_resolution == timedelta(0):
+            if df.lineage.number_of_events != len(df):
+                raise NotImplementedError("Please file a GitHub ticket.")
+            return belief_utils.resample_instantaneous_events(df, event_resolution)
+
         belief_timing_col = (
             "belief_time" if "belief_time" in df.index.names else "belief_horizon"
         )

timely_beliefs/beliefs/utils.py

@@ -1,9 +1,12 @@
+from __future__ import annotations
+
 import warnings
 from datetime import datetime, timedelta
 from typing import List, Optional, Union
 
 import numpy as np
 import pandas as pd
+import pytz
 from packaging import version
 from pandas.core.groupby import DataFrameGroupBy
 
@@ -283,7 +286,9 @@ def join_beliefs(
     if output_resolution > input_resolution:
 
         # Create new BeliefsDataFrame with downsampled event_start
-        if output_resolution % input_resolution != timedelta(0):
+        if input_resolution == timedelta(
+            0
+        ) or output_resolution % input_resolution != timedelta(0):
             raise NotImplementedError(
                 "Cannot downsample from resolution %s to %s."
                 % (input_resolution, output_resolution)
@@ -778,6 +783,111 @@ def extreme_timedeltas_not_equal(
     return td_a != td_b
 
 
+def resample_instantaneous_events(
+    df: pd.DataFrame | "classes.BeliefsDataFrame",
+    resolution: timedelta,
+    method: str | None = None,
+    dropna: bool = True,
+) -> pd.DataFrame | "classes.BeliefsDataFrame":
+    """Resample data representing instantaneous events.
+
+    Updates the event frequency of the resulting data frame, and possibly also its event resolution.
+    The event resolution is only updated if the resampling method computes a characteristic of a period of events,
+    like 'mean' or 'first'.
+
+    Note that, for resolutions over 1 hour, the data frequency may not turn out to be constant per se.
+    This is due to DST transitions:
+    - The duration between events is typically longer for the fall DST transition.
+    - The duration between events is typically shorter for the spring DST transition.
+    This is done to keep the data frequency in step with midnight in the sensor's timezone.
+    """
+
+    # Default resampling method for instantaneous sensors
+    if method is None:
+        method = "asfreq"
+
+    # Use event_start as the only index level
+    index_names = df.index.names
+    df = df.reset_index().set_index("event_start")
+
+    # Resample the data in each unique fixed timezone offset that belongs to the given IANA timezone, then recombine
+    unique_offsets = df.index.map(lambda x: x.utcoffset()).unique()
+    resampled_df_offsets = []
+    for offset in unique_offsets:
+        df_offset = df.copy()
+        # Convert all the data to given timezone offset
+        df_offset.index = df.index.tz_convert(
+            pytz.FixedOffset(offset.seconds // 60)
+        )  # offset is max 1439 minutes, so we don't need to check offset.days
+        # Resample all the data in the given timezone offset, using the given method
+        resampled_df_offset = getattr(df_offset.resample(resolution), method)()
+        # Convert back to the original timezone
+        if isinstance(df.index, pd.DatetimeIndex) and df.index.tz is not None:
+            resampled_df_timezone = resampled_df_offset.tz_convert(df.index.tz)
+        elif isinstance(df, classes.BeliefsDataFrame):
+            # As a backup, use the original timezone from the BeliefsDataFrame's sensor
+            resampled_df_timezone = resampled_df_offset.tz_convert(df.sensor.timezone)
+        else:
+            ValueError("Missing original timezone.")
+        # See which resampled rows still fall in the given offset, in this timezone
+        resampled_df_timezone = resampled_df_timezone[
+            resampled_df_timezone.index.map(lambda x: x.utcoffset()) == offset
+        ]
+        resampled_df_offsets.append(resampled_df_timezone)
+    resampled_df = pd.concat(resampled_df_offsets).sort_index()
+
+    # If possible, infer missing frequency
+    if resampled_df.index.freq is None and len(resampled_df) > 2:
+        resampled_df.index.freq = pd.infer_freq(resampled_df.index)
+
+    # Restore the original index levels
+    resampled_df = resampled_df.reset_index().set_index(index_names)
+
+    if method in (
+        "mean",
+        "max",
+        "min",
+        "median",
+        "count",
+        "nunique",
+        "first",
+        "last",
+        "ohlc",
+        "prod",
+        "size",
+        "sem",
+        "std",
+        "sum",
+        "var",
+        "quantile",
+    ):
+        # These methods derive properties of a period of events.
+        # Therefore, the event resolution is updated.
+        # The methods are typically used for downsampling.
+        resampled_df.event_resolution = resolution
+    elif method in (
+        "asfreq",
+        "interpolate",
+        "ffill",
+        "bfill",
+        "pad",
+        "backfill",
+        "nearest",
+    ):
+        # These methods derive intermediate events.
+        # Therefore, the event resolution is unaffected.
+        # The methods are typically used for upsampling.
+        pass
+    else:
+        raise NotImplementedError(
+            f"Please file a GitHub ticket for timely-beliefs to support the '{method}' method."
+        )
+
+    if dropna:
+        return resampled_df.dropna()
+    return resampled_df
+
+
 def meta_repr(
     tb_structure: Union["classes.BeliefsDataFrame", "classes.BeliefsSeries"]
 ) -> str:

timely_beliefs/sensors/classes.py

@@ -97,8 +97,10 @@ class SensorDBMixin(Sensor):
     Mixin class for a table with sensors.
     """
 
+    # db field that is not a Sensor attribute
     id = Column(Integer, primary_key=True)
-    # overwriting name as db field
+
+    # db fields that overwrite Sensor attributes
     name = Column(String(120), nullable=False, default="")
     unit = Column(String(80), nullable=False, default="")
     timezone = Column(String(80), nullable=False, default="UTC")