Skip to content

Assignment Store#

The AssignmentStore manages per-user notification scheduling to prevent conflicts when multiple users capture data for the same target.

Overview#

When scheduling notifications, you need to track:

  • Which user has been assigned to which time slot
  • Whether assignments have expired
  • Whether a new capture invalidates existing assignments

The AssignmentStore handles all of this automatically.

Basic Usage#

from datetime import timedelta
import capturegraph.scheduling as cgsh

store = cgsh.organize.AssignmentStore(
    persistence=persistence,
    expire_after=timedelta(minutes=30),
    last_capture=sessions[-1].date if sessions else None,
)

# Assign next available slot to a user
next_slot = store.assign_next(user_id, candidate_slots)

# Check existing assignment
existing = store[user_id]

Key Features#

Feature Description
Unique slots Each user gets a unique slot from the candidate pool
Automatic expiry Stale assignments are cleaned up after expire_after
Capture-aware cleanup When a new capture arrives, old assignments are discarded
Drift reconciliation Rounds stored assignments to nearest candidate slots

Complete Example#

from datetime import timedelta
import capturegraph.data as cg
import capturegraph.scheduling as cgsh

def __notification__(user_id, target, persistence):
    sessions = target.Bridge

    # 1. Generate candidate slots using Void & Cluster
    potential = cg.List()
    potential.date = cgsh.forecast.times(span=timedelta(hours=24))
    potential.location = target._metadata._capture_location
    potential.solar_angle = cgsh.forecast.solar_position(
        potential.location, potential.date
    )

    distance_fn = cgsh.distance.combine(
        solar_angle=cgsh.distance.solar(sigma_deg=2.0),
    )

    slots = cgsh.select_sessions(
        potential, sessions, distance_fn, selections=10
    )

    # 2. Initialize assignment store
    store = cgsh.organize.AssignmentStore(
        persistence=persistence,
        expire_after=timedelta(minutes=30),
        last_capture=sessions[-1].date if sessions else None,
    )

    # 3. Assign next available slot
    return store.assign_next(user_id, slots)

See Also#