Control Flow#

The recorder gives you side-effect steps, conditionals, loops, and optional blocks as ordinary Python — with blocks and a handful of functions.

Void steps: cg.do#

A step with no stored value (a metronome, an assertion, an acknowledgement) is recorded with cg.do:

import capturegraph as cg


@cg.procedure(MyTarget)
def capture(root):
    cg.do(cg.ShowMetronome(bpm=60.0))
    root.reference |= cg.CaptureImage(label="Reference")

cg.do records a cg.Void node in order; everything else in the body runs relative to it.

Instructions: cache the acknowledgement#

cg.ShowInstructions is not a void step — it returns a cg.Bool (the user's acknowledgement), so never cg.do(cg.ShowInstructions(...)). Cache it into a hidden field with |=, and it shows once: the first session records the acknowledgement, and every later session finds the slot filled and skips the screen.

class MyTarget(cg.Struct):
    _saw_instructions: cg.Bool
    reference: cg.Image
    sessions: cg.Map[cg.Date, Session]


@cg.procedure(MyTarget)
def capture(root):
    root._saw_instructions |= cg.ShowInstructions(text="Stand on the marked spot.")
    root.reference |= cg.CaptureImage(label="Reference")

The leading underscore hides the field on disk (it lives at .saw_instructions); &= instead of |= would re-show the instructions every session.

Conditionals: cg.when#

with cg.when(condition): runs the block only when the condition is true. It lowers to an IfThenElse whose else-branch does nothing.

@cg.procedure(MyTarget)
def capture(root):
    hi_res = cg.UserInputBool(label="High resolution?")
    with cg.when(hi_res):
        session = root.sessions[cg.CaptureTime()]
        session.primary &= cg.CaptureImage(label="Primary")

The condition is a node that produces a cg.Bool — typically a UserInputBool, or a boolean expression (below).

For a two-sided branch, build the IfThenElse node directly: both true_branch and false_branch must produce the same type, and the node takes on that type.

Guarding on a stored value: path.exists()#

path.exists() is a cg.Bool that is true once a value has been stored at a position. It reads the position from disk and reports whether that read completes (a ProcedureCompleted over a LoadPath), so a missing value yields false instead of aborting the run. Pair it with cg.when to run a block only after some other run has written the slot — e.g. capture only once a survey boundary traced in an earlier session exists:

@cg.procedure(MyTarget)
def capture(root):
    root.gps_bounds |= cg.UserInputLocationSequence(label="Trace the boundary")
    with cg.when(root.gps_bounds.exists()):
        session = root.sessions[cg.CaptureTime()]
        session.primary &= cg.CaptureImage(label="Primary")

The read goes to disk regardless of any value cached at the slot this run, so it tests what was present before the run: the first session only traces the boundary, and later sessions — finding it on disk — capture.

Branching on a choice#

To let the user pick one of several options and act on the choice, take a cg.String from cg.UserInputSelectString and guard one block per option with cg.when, comparing the choice against a constant with cg.ProceduresEqual. Each block caches into its own slot, so only the matching branch runs:

class Session(cg.Struct):
    photo: cg.Image
    video: cg.Video
    notes: cg.String


@cg.procedure(MyTarget)
def capture(root):
    session = root.sessions[cg.CaptureTime()]
    choice = cg.UserInputSelectString(
        label="What would you like to capture?",
        options=["Photo", "Video", "Notes"],
    )
    with cg.when(cg.ProceduresEqual(left=choice, right=cg.const("Photo"))):
        session.photo &= cg.CaptureImage(label="Photo")
    with cg.when(cg.ProceduresEqual(left=choice, right=cg.const("Video"))):
        session.video &= cg.CaptureVideo(label="Video")
    with cg.when(cg.ProceduresEqual(left=choice, right=cg.const("Notes"))):
        session.notes &= cg.UserInputString(label="Notes")

cg.ProceduresEqual compares the results of two nodes, so wrap each option string in cg.const. This same shape scales to a clinical decision tree — one when per answer, each guarding the step that answer calls for.

Optional blocks: cg.optional#

with cg.optional(): runs the block but tolerates it failing or being skipped — its failure never fails the surrounding procedure. Use it for secondary captures and background work.

@cg.procedure(MyTarget)
def capture(root):
    session = root.sessions[cg.CaptureTime()]
    session.primary &= cg.CaptureImage(label="Primary")
    with cg.optional():
        session.weather &= cg.CaptureWeather()

Loops#

with cg.while_repeat(condition, array_path) as element: repeats the block, appending one entry to array_path per iteration. The body runs, then the condition is evaluated; while it is true (and max_iterations is not reached) the body runs again.

class Session(cg.Struct):
    extras: cg.Array[cg.Image]


@cg.procedure(MyTarget)
def capture(root):
    session = root.sessions[cg.CaptureTime()]
    with cg.while_repeat(cg.UserInputBool(label="Another?"), session.extras) as item:
        item &= cg.CaptureImage(label="Extra photo")

element is the freshly appended array entry; assign into it like any path. Pass max_iterations= to bound the loop. On the device, the body and condition nodes are cloned per iteration, so each gets independent state; the serializer records which nodes belong to the body in the loop's template_nodes setting automatically.

Boolean and numeric expressions#

Nodes that produce a cg.Bool or cg.Number support Python operators, so conditions read naturally:

a = cg.UserInputBool(label="A")
b = cg.UserInputBool(label="B")
cond = a & ~b                 # AND, NOT; | is OR

n = cg.UserInputNumber(label="Reading")
is_high = n > cg.const(10)    # <, >, <=, >= ; wrap a literal with cg.const

See Also#

• Recorder and Caching — path navigation and |= / &=
• Types and Nodes — the node set
• The JSON Format — how loops serialize