What an insurer can read after a run: the log, its limits pack, and incident reports

# Scenario 059: What an insurer can read after a run: the log, its limits pack, and incident reports
# After a run ends, the log names the device, firmware and the exact safety-limits pack that was in force, lists safety events, and marks each step's envelope check. Incidents are reported separately and anonymously.
# Run a hub first: python hub/cookwala_hub.py --recipes examples
import json, os
from cookwala.client import CookwalaClient, CookwalaProblem

def load(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)

def pick(obj, path):
    for part in path.replace("]", "").replace("[", ".").split("."):
        obj = obj[int(part)] if part.isdigit() else (len(obj) if part == "length" else obj.get(part))
    return obj

c = CookwalaClient(os.environ.get("COOKWALA_HUB", "http://localhost:7878"))


# Step 1: Start a shakshuka run with a person present.
try:
    started = c.start_execution({"core": "0.2.0", "kind": "ExecuteRequest", "id": "ex-059-ins", "recipe": "cw:cookwala.ai:example-shakshuka", "recipeHash": "sha256:2ea6fab93cc673fc6784f9bc9fe401d2387bb7779dc537df97286b73807c8841", "servings": 4, "requestedBy": "agent:planner-1", "idempotencyKey": "ex-059-ins-a"}, "ex-059-ins-key", True)
except CookwalaProblem as p:
    started = p.body  # a refusal is a result, not a crash
    print("refused:", p.refusal)
assert pick(started, "state") == "accepted", pick(started, "state")
print("ok", "state", "=", "accepted")

# Step 2: The arm cuts; the run waits for a person.
import time; time.sleep(4)

# Step 3: The run is stopped by the operator.
try:
    stop = c.stop_execution("ex-059-ins", "shift change")
except CookwalaProblem as p:
    stop = p.body  # a refusal is a result, not a crash
    print("refused:", p.refusal)
assert pick(stop, "state") == "stopping", pick(stop, "state")
print("ok", "state", "=", "stopping")

# Step 4: The device completes the stop.
import time; time.sleep(1.5)

# Step 5: The log: device vendor, model and firmware, the limits pack by id and version, no safety events, every step with envelopeOk true, and the recipe hash that was cooked.
log = c.execution_log("ex-059-ins")
assert pick(log, "device.vendor") == "Example Robotics", pick(log, "device.vendor")
print("ok", "device.vendor", "=", "Example Robotics")
assert pick(log, "device.safetyLimits") == "[email protected]", pick(log, "device.safetyLimits")
print("ok", "device.safetyLimits", "=", "[email protected]")
assert pick(log, "safetyEvents.length") == 0, pick(log, "safetyEvents.length")
print("ok", "safetyEvents.length", "=", 0)
assert pick(log, "steps[0].envelopeOk") == True, pick(log, "steps[0].envelopeOk")
print("ok", "steps[0].envelopeOk", "=", True)
assert pick(log, "recipeHash") == "sha256:2ea6fab93cc673fc6784f9bc9fe401d2387bb7779dc537df97286b73807c8841", pick(log, "recipeHash")
print("ok", "recipeHash", "=", "sha256:2ea6fab93cc673fc6784f9bc9fe401d2387bb7779dc537df97286b73807c8841")
assert pick(log, "outcome") == "aborted_safe", pick(log, "outcome")
print("ok", "outcome", "=", "aborted_safe")

# Step 6: An incident, had there been one, goes to the catalog as an anonymous report; here the example report.
inc = c.report_incident(load("examples/core/incident.json"))
assert pick(inc, "received") == True, pick(inc, "received")
print("ok", "received", "=", True)

print("scenario complete")