#!/usr/bin/env python3
"""One stream, one "5-minute" width, four window operators, four groupings -- DDIA 2e Ch. 12.

The book's two example events arrive at 10:03:39 and 10:08:12 -- 4 minutes 33
seconds apart, so strictly less than 5 minutes. Intuition says "a 5-minute
window" is a 5-minute window: feed the pair through any window called "5 minutes"
and they should land together (or not) the same way. They don't.

The same two events are grouped DIFFERENTLY depending on the window TYPE:

  - SLIDING (a 5-min window anchored at each event, looking back)   -> TOGETHER
  - TUMBLING (fixed non-overlapping 5-min buckets on a :00/:05 grid) -> SPLIT
  - HOPPING  (5-min windows advancing by a 1-min hop, overlapping)   -> disjoint sets
  - SESSION  (grouped by a 30-min inactivity gap, not by the clock)  -> activity-defined

"5 minutes" fixes the SIZE of a window but not its BOUNDARIES, its OVERLAP rule,
or whether boundaries exist at all. The grouping is a property of the operator,
not of the duration.

Pure standard library, one fixed event list, deterministic -> reproducible.
"""
from datetime import datetime, timedelta

# ---------------------------------------------------------------------------
# The stream. Event-time timestamps (when each event actually happened). The
# first two are the book's canonical pair. Events 3-5 add a burst-then-pause
# shape so the session window has something to chew on.
# ---------------------------------------------------------------------------
DAY = "2024-01-01 "
def t(s):
    return datetime.strptime(DAY + s, "%Y-%m-%d %H:%M:%S")

EVENTS = [
    ("e1", t("10:03:39")),   # book's first event
    ("e2", t("10:08:12")),   # book's second event -- 4m33s after e1
    ("e3", t("10:10:00")),   # still part of the opening burst
    ("e4", t("10:45:00")),   # after a 35-minute silence: a new burst
    ("e5", t("10:47:30")),   # 2m30s after e4
]

# The two events the whole exercise is about.
E1_NAME, E1_TIME = EVENTS[0]
E2_NAME, E2_TIME = EVENTS[1]

SIZE = timedelta(minutes=5)     # every operator below is a "5-minute" window...
HOP = timedelta(minutes=1)      # ...but hopping advances by 1 minute
GAP = timedelta(minutes=30)     # ...and session closes after 30 min of silence

GRID = t("10:00:00")            # tumbling / hopping boundaries snap to this grid


def hm(dt):
    return dt.strftime("%H:%M")

def hms(dt):
    return dt.strftime("%H:%M:%S")

def win(start):
    return "[{}, {})".format(hm(start), hm(start + SIZE))

def rule(title):
    print()
    print("=" * 74)
    print(title)
    print("=" * 74)


# ---------------------------------------------------------------------------
# TUMBLING: fixed non-overlapping buckets snapped to the :00/:05/:10 grid.
# Each event lands in exactly one bucket: the one it falls inside.
# ---------------------------------------------------------------------------
def tumbling_start(dt):
    """The single 5-min bucket [start, start+5) on the grid that contains dt."""
    elapsed = dt - GRID
    n = int(elapsed.total_seconds() // SIZE.total_seconds())
    return GRID + n * SIZE

def run_tumbling():
    rule('TUMBLING  --  fixed 5-min buckets on a :00 / :05 / :10 grid (no overlap)')
    counts = {}
    assign = {}
    for name, dt in EVENTS:
        s = tumbling_start(dt)
        assign[name] = s
        counts.setdefault(s, []).append(name)
    for name, dt in EVENTS:
        print("    {} @ {}  ->  bucket {}".format(name, hms(dt), win(assign[name])))
    print()
    for s in sorted(counts):
        members = counts[s]
        print("    bucket {}: {}  (count {})".format(win(s), ", ".join(members), len(members)))
    print()
    same = assign[E1_NAME] == assign[E2_NAME]
    print("    {} in {}   {} in {}".format(
        E1_NAME, win(assign[E1_NAME]), E2_NAME, win(assign[E2_NAME])))
    print("    -> {} and {} are {}  ({}).".format(
        E1_NAME, E2_NAME,
        "TOGETHER" if same else "SPLIT into DIFFERENT buckets",
        "count 2 in one bucket" if same else "count 1 + count 1"))


# ---------------------------------------------------------------------------
# HOPPING: fixed 5-min windows, but a new one starts every HOP (1 min). Windows
# overlap, so each event belongs to SIZE/HOP = 5 windows at once.
# ---------------------------------------------------------------------------
def hopping_starts(dt):
    """Every grid-aligned window [s, s+5) with s advancing by HOP that contains dt."""
    out = []
    # earliest possible start is dt - SIZE (exclusive); walk the hop grid forward
    k = int((dt - SIZE - GRID).total_seconds() // HOP.total_seconds())
    s = GRID + k * HOP
    while s <= dt:
        if s <= dt < s + SIZE:
            out.append(s)
        s += HOP
    return out

def run_hopping():
    rule('HOPPING   --  5-min windows, new one every 1 min (they OVERLAP)')
    s1 = hopping_starts(E1_TIME)
    s2 = hopping_starts(E2_TIME)
    print("    {} @ {}  ->  {} windows: {}".format(
        E1_NAME, hms(E1_TIME), len(s1), ", ".join(win(s) for s in s1)))
    print("    {} @ {}  ->  {} windows: {}".format(
        E2_NAME, hms(E2_TIME), len(s2), ", ".join(win(s) for s in s2)))
    print()
    shared = [s for s in s1 if s in s2]
    if shared:
        print("    shared window(s): {}  -> {} and {} TOGETHER".format(
            ", ".join(win(s) for s in shared), E1_NAME, E2_NAME))
    else:
        print("    shared windows: NONE")
        print("    -> {} and {} land in overlapping-but-DIFFERENT sets of windows;".format(
            E1_NAME, E2_NAME))
        print("       no single 5-min window on the hop grid straddles the pair.")


# ---------------------------------------------------------------------------
# SLIDING: one 5-min window per event, anchored at the event and looking back.
# Two events co-occur when one falls inside the other's lookback window.
# ---------------------------------------------------------------------------
def sliding_members(anchor_time):
    """Events inside the window (anchor - SIZE, anchor]."""
    lo = anchor_time - SIZE
    return [n for n, dt in EVENTS if lo < dt <= anchor_time]

def run_sliding():
    rule('SLIDING   --  a 5-min window anchored at each event, looking back')
    for name, dt in EVENTS:
        members = sliding_members(dt)
        lo = dt - SIZE
        print("    window for {} = ({}, {}]: {}  (count {})".format(
            name, hms(lo), hms(dt), ", ".join(members), len(members)))
    print()
    m2 = sliding_members(E2_TIME)
    together = E1_NAME in m2 and E2_NAME in m2
    print("    window anchored at {} ({}) looks back to {}".format(
        E2_NAME, hms(E2_TIME), hms(E2_TIME - SIZE)))
    print("    {} @ {} falls inside it -> {} and {} {}  (count {}).".format(
        E1_NAME, hms(E1_TIME), E1_NAME, E2_NAME,
        "TOGETHER" if together else "apart", len(m2)))


# ---------------------------------------------------------------------------
# SESSION: no clock grid at all. Walk events in time order; a gap larger than
# GAP (30 min) closes the current session and opens a new one.
# ---------------------------------------------------------------------------
def sessions():
    out = []
    cur = []
    prev = None
    for name, dt in sorted(EVENTS, key=lambda e: e[1]):
        if prev is not None and (dt - prev) > GAP:
            out.append(cur)
            cur = []
        cur.append((name, dt))
        prev = dt
    if cur:
        out.append(cur)
    return out

def run_session():
    rule('SESSION   --  grouped by a 30-min inactivity gap (clock ignored)')
    ses = sessions()
    prev = None
    for name, dt in sorted(EVENTS, key=lambda e: e[1]):
        gap = "" if prev is None else "  (gap {:>5}s = {})".format(
            int((dt - prev).total_seconds()),
            "> 30m: NEW session" if (dt - prev) > GAP else "<= 30m: same session")
        print("    {} @ {}{}".format(name, hms(dt), gap))
        prev = dt
    print()
    for i, s in enumerate(ses, 1):
        names = [n for n, _ in s]
        span = "{}-{}".format(hms(s[0][1]), hms(s[-1][1]))
        print("    session {} [{}]: {}  (count {})".format(i, span, ", ".join(names), len(names)))
    print()
    # which session holds the two canonical events?
    for i, s in enumerate(ses, 1):
        names = [n for n, _ in s]
        if E1_NAME in names and E2_NAME in names:
            print("    -> {} and {} are TOGETHER in session {} -- grouped by activity,".format(
                E1_NAME, E2_NAME, i))
            print("       not by any 5-minute clock boundary.")
            break


def main():
    print("One stream. One 5-minute width. Four window operators.")
    print("The book's two events:")
    print("    {} @ {}".format(E1_NAME, hms(E1_TIME)))
    print("    {} @ {}   ({} - {} = {}m{}s apart, so < 5 minutes)".format(
        E2_NAME, hms(E2_TIME), hms(E2_TIME), hms(E1_TIME),
        int((E2_TIME - E1_TIME).total_seconds()) // 60,
        int((E2_TIME - E1_TIME).total_seconds()) % 60))
    print("Full stream (adds a burst-then-pause tail for the session window):")
    for name, dt in EVENTS:
        print("    {} @ {}".format(name, hms(dt)))

    run_sliding()
    run_tumbling()
    run_hopping()
    run_session()

    rule('SUMMARY  --  same "5-minute" width, {} and {} grouped four ways'.format(
        E1_NAME, E2_NAME))
    print("    SLIDING : TOGETHER (one window holds both, count 2)")
    print("    TUMBLING: SPLIT    (different grid buckets, count 1 + 1)")
    print("    HOPPING : disjoint (overlapping windows, but no shared one)")
    print("    SESSION : TOGETHER (same activity burst, gap < 30m)")
    print()
    print('    A "5-minute window" is not a 5-minute window. The width is the same;')
    print("    the grouping is the operator's, not the duration's.")


if __name__ == "__main__":
    main()