pcalc.time_model

  1# -*- coding: utf-8 -*-
  2# Copyright (c) 2025 Krishna Kumar
  3# SPDX-License-Identifier: MIT
  4"""
  5## Introduction
  6
  7`TimeModel` is a calendar-aware utility that maps external `datetime` values
  8to continuous float time relative to a chosen origin and unit of measure.
  9
 10The presence calculus operates over real time axis, and time is therefore
 11represented as a floating point value in all calculations.
 12
 13To avoid working with arbitrarily large floating point numbers unnecessarily—which can
 14introduce interpretability or numerical precision issues—it is recommended
 15that you define an explicit origin for your time axis and express all time values
 16relative to this origin.
 17
 18This model supports both fixed-duration units (e.g., 'seconds', 'minutes',
 19'days') and calendar-relative units (e.g., 'months', 'quarters', 'years').
 20
 21`TimeModel` is most useful when interfacing with external timestamped data
 22sources—such as logs, CRM systems, or real-time feeds—where time is expressed
 23as absolute wall-clock timestamps.
 24
 25In contrast, when working with synthetic or simulated data (e.g., from a
 26simulation engine), such translation is typically unnecessary, as the
 27simulation clock already provides floating point time relative to the start
 28of the simulation.
 29
 30Parameters:
 31- origin: the reference `datetime` from which time is measured
 32- unit: one of 'seconds', 'minutes', 'hours', 'days', 'months', 'quarters', 'years'
 33
 34### Examples
 35
 36```python
 37from datetime import datetime
 38from pc.entity import Entity
 39from pc.presence import Presence
 40from pc.time import TimeModel
 41
 42# Define a time model using seconds since midnight
 43time_model = TimeModel(origin=datetime(2025, 5, 1), unit="seconds")
 44
 45# External datetime timestamps
 46start_time = datetime(2025, 5, 1, 9, 30)
 47end_time = datetime(2025, 5, 1, 10, 45)
 48
 49# Convert to float using the time model
 50start = time_model.to_float(start_time)  # 34200.0
 51end = time_model.to_float(end_time)      # 38700.0
 52
 53# Define domain entities
 54element = Entity(id="cust-001", name="Alice Chen", metadata={"type": "customer"})
 55boundary = Entity(id="seg-enterprise", name="Enterprise Segment")
 56
 57# Create the presence assertion
 58presence = Presence(
 59    element=element,
 60    boundary=boundary,
 61    start=start,
 62    end=end,
 63    provenance="crm"
 64)
 65
 66print(presence)
 67```
 68"""
 69from datetime import datetime, timedelta
 70from typing import Literal, Union
 71from dateutil.relativedelta import relativedelta
 72
 73CalendarUnit = Literal["seconds", "minutes", "hours", "days", "months", "quarters", "years"]
 74
 75class TimeModel:
 76
 77
 78    def __init__(self, origin: datetime, unit: CalendarUnit = "seconds"):
 79        self.origin = origin
 80        self.unit = unit
 81
 82    def to_float(self, t: datetime) -> float:
 83        delta = t - self.origin
 84        seconds = delta.total_seconds()
 85
 86        if self.unit == "seconds":
 87            return seconds
 88        if self.unit == "minutes":
 89            return seconds / 60.0
 90        if self.unit == "hours":
 91            return seconds / 3600.0
 92        if self.unit == "days":
 93            return delta.days + (delta.seconds / 86400.0)
 94        if self.unit == "months":
 95            return self._months_between(self.origin, t)
 96        if self.unit == "quarters":
 97            return self._months_between(self.origin, t) / 3.0
 98        if self.unit == "years":
 99            return self._months_between(self.origin, t) / 12.0
100
101        raise ValueError(f"Unsupported time unit: {self.unit}")
102
103    def from_float(self, t: float) -> datetime:
104        if self.unit == "seconds":
105            return self.origin + timedelta(seconds=t)
106        if self.unit == "minutes":
107            return self.origin + timedelta(minutes=t)
108        if self.unit == "hours":
109            return self.origin + timedelta(hours=t)
110        if self.unit == "days":
111            return self.origin + timedelta(days=t)
112        if self.unit in {"months", "quarters", "years"}:
113            months = {
114                "months": int(round(t)),
115                "quarters": int(round(t * 3)),
116                "years": int(round(t * 12)),
117            }[self.unit]
118            return self.origin + relativedelta(months=+months)
119
120        raise ValueError(f"Unsupported time unit: {self.unit}")
121
122    @staticmethod
123    def _months_between(d1: datetime, d2: datetime) -> float:
124        """Return the number of calendar months between two dates, fractional."""
125        rd = relativedelta(d2, d1)
126        return rd.years * 12 + rd.months + (rd.days / 30.0)