Split a sorted event stream into sessions on a 30-minute gap
timestamps holds one user's event times in Unix seconds, sorted ascending. A gap of more than 30 minutes (1800 seconds) between consecutive events starts a new session. Return a list of session lengths, where a session's length is its last timestamp minus its first (in seconds).
A lone event is a session of length 0. An empty input returns [].
def session_lengths(timestamps: list[int], gap: int = 1800) -> list[int]:
# your code hereWrite the implementation.
Walk the sorted timestamps once, tracking the current session's start and the previous event. When the step to the next event exceeds the gap, close the session — last minus first — and start a new one. Flush the final session after the loop; it is one O(n) pass.
- ✗Measuring the gap from the session start instead of the previous event
- ✗Bucketing by fixed clock windows rather than by inter-event gaps
- ✗Forgetting to flush the final session after the loop
- →How would you return event counts per session instead of durations?
- →What changes if the input is not guaranteed sorted?
Track the current session's start and the prev timestamp. A gap larger than gap closes the session; append prev - start. Flush the last session after the loop.
def session_lengths(timestamps, gap=1800):
if not timestamps:
return []
lengths = []
start = prev = timestamps[0]
for t in timestamps[1:]:
if t - prev > gap:
lengths.append(prev - start)
start = t
prev = t
lengths.append(prev - start)
return lengthsFor times [0, 600, 1200, 5000, 5600] the 5000 - 1200 = 3800-second jump splits into two sessions: [0, 600, 1200] of length 1200 and [5000, 5600] of length 600, so the result is [1200, 600].
