Skip to content

From Zero to a Scheduled Target#

This page builds a hosted target from an empty directory to one the server actively coordinates: it renders a thumbnail for every session, tells each participant when to capture next, and, in a second target, where, then shows the crowd what it has captured on pages the app opens. The finished modules ship in the repository under capturegraph-server/examples/, and tests drive them through the client API and the page routes exactly as a phone would, so the module listings here are live code.

If you only want the reference for the decorators and the rules the server enforces, read Targets and Interceptors instead.

The mental model#

Three ideas carry the whole page:

  1. A target is a schema plus a procedure. The schema fixes every position a capture can occupy; the phone fills positions by running the procedure.
  2. An interceptor fills a position from the server instead. It is a plain function the server calls on every download of that position, per user, so a position can answer differently for each participant and change without an app update.
  3. Anything an interceptor must remember is a recipe. A pure_function memoizes deterministic work by the content of its inputs; an ImpureState holds the coordination that evolves, one keyed book with methods that read, modify, and commit it under a lock. There is no other persistence: no database, no per-target state file.
phone ──GET position──▶ server ──▶ interceptor(profile, target, node)
                                        │
                      ┌─────────────────┴──────────────────┐
              @cg.pure_function                      cg.ImpureState
           memoized by input content            keyed, one writer at a time
                      └─────────────────┬──────────────────┘
                                 <root>/.cg-scratch

Step 1: A root and a server#

The server runs over a root directory; every subdirectory holding a target module becomes a hosted target. The root can be any directory, inside the workspace or not; run the command from capturegraph-server so uv run finds it, and name the root and the port as flags (they default to the current directory and 4433):

mkdir ~/my-deployment
cd capturegraph-server && uv run capturegraph-server serve --root ~/my-deployment --port 4433

On first launch the server writes a commented config.toml and opens the recipe scratch at <root>/.cg-scratch (or wherever scratch_path in the [server] table points). Every recipe an interceptor calls reads and writes there, so the scratch is the one place server-side memory lives. It is a cache over durable data plus in-flight coordination, so deleting it costs a recompute and forgets who currently holds which slot, nothing more.

To skip ahead, serve the shipped examples directly:

cd capturegraph-server && uv run capturegraph-server serve --root examples

Every URL below assumes port 4433.

Step 2: The target module#

A target directory holds one Python file that marks a procedure with @cgserver.target. The directory's name is the target's name; the module file's name inside it is free. The management panel names the file after the directory (PlantJournal/PlantJournal.py) and the examples use plant_journal.py. Here is the whole target before any server-side logic:

import capturegraph as cg
from server import authoring as cgserver


class Session(cg.Struct):
    photo: cg.Image
    notes: cg.String
    _metadata: cg.Metadata


class PlantJournal(cg.Struct):
    sessions: cg.Map[cg.Date, Session]
    _saw_instructions: cg.Bool
    _metadata: cg.Metadata


@cgserver.target
@cg.procedure(PlantJournal, label="Plant Journal")
def plant_journal(root: cg.Procedure[cg.Path[PlantJournal]]) -> None:
    root._saw_instructions |= cg.ShowInstructions(
        text="Photograph the plant from the marked spot.",
    )
    root._metadata._location |= cg.CaptureLocation()
    session = root.sessions[cg.CaptureTime()]
    session.photo &= cg.CaptureImage(label="Photograph the plant")
    session.notes &= cg.UserInputString(label="Notes")

Sessions are keyed by cg.Date, a readable alias of cg.Time: the same class, so a session's key is its capture instant, stored as a 16-hex-digit session id, and the server's error messages call the key type Time.

Two details matter for what follows. The target-level _metadata._location is captured once (|= skips a position that already exists), so the scheduler later knows where the plant is. And both the target and each session declare a cg.Metadata folder, whose three hidden leaves are the ones the app reads on its own: _thumbnail for its gallery, _location to check a user is in the right place, and _next_notification to schedule a reminder.

Drop the directory into the root and it is hosted within target_refresh seconds. The server re-executes the module whenever the file changes, so saving an edit deploys it.

Step 3: Interceptors#

An interceptor claims a schema position with @cgserver.intercept(Schema, selector). The selector navigates the schema from its root (lambda s: s._metadata._next_notification) and runs once at import, so a field the schema lacks fails the load, never a request. Crossing a Map or Array is explicit and means every element: lambda s: s.sessions.for_each()._metadata._thumbnail claims the thumbnail of each session, including the ones whose keys nobody can name ahead of time, while lambda s: s.sessions names the map itself and indexing one is a load error. The position's type is fixed by the schema: write the return annotation to match it, and know that the value is what the server enforces, coerced to that type on every request.

A function declares any subset of three parameters, by name, and the server passes exactly those:

Parameter What it is
profile The identified caller: profile.user_id (a str) and profile.location (a cg.Location, or None unless they opted in to sharing it). An anonymous request never reaches a function that declares this.
target The target's captured data, loaded against the schema and refreshed after every upload.
node The Struct nearest enclosing the requested position (the session, for a session thumbnail); the whole target when no crossed container holds structs.

Step 4: Recipes, or what an interceptor may remember#

An interceptor runs on every request from every participant, across the server's worker threads and, if you configure several, its worker processes. It holds nothing itself. Instead it calls two kinds of recipe, and the scratch holds the result:

Recipe Contract Reach for it when
@cg.pure_function("name/1") Once a result is sealed, every later call with the same input content is served from the cache, across restarts. Callers arriving together on a miss run the body once: the first claims it, the rest wait for its seal and then hit. The answer depends only on its arguments: rendering, forecasting, a selection over the captured data.
cg.ImpureState subclass Book.open(key).method(...) loads the committed state, runs the method on it, and commits it, all under an exclusive lock on the key. Never cached; every call sees the latest state. The answer depends on what other participants were told: reservations, leases, counters.

Both are ordinary synchronous calls. A pure function's identity is the string you give it: edit the body without bumping "/1" to "/2" and the cache keeps serving the old results, deliberately. An ImpureState's identity is its module-qualified class name plus its field schema, so editing a method keeps the book while renaming, moving, or reshaping the class starts a fresh one. The open key shards books, so use one key per target (open("plant-journal")) and change it if you copy the module.

Step 5: A thumbnail interceptor#

The app's gallery reads sessions/<id>/.metadata/.thumbnail. A procedure can write one from the phone with cg.ConvertImageToThumbnail, but a session synced from elsewhere, or captured by an older procedure, has none. The server fills the gap:

@cg.pure_function("session_thumbnail/1")
def session_thumbnail(photo: cg.Image) -> cg.Thumbnail:
    thumbnail = cg.Thumbnail.new()
    photo.pil(max_axis=256).save(thumbnail)
    return thumbnail


@cgserver.intercept(
    PlantJournal,
    lambda s: s.sessions.for_each()._metadata._thumbnail,
    skip_if_exists=True,
)
def thumbnail(node: Session) -> cg.Thumbnail:
    if cg.is_missing(node.photo):
        raise LookupError("no photo to render yet")
    return session_thumbnail(node.photo)

Reading it bottom up:

  • skip_if_exists=True makes the interceptor disk-first: a thumbnail the phone wrote is served as is, and the phone may still upload one (the write refusal that shadows a claimed position is lifted). Without it the server's rendition always wins and uploads to that position get 409.
  • The selector's for_each() crosses sessions, so the one function answers for every session's .metadata/.thumbnail, the pattern sessions/*/.metadata/.thumbnail.
  • node is the session the requested thumbnail belongs to, so the function never parses a path. A session whose photo has not landed yet raises, which the server logs and answers with whatever is on disk, here a 404 the app treats as "no thumbnail yet".
  • The render is a pure function, keyed by the photo's content hash. Each photo is decoded once, however many participants open the gallery, and the result survives restarts. cg.Thumbnail.new() hands the body a fresh output file inside the call's workspace; the framework hashes and files it on return. To change the rendering, bump the identity to "session_thumbnail/2".

Step 6: Temporal scheduling#

_next_notification is the time the app will remind this participant to capture. The goal is distribution-aware: offer times whose solar conditions the dataset is still missing, and give every participant a different one.

from datetime import UTC, datetime, timedelta

import capturegraph.scheduling as cgsh
from server.notifications import hourly_bucket, notification_candidates


@cgserver.intercept(PlantJournal, lambda s: s._metadata._next_notification)
def notification(profile: cgserver.RequestProfile, target: PlantJournal) -> cg.Time:
    now = datetime.now(UTC)
    location = target._metadata._location
    if cg.is_missing(location):
        return cg.Time(now + timedelta(hours=1))

    candidates = notification_candidates(target.sessions, location, hourly_bucket(now))
    chosen = cgsh.ReservationBook.open("plant-journal").reserve(
        profile.user_id,
        [candidate.value for candidate in candidates],
        captured=max(target.sessions.to_dict(), default=None),
        grace=timedelta(minutes=30),
        now=now,
    )
    return cg.Time(chosen if chosen is not None else now + timedelta(hours=1))

The two recipes divide the work along the pure/impure line:

  • [notification_candidates][server.notifications.candidates.notification_candidates] is a pure function shipped with the server. It forecasts the solar angle of every hour in the coming day, runs Void & Cluster against the sessions already captured, and returns the ten hours that best fill the gaps. Its key is the content of target.sessions, the location, and the current hour bucket, so a crowd's polls within one hour share one selection, and a new capture or a new hour computes it again.
  • ReservationBook is the impure state. reserve hands this user a candidate nobody else holds and hands the same one back on every poll. A hold is released once its time is more than grace in the past, and every hold at or before the latest capture plus grace (captured=) is released too, because that moment's conditions were just sampled. The whole read-modify-write runs under the key's lock, so two workers never promise one slot twice.

Until the plant's location has been captured there is nothing to forecast, so the interceptor answers "an hour from now" and lets the first capture happen. now is passed explicitly so the same code is deterministic under test.

Step 7: Spatial scheduling#

A survey wants coverage over an area rather than over time. The second example, ParkSurvey, intercepts the target's _location instead, the position the app uses to guide a participant to the right spot:

BOUNDS = cg.Array[cg.Location](
    [
        cg.Location(latitude=42.4440, longitude=-76.5020),
        cg.Location(latitude=42.4440, longitude=-76.5000),
        cg.Location(latitude=42.4425, longitude=-76.5000),
        cg.Location(latitude=42.4425, longitude=-76.5020),
    ]
)
DISTANCE = cgsh.distance.location(sigma_m=10.0)


@cg.pure_function("survey_batch/1")
def survey_batch(covered: cg.Array[cg.Location]) -> cg.Array[cg.Location]:
    return cgsh.select_sessions(
        potential_sessions=cgsh.forecast.locations_area(BOUNDS, resolution_meters=10.0),
        previous_sessions=covered,
        distance_fn=DISTANCE,
        energy_fn=lambda d: -d,
        energy_mode="max",
        selections=8,
    )


@cgserver.intercept(ParkSurvey, lambda s: s._metadata._location)
def next_spot(profile: cgserver.RequestProfile, target: ParkSurvey) -> cg.Location:
    covered = cg.Array[cg.Location](
        [
            session._metadata._location
            for session in target.sessions.values()
            if not cg.is_missing(session._metadata._location)
        ]
    )
    chosen = cgsh.CoverageBook.open("park-survey").reserve(
        profile.user_id,
        at=profile.location,
        fulfilled_by=list(covered),
        distance=DISTANCE,
        threshold=1.0,
        refill=list(survey_batch(covered)),
    )
    if chosen is None:
        raise LookupError("every survey point is covered")
    return chosen

Same division of labour, different quantities:

  • survey_batch is pure: a grid over the park is the candidate pool, the captured locations are what is already covered, and Void & Cluster picks eight well-spread uncovered points. Its key is the content of covered, so the batch is recomputed exactly when a new capture lands.
  • CoverageBook is the impure state. reserve leases this user the batch point closest to where they stand (at=profile.location, or the oldest point when they share no location), keeps handing it back while the lease holds (30 minutes by default), retires any leased point that a capture in fulfilled_by has come within threshold of, and installs refill= as the next batch once the current one is exhausted. threshold is in the units DISTANCE returns, so 1.0 here means ten metres.
  • When nothing is left to hand out the function raises, which leaves the position empty for that participant.

Step 8: Pages#

Participants can see the thumbnails in the app, one target at a time, but nobody can see the study: where the captures are, how many there are, what is still missing. A page is that view: a function in the target module that takes the value at a schema position and returns a website as a folder, with all the data inside. The server computes it on the first visit, caches it by the value's content hash, hosts the folder, and the app's Visualizations button opens it. PlantJournal declares two: one at the root, for the whole journal, and one at every session. The root page counts with Counter from the standard library.

@cgserver.page(PlantJournal, version=1)
@cg.ui.PageBuilder()
def journal(target: PlantJournal) -> None:
    sessions = target.sessions.to_dict()
    photographed = sum(1 for session in sessions.values() if not cg.is_missing(session.photo))
    days = {instant.date() for instant in sessions}
    cg.ui.Title("Plant Journal")
    cg.ui.Markdown(f"## Sessions\n\n{photographed} photographs of the plant so far.")
    with cg.ui.HStack():
        cg.ui.Stat("Sessions", len(sessions))
        cg.ui.Stat("Photographs", photographed)
        cg.ui.Stat("Days with captures", len(days))
    cg.ui.Chart(spec=sessions_per_day(list(sessions)))
    cg.ui.MapViewer.of(target.sessions, lambda session: session._metadata._location)
    cg.ui.Gallery.of(target.sessions, lambda s: session_thumbnail(s.photo))
    for instant, session in sessions.items():
        cg.ui.PageLink(session, title=instant.date().isoformat())


def sessions_per_day(captured: list[datetime]) -> dict[str, cg.JSONValue]:
    """A Vega-Lite bar chart of how many sessions were captured per day.

    Args:
        captured: Every session's capture instant.

    Returns:
        The specification with its rows inline, one per day.
    """
    counts = Counter(instant.date().isoformat() for instant in captured)
    rows: list[cg.JSONValue] = [
        {"day": day, "sessions": count} for day, count in sorted(counts.items())
    ]
    return {
        "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
        "width": "container",
        "data": {"values": rows},
        "mark": "bar",
        "encoding": {
            "x": {"field": "day", "type": "temporal", "timeUnit": "yearmonthdate", "title": "Day"},
            "y": {"field": "sessions", "type": "quantitative", "title": "Sessions"},
        },
    }


@cgserver.page(PlantJournal, lambda s: s.sessions.for_each(), version=1)
@cg.ui.PageBuilder()
def session(session: Session) -> None:
    cg.ui.Title("Session")
    cg.ui.ImageViewer(session.photo)
    size = session.photo.pil().size
    with cg.ui.HStack():
        cg.ui.Stat("Width", size[0], unit="px")
        cg.ui.Stat("Height", size[1], unit="px")
    cg.ui.Markdown(session.notes.value)

Reading it the same way:

  • @cgserver.page(PlantJournal, version=1) declares the page at the root, /pages/PlantJournal/, and makes the function a memoized recipe under "journal/1". It takes the whole journal as its one parameter, loaded against the schema, so its cache key is the content of the journal: an unchanged target serves the sealed folder at once, and the first visit after an upload computes a new one.
  • @cgserver.page(PlantJournal, lambda s: s.sessions.for_each(), version=1) stops where the thumbnail interceptor's selector stops, at the session. for_each() means every session, so this one function is the page of each, at /pages/PlantJournal/sessions/<id>/, keyed by that session alone: an upload to one session changes the key of its page and the root's, no other.
  • @cg.ui.PageBuilder() makes the body the page. Every component the body constructs attaches to the page in order, the way procedure nodes attach to the recording context under @cg.procedure; with cg.ui.HStack(): puts the three figures side by side; a for loop adds one cg.ui.PageLink per session. The body returns nothing, and the wrapper returns the cg.Page: a Markdown summary, three figures, a Vega-Lite chart of sessions per day, a map of each session's location, a gallery, a link per session; one cg.ui.ImageViewer for the session's single photograph with its dimensions. Building the page renders the tree to index.html, copies every file it shows under files/ and every browser library it uses under lib/, and the server hosts that folder byte for byte at the position's ~/. The module never touches HTML.
  • Absence renders nothing. A session page for a session whose photo has not landed keeps its title and nothing else: session.photo is Missing, so cg.ui.ImageViewer(session.photo) renders nothing, session.photo.pil().size chains to Missing and both Stats render nothing, and cg.ui.Markdown(session.notes.value) renders nothing until notes are entered. The body reads the values it wants and never guards them.
  • cg.ui.PageLink(session) links the root to each session's page by the relative path between the two positions on disk: the session came from cg.load, so it knows where it lives, and ../sessions/<id>/ is a URL the server already answers. Nothing is computed to write the link. Only PageLink and the of constructors link; a plain ImageViewer or Gallery shows its pictures and links nowhere.
  • The gallery reuses session_thumbnail. The root page does no per-photo work of its own: each thumbnail is the same cached call the interceptor made, keyed by the photo, so a new upload costs one thumbnail plus the assembly. Keeping the root a thin assembly over per-session recipes is the rule that keeps recomputes cheap.
  • cg.ui.Gallery.of(target.sessions, lambda s: session_thumbnail(s.photo)) makes each thumbnail a link. A thumbnail is computed, not loaded, so it has no position of its own; of walks the sessions map with pmap, so the thumbnails decode in parallel, and links every picture to its session's page, the way the cards below do. A session whose photo has not landed gets Missing back from the recipe and is left out.
  • cg.ui.Title("Plant Journal") sets the page's <title> and renders nothing in the body. The title shows wherever the page is listed: the browser tab, the card for PlantJournal at /pages, and the position's ~status.

Save the module, open http://localhost:4433/pages/PlantJournal/, and the first visit shows Calculating… with the thumbnail calls ticking by until the folder seals. On a target nobody has captured to yet the page still renders, with a map of only the plant's marker once its location is captured, an empty gallery, no links, three zeros, and fills in as sessions arrive from the app; the pages guide's quick start seeds a session from the shell and drives a page with curl. Pages and Visualizations is the reference for all of it, including the URL table, links, and components.

Step 9: Try it#

Serve the examples and ask for a position as two different participants. The caller's identity travels in a header, and an opted-in position in another:

curl -H 'X-CG-User-Id: alice' \
  http://localhost:4433/api/v1/targets/PlantJournal/files/.metadata/.next_notification
curl -H 'X-CG-User-Id: bob' \
  http://localhost:4433/api/v1/targets/PlantJournal/files/.metadata/.next_notification
curl -H 'X-CG-User-Id: alice' -H 'X-CG-Location: 42.4438,-76.5018,0,0' \
  http://localhost:4433/api/v1/targets/ParkSurvey/files/.metadata/.location

Each answer is the position's JSON encoding (a cg.Time is epoch seconds), and two participants get two different answers. The pages are a browser away: http://localhost:4433/pages/PlantJournal/ for the journal, a session's sessions/<id>/ beneath it, and /pages for the index, a card per target with pages. The user header is what an interceptor needs; the location header only changes which spot the survey leases (without it, the oldest uncovered one). Without the user header the position is reported as missing, which is why capturegraph-sync never downloads intercepted positions.

Every interceptor whose selector crosses no container is also dry-run when its target loads (one crossing a container has no single position to run), inside a throwaway scratch so live reservations are untouched; the result, or the error, shows in the target's status in the management panel.

Step 10: Test it#

Tests drive the same API. The shipped capturegraph-server/tests/targets/test_example_targets.py covers both targets; trimmed to its essentials, and run from the repository root, a test looks like this:

import shutil
from pathlib import Path

from fastapi.testclient import TestClient
from server.app import build_app
from server.config import load_or_create_config

EXAMPLES = Path("capturegraph-server/examples")
NOTIFICATION = "/api/v1/targets/PlantJournal/files/.metadata/.next_notification"


def test_two_participants_get_distinct_times(tmp_path: Path) -> None:
    shutil.copytree(EXAMPLES / "PlantJournal", tmp_path / "PlantJournal")
    app = build_app(root=tmp_path, config=load_or_create_config(tmp_path))
    with TestClient(app) as client:
        client.put(
            "/api/v1/targets/PlantJournal/files/.metadata/.location",
            content=b'{"latitude": 42.44, "longitude": -76.5}',
            headers={"X-CG-File-Extension": "json"},
        )
        alice = client.get(NOTIFICATION, headers={"X-CG-User-Id": "alice"})
        bob = client.get(NOTIFICATION, headers={"X-CG-User-Id": "bob"})
    assert alice.json() != bob.json()

The app's startup opens the scratch under the temporary root, so the test needs no fixture. Recipes can also be tested without HTTP: the cg_scratch pytest fixture, shipped with the library, activates an isolated scratch for one test, and notification_candidates or a book is then called directly.

Step 11: The same recipes, offline#

A recipe does not know it is on a server. Pull the target with sync, load it over the same schema, and call the same function from a notebook; the scratch you configure locally serves the cache. Importing the target module needs the server package installed, which the uv workspace provides; elsewhere, redeclare the schema and the recipe:

import capturegraph as cg

from plant_journal import PlantJournal, session_thumbnail

cg.scratch.configure("~/.cache/capturegraph")
journal = cg.load("./PlantJournal", PlantJournal)
thumbnails = journal.sessions.photo.pmap(session_thumbnail)

Only the sessions that are new since the last run render. The reservation books are server state and stay on the server; ReservationBook.open(key).peek() against the server's scratch reads a snapshot of who holds what.

Operating notes#

  • A module edit goes live on the next request, valid or not, when made on disk; the panel's editor validates in a subprocess first and keeps history. Keep module scope cheap: it runs at every reload and validation.
  • Bump identities, never bodies alone. A versioned pure function serves its old cache until the version changes; a book whose fields change starts empty.
  • The scratch is one machine's. File locks and free-space accounting are unreliable on network filesystems, so keep scratch_path on a local disk. Several uvicorn workers on that machine share it safely through the index and locks.
  • A raising interceptor is logged, not surfaced. The client gets whatever is on disk at that position, usually a 404. When an intercepted position "does not exist", read the server log or the target status in the panel.

The complete modules#

The two files, exactly as they ship:

# SPDX-License-Identifier: Apache-2.0

"""A plant time-lapse a crowd captures together, scheduled by the server.

The server fills two kinds of position the phone would otherwise leave to
chance: each session's thumbnail, rendered once per photo, and every
participant's next capture time, spread across the solar angles the dataset is
still missing, one distinct time per person. Its root page shows the crowd
what it has captured: how many sessions on which days, where, a thumbnail per
session from the same cached per-photo call the thumbnail interceptor uses,
and a link to each session's own page, which shows the photograph at full size.
"""

from collections import Counter
from datetime import UTC, datetime, timedelta

import capturegraph as cg
import capturegraph.scheduling as cgsh
from server import authoring as cgserver
from server.notifications import hourly_bucket, notification_candidates


class Session(cg.Struct):
    photo: cg.Image
    notes: cg.String
    _metadata: cg.Metadata


class PlantJournal(cg.Struct):
    sessions: cg.Map[cg.Date, Session]
    _saw_instructions: cg.Bool
    _metadata: cg.Metadata


@cgserver.target
@cg.procedure(PlantJournal, label="Plant Journal")
def plant_journal(root: cg.Procedure[cg.Path[PlantJournal]]) -> None:
    root._saw_instructions |= cg.ShowInstructions(
        text="Photograph the plant from the marked spot.",
    )
    root._metadata._location |= cg.CaptureLocation()
    session = root.sessions[cg.CaptureTime()]
    session.photo &= cg.CaptureImage(label="Photograph the plant")
    session.notes &= cg.UserInputString(label="Notes")


@cg.pure_function("session_thumbnail/1")
def session_thumbnail(photo: cg.Image) -> cg.Thumbnail:
    thumbnail = cg.Thumbnail.new()
    photo.pil(max_axis=256).save(thumbnail)
    return thumbnail


@cgserver.intercept(
    PlantJournal,
    lambda s: s.sessions.for_each()._metadata._thumbnail,
    skip_if_exists=True,
)
def thumbnail(node: Session) -> cg.Thumbnail:
    if cg.is_missing(node.photo):
        raise LookupError("no photo to render yet")
    return session_thumbnail(node.photo)


@cgserver.page(PlantJournal, version=1)
@cg.ui.PageBuilder()
def journal(target: PlantJournal) -> None:
    sessions = target.sessions.to_dict()
    photographed = sum(1 for session in sessions.values() if not cg.is_missing(session.photo))
    days = {instant.date() for instant in sessions}
    cg.ui.Title("Plant Journal")
    cg.ui.Markdown(f"## Sessions\n\n{photographed} photographs of the plant so far.")
    with cg.ui.HStack():
        cg.ui.Stat("Sessions", len(sessions))
        cg.ui.Stat("Photographs", photographed)
        cg.ui.Stat("Days with captures", len(days))
    cg.ui.Chart(spec=sessions_per_day(list(sessions)))
    cg.ui.MapViewer.of(target.sessions, lambda session: session._metadata._location)
    cg.ui.Gallery.of(target.sessions, lambda s: session_thumbnail(s.photo))
    for instant, session in sessions.items():
        cg.ui.PageLink(session, title=instant.date().isoformat())


def sessions_per_day(captured: list[datetime]) -> dict[str, cg.JSONValue]:
    """A Vega-Lite bar chart of how many sessions were captured per day.

    Args:
        captured: Every session's capture instant.

    Returns:
        The specification with its rows inline, one per day.
    """
    counts = Counter(instant.date().isoformat() for instant in captured)
    rows: list[cg.JSONValue] = [
        {"day": day, "sessions": count} for day, count in sorted(counts.items())
    ]
    return {
        "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
        "width": "container",
        "data": {"values": rows},
        "mark": "bar",
        "encoding": {
            "x": {"field": "day", "type": "temporal", "timeUnit": "yearmonthdate", "title": "Day"},
            "y": {"field": "sessions", "type": "quantitative", "title": "Sessions"},
        },
    }


@cgserver.page(PlantJournal, lambda s: s.sessions.for_each(), version=1)
@cg.ui.PageBuilder()
def session(session: Session) -> None:
    cg.ui.Title("Session")
    cg.ui.ImageViewer(session.photo)
    size = session.photo.pil().size
    with cg.ui.HStack():
        cg.ui.Stat("Width", size[0], unit="px")
        cg.ui.Stat("Height", size[1], unit="px")
    cg.ui.Markdown(session.notes.value)


@cgserver.intercept(PlantJournal, lambda s: s._metadata._next_notification)
def notification(profile: cgserver.RequestProfile, target: PlantJournal) -> cg.Time:
    now = datetime.now(UTC)
    location = target._metadata._location
    if cg.is_missing(location):
        return cg.Time(now + timedelta(hours=1))

    candidates = notification_candidates(target.sessions, location, hourly_bucket(now))
    chosen = cgsh.ReservationBook.open("plant-journal").reserve(
        profile.user_id,
        [candidate.value for candidate in candidates],
        captured=max(target.sessions.to_dict(), default=None),
        grace=timedelta(minutes=30),
        now=now,
    )
    return cg.Time(chosen if chosen is not None else now + timedelta(hours=1))

# SPDX-License-Identifier: Apache-2.0

"""A tree survey over a park, where the server sends each participant to an uncovered spot.

The park is a polygon and a grid over it is the pool of survey points. A pure
recipe picks a well-spread batch of points the captures so far leave uncovered,
and a coverage book leases one to each participant, the closest to where they
stand, retiring a point once a capture lands on it.
"""

import capturegraph as cg
import capturegraph.scheduling as cgsh
from server import authoring as cgserver

BOUNDS = cg.Array[cg.Location](
    [
        cg.Location(latitude=42.4440, longitude=-76.5020),
        cg.Location(latitude=42.4440, longitude=-76.5000),
        cg.Location(latitude=42.4425, longitude=-76.5000),
        cg.Location(latitude=42.4425, longitude=-76.5020),
    ]
)
DISTANCE = cgsh.distance.location(sigma_m=10.0)


class Session(cg.Struct):
    photo: cg.Image
    _metadata: cg.Metadata


class ParkSurvey(cg.Struct):
    sessions: cg.Map[cg.Date, Session]
    _metadata: cg.Metadata


@cgserver.target
@cg.procedure(ParkSurvey, label="Park Survey")
def park_survey(root: cg.Procedure[cg.Path[ParkSurvey]]) -> None:
    session = root.sessions[cg.CaptureTime()]
    session._metadata._location &= cg.CaptureLocation()
    session.photo &= cg.CaptureImage(label="Photograph the nearest tree")


@cg.pure_function("survey_batch/1")
def survey_batch(covered: cg.Array[cg.Location]) -> cg.Array[cg.Location]:
    return cgsh.select_sessions(
        potential_sessions=cgsh.forecast.locations_area(BOUNDS, resolution_meters=10.0),
        previous_sessions=covered,
        distance_fn=DISTANCE,
        energy_fn=lambda d: -d,
        energy_mode="max",
        selections=8,
    )


@cgserver.intercept(ParkSurvey, lambda s: s._metadata._location)
def next_spot(profile: cgserver.RequestProfile, target: ParkSurvey) -> cg.Location:
    covered = cg.Array[cg.Location](
        [
            session._metadata._location
            for session in target.sessions.values()
            if not cg.is_missing(session._metadata._location)
        ]
    )
    chosen = cgsh.CoverageBook.open("park-survey").reserve(
        profile.user_id,
        at=profile.location,
        fulfilled_by=list(covered),
        distance=DISTANCE,
        threshold=1.0,
        refill=list(survey_batch(covered)),
    )
    if chosen is None:
        raise LookupError("every survey point is covered")
    return chosen