Light automation

A real-world automation: ensure minimum brightness in an office. This example demonstrates secrets, reusable resource recipes, external service integration, and cron scheduling.

Imports

load("pkg.flowmetal.io/prelude/v1",
     "args",
     "flow",
     "FlowContext",
     "resource",
     "unwrap",
)
load("pkg.flowmetal.io/cron/v1", "cron")
load("pkg.flowmetal.io/home-assistant/v1", "home_assistant_resource")
load("pkg.flowmetal.io/one-password/v1", "op_resource")

Reusable resource recipe

resource() is to inline resource lambdas as flow() is to normal defs — a declarative, composable recipe. This wraps a Home Assistant client to pull its API key from 1Password at open time.

hass = resource(
    open = lambda ctx: home_assistant_resource(
        url = ctx.args.url,
        token = ctx.resources.op.item(
            ctx.args.vault_id,
            ctx.args.secret_id,
        ).field("api-key").value,
    ),
    close = lambda self: self.close(),
    args = {
        "url": args.string(),
        "vault_id": args.string(),
        "secret_id": args.string(),
        "op": args.resource(op_resource),
    },
)

Flow declaration

The flow declares its arguments, secrets, and resources up front. Secrets are validated at startup — if any are missing, the flow fails immediately rather than halfway through execution. Resource recipes can depend on other resources and on secrets; the runtime manages the dependency order.

auto_lights = flow(
    args = {
        "hass_url": args.string(),
        "hass_secret_id": args.string(),
        "luminosity_sensor_id": args.string(),
        "luminosity_threshold": args.int(min = 0, default = 100),
        "occupancy_sensor_id": args.string(),
        "light_id": args.string(),
        "light_on_data": args.json(
            schema = {"properties": {"brightness_pct": {"type": "int"}}},
            default = {"brightness_pct": 100},
        ),
    },
    secrets = ["op_token", "op_url", "op_vault"],
    resources = {
        "op": lambda fctx: op_resource(
            url = fctx.secrets.op_url,
            token = fctx.secrets.op_token,
        ),
        "hass": lambda fctx: hass(
            url = fctx.args.hass_url,
            vault_id = fctx.secrets.op_vault,
            secret_id = fctx.args.hass_secret_id,
            op = fctx.resources.op,
        ),
    },
    implementation = _auto_lights_impl,
)

Implementation

The implementation is short — read sensors, make a decision, act. All the complexity (authentication, resource lifecycle, failure recovery) is handled by the declarations above.

def _auto_lights_impl(fctx: FlowContext):
    occ_state = fctx.resources.hass.states(fctx.args.occupancy_sensor_id)
    lux_state = fctx.resources.hass.states(fctx.args.luminosity_sensor_id)

    if unwrap(occ_state) and unwrap(lux_state) < fctx.args.luminosity_threshold:
        fctx.resources.hass.service(
            "light/turn_on",
            fctx.args.light_id,
            data = fctx.args.light_on_data,
        )

One-shot and scheduled execution

Call the flow directly for a one-shot run. For recurring execution, cron() is just a library flow — it computes the next fire time, sleep()s, calls your function, and repeats. Checkpointing makes it durable. spawn() detaches from the supervision tree so the cron loop outlives this script.

def _main_impl(fctx: FlowContext):
    # One-shot
    auto_lights(
        hass_url = "https://hass.tirefireind.us",
        hass_secret_id = "<redacted>",
        occupancy_sensor_id = "<redacted>",
        luminosity_sensor_id = "<redacted>",
        luminosity_threshold = 100,
        light_id = "<redacted>",
        light_on_data = {"brightness_pct": 35},
    )

    # Scheduled: every 5 minutes
    cron_ref = unwrap(fctx.actions.spawn(
        lambda: cron(
            pattern = "*/5 * * * *",
            on_tick = lambda: auto_lights(
                hass_url = "https://hass.tirefireind.us",
                hass_secret_id = "<redacted>",
                occupancy_sensor_id = "<redacted>",
                luminosity_sensor_id = "<redacted>",
                luminosity_threshold = 100,
                light_id = "<redacted>",
                light_on_data = {"brightness_pct": 35},
            ),
        ),
        task_id = "lights-cron",
        on_conflict = "replace",
    ))

    print(cron_ref)

main = flow(implementation = _main_impl)

if __name__ == "__flow__":
    main()

View the complete example