Light automation

"""
Turn the lights on in my office when it gets dim
"""

# Explicitly bring in the prelude
# This is done for you but for examples exlicit helps.
load("pkg.flowmetal.io/prelude/v1",
     "cron_trigger",
     "current_flow",
     "flow",
     "resource",
     "retry",
)
# Bring in
load("pkg.flowmetal.io/home-assistant/v1",
     "home_assistant_resource",
)
load("pkg.flowmetal.io/one-password/v1",
     "op_resource",
)

# Wrap the hass client to use a secret from 1Password.
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,
    ),
    # This is the default close, but spell it out
    close = lambda self: self.close(),
    args = {
        "url": args.string(),
        "vault_id": args.string(),
        "secret_id": args.string(),
        # This means "an instace of op" not "the op"
        "op": args.resource(op),
    }
)

# A JSON schema describing allowed data
SCHEMA = {"properties": {
    "brightness_pct": {"type": "int"}
}}

# Our workflow
auto_lights = flow(
    # Arguments the workflow caller will provide
    args = {
        "hass_url": args.string(),
        "hass_secret_id": args.string(),
        "luminocity_sensor_id": args.string(),
        "luminocity_threshold": args.int(
            min=0,
            default=100,
        ),
        "occupancy_sensor_id": args.string(),
        "light_id": args.string(),
        "light_on_data": args.json(
            schema=SCHEMA,
            default={"brightness_pct": 100},
        ),
    },
    # Arguments we read as eponymous secrets.
    # Secrets come from the user and from Flowmetal.
    secrets = [
        "op_token",
        "op_url",
        "op_vault",
    ],
    # Recipes for building resources used in this flow.
    # Resource recipes can consume other resources.
    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 = ctx.resources.op,
        ),
    },
    # And this is the code to run
    implementation = _flow,
)

def _flow(fctx: FlowContext):
    # Read the sensor state
    occ_state = fctx.resources.hass.states(
        fctx.args.occupancy_id,
    )
    lux_state = fctx.resources.hass.states(
        fctx.args.luminocity_sensor_id,
    )

    # If the room is occupied and the room is
    # below the lux threshold, turn the light on
    if unwrap(occ_state) \
       and fctx.args.luminocity_threshold < unwrap(lux_state):
        fctx.resources.hass.service(
            "light/turn_on",
            fctx.args.light_id,
            data = fctx.args.light_on_data,
        )


if __name__ == "__flow__":
    # This body itself happens as a flow. We could
    # just call `auto_lights`. That would be a
    # one-shot job. Instead we want to create a
    # daemon with a time trigger.
    l = lambda *_, **__: auto_lights(
        hass_url = "https://hass.tirefireind.us",
        hass_secret_id = "<redacted>",
        occupancy_sensor_id = "<redacted>",
        luminocity_sensor_id = "<redacted>",
        luminocity_threshold = 100,
        light_id = "<redacted>",
         # Prefer mood lighting in the office
        light_on_data = dict(brightness_pct = 35),
    )
    # One-shot call the lights flow in the current context
    l()

    # We can wrap up the lights flow into something
    # that'll run on a schedule and spawn that off
    # in order to create a service.
    spawn_res = current_flow().spawn(
        l,
        task_id = "lights",
        on_conflict = "replace",
        triggers = [
            cron_trigger(
                pattern = "*/5 * * * *",
            ),
        ]
)