Scheduling Captures#

capturegraph.scheduling decides when and where the next capture is worth taking. Instead of "every five minutes", it treats coverage as an importance sampling problem: forecast the conditions each candidate capture would happen under, score how far each lies from what the dataset already holds, and interrupt a participant only for the candidates that fill a real gap.

import capturegraph.scheduling as cgsh

The three stages#

The library is three stages, and they map onto the two kinds of recipe a server interceptor calls:

Forecasting and selection depend only on their inputs, so an interceptor wraps them in a cg.pure_function and a whole crowd's polls share one computation. Reservation is honestly stateful, so the books are cg.ImpureState classes: every call is a locked read-modify-write of the book, and two server workers never hand out the same slot.

forecast ──▶ select ──▶ reserve ──▶ @cgserver.intercept(...) ──▶ the app's next reminder
   pure        pure      impure

Scheduling code runs on the server, inside a target's interceptor; the same functions run unchanged in a notebook to simulate a campaign. From Zero to a Scheduled Target builds both a temporal and a spatial interceptor step by step.

Quick start#

A complete temporal schedule, from candidates to two participants holding different times. The book needs a scratch to persist in; a script configures one, and the server configures its own.

from datetime import UTC, datetime, timedelta

import capturegraph as cg
import capturegraph.scheduling as cgsh

cg.scratch.configure("~/.cache/capturegraph")

location = cg.Location(latitude=42.44, longitude=-76.5)
now = datetime.now(UTC)


def solar_session(when: datetime) -> dict[str, object]:
    return {"date": when, "solar_angle": cgsh.forecast.solar_position(location, when)}


# 1. Forecast: every hour of the coming day, with the solar angle it would see.
potential = cg.Array(
    [
        solar_session(when)
        for when in cgsh.forecast.times(
            resolution=timedelta(hours=1), span=timedelta(hours=24), start=now
        )
    ]
)
# The sessions already captured, in the same shape.
previous = cg.Array([solar_session(now - timedelta(hours=20)), solar_session(now - timedelta(hours=3))])

# 2. Select: the ten candidates whose solar angle the dataset lacks most.
selected = cgsh.select_sessions(
    potential,
    previous,
    cgsh.distance.combine(solar_angle=cgsh.distance.solar(sigma_deg=2.0)),
    selections=10,
)
times = [session["date"] for session in selected]

# 3. Reserve: one distinct, self-expiring time per participant.
book = cgsh.ReservationBook.open("my-target")
alice = book.reserve("alice", times, now=now)
bob = book.reserve("bob", times, now=now)
assert alice != bob

Each session is a plain mapping whose keys the distance reads by name: combine(solar_angle=...) scores the solar_angle attribute of both sides. See select_sessions for the full signature.

Section contents#