Calculate the number of people occupied at each time interval

Question

The following scripts calculate a list with the number of people occupied at each time interval. The intervals are defined by the timestamps of two consecutive events that come from the sorted list events_timeline. Also, start_ts and end_ts are taking into account for intervals calculation and for each event, one person is occupied.

input data to be used in the following scripts:

MINIMUM_EVENT_INTERVAL = 1
start_ts = 1598939760.0
end_ts = 1598939792.0
events_timeline = [(1598939761.0, 's'), (1598939771.0, 's'), (1598939781.0, 'e'),
                   (1598939791.0, 'e')]  # 's' and 'e' denote start and end event respectively

1st approach:

people_occupied = []
start_events = 0
end_events = 0

num_events_timeline = len(events_timeline)

if events_timeline[0][0] - start_ts >= MINIMUM_EVENT_INTERVAL:
    people_occupied.append(0)

for index, (ts, event) in enumerate(events_timeline):
    if event == 's':
        start_events += 1
    else:
        end_events += 1

    if index != num_events_timeline - 1 and ts == events_timeline[index + 1][0]:
        continue

    if index == num_events_timeline - 1 and end_ts - ts < MINIMUM_EVENT_INTERVAL:
        break

    people_occupied.append(start_events - end_events)

print(f'1st approach: {people_occupied}')

2nd approach:

currently_occupied = 0
people_occupied = []

if events_timeline[0][0] - start_ts >= MINIMUM_EVENT_INTERVAL:
    people_occupied.append(0)

for (start_time, event_type), (end_time, _) in zip(events_timeline[:-1], events_timeline[1:]):
    if event_type == 's':
        currently_occupied += 1
    else:
        currently_occupied -= 1

    if end_time > start_time:
        people_occupied.append(currently_occupied)

if end_ts - events_timeline[-1][0] >= MINIMUM_EVENT_INTERVAL:
    people_occupied.append(0)

print(f'2nd approach: {people_occupied}')

Both approaches do the job, but which one of them should be preferred and why?

Answer 1

Generally, at the edge of your program you shouldn't be entering datetimes as timestamp numeric literals, but as human-legible datetimes. On the inside of your program, you also shouldn't be processing datetimes as numeric timestamps, but instead as well-typed datetime.datetime instances.

The interior of your loop ignores MINIMUM_EVENT_INTERVAL and only pays attention to that at the bounds of your event sequence. This seems unlikely to be a good idea. Ideally, you would pay attention to that minimum for every event, and treat your bounds start_ts and end_ts the same way that you treat everything in your main timeline.

Generally I think your second approach using zip is preferable, but it isn't enough. You should write well-typed functions. Iterator functions are a convenient way to capture your logic.

Suggested

This (intentionally) isn't exactly equivalent in terms of logic, but since all of your timestamps already meet the interval minimum, the output is the same.

from datetime import datetime, timedelta
from typing import Iterable, Iterator, Literal, Sequence

MINIMUM_EVENT_INTERVAL = timedelta(seconds=1)


Event = tuple[
    datetime,
    # 's' and 'e' denote start and end event respectively
    Literal['s', 'e'],
]


def get_events_with_endpoints(
    start_ts: datetime,
    end_ts: datetime,
    events: Iterable[Event],
) -> Iterator[Event]:
    yield start_ts, 's'
    yield from events
    yield end_ts, 'e'


def get_occupied(events: Sequence[Event]) -> Iterator[int]:
    currently_occupied = 0

    for (start_time, _), (end_time, event_type) in zip(events[:-1], events[1:]):
        if end_time - start_time >= MINIMUM_EVENT_INTERVAL:
            yield currently_occupied
        if event_type == 's':
            currently_occupied += 1
        else:
            currently_occupied -= 1


def main() -> None:
    timeline = (
        (datetime(2020, 9, 1, 3, 26,  1), 's'),
        (datetime(2020, 9, 1, 3, 26, 11), 's'),
        (datetime(2020, 9, 1, 3, 26, 21), 'e'),
        (datetime(2020, 9, 1, 3, 26, 31), 'e'),
    )

    events = get_events_with_endpoints(
        start_ts=datetime(2020, 9, 1, 3, 26,  0),
        end_ts  =datetime(2020, 9, 1, 3, 26, 32),
        events=timeline,
    )

    print(tuple(get_occupied(tuple(events))))


if __name__ == '__main__':
    main()

Output

(0, 1, 2, 1, 0)