Scenario 2 — A Door and a Mobile Robot

The other half of the picture from scenario 1. Here there is no operation queue at all. A mobile robot finds a door in the knowledge graph, reads the door’s live state over REST and, finding it closed, calls the door’s open workflow directly at the URL the graph gave it.

Use this pattern when the caller needs the answer now. Operation coordination (scenario 1) is for work that is dispatched, queued and picked up later; direct invocation is for a request that returns its result to the caller that made it. The same middleware serves both — what differs is whether the caller goes through the graph’s Operation queue or straight to the endpoint.

The scenario also shows what the graph deliberately does not hold. The door’s opened/closed status is served from memory and is never written to the graph, because it is only true for seconds. What the graph holds is where to ask.

Before you run this. A GraphDB must be reachable, with GRAPHDB_URL, GRAPHDB_USERNAME and GRAPHDB_PASSWORD set to point at it. Every cell below works against the kapps-demo repository, which is named in the code rather than read from the environment. Step 1 erases that repository completely, so make sure GRAPHDB_URL names a server you are allowed to overwrite.

import logging
import threading
import time

import httpx
import uvicorn
from rdflib.namespace import RDF

from handlers import door_close, door_open, door_status, reset_door
from kapps_ogm import OGM
from kapps_semantic_middleware import SemanticMiddleware
from kapps_semantic_middleware.credentials import DEMO_REPOSITORY, graphdb_for
from kapps_semantic_middleware.registration import mint_state_property_iri, mint_workflow_iri
from kapps_semantic_middleware.vocabulary import SVC

import seed

# The libraries above log at INFO to stderr. A stderr stream renders as an error block on the
# documentation site, and it is grouped ahead of this notebook's own stdout -- so the HTTP
# request lines would appear above the step that made them. The narrative here is the print
# output; keep third-party logging at WARNING so that is what you read.
for _name in ('', 'httpx', 'httpcore', 'GraphDB', 'KafkaManager', 'kapps_ogm'):
    logging.getLogger(_name).setLevel(logging.WARNING)


def serve(middleware, port):
    """Start `middleware` on a background thread and wait until it has registered.

    The thread is what lets this notebook keep executing cells while a server runs, and it is
    also what makes `stop` below mandatory. Uvicorn installs signal handlers only on the main
    thread, so off it SIGTERM never reaches the ASGI lifespan and the middleware's on_shutdown
    callbacks -- deregistration among them -- never run. `stop` sets should_exit and joins,
    which runs that same lifespan shutdown with no signal involved. Copy this helper and you
    inherit both halves: a copy that never calls `stop` leaks a Service on every run.
    """
    config = uvicorn.Config(middleware.app, host='127.0.0.1', port=port, log_level='warning')
    server = uvicorn.Server(config)
    thread = threading.Thread(target=server.run, daemon=True)
    thread.start()
    t0 = time.time()
    while not server.started and time.time() - t0 < 30:
        time.sleep(0.05)
    if not server.started:
        raise RuntimeError(f'server on port {port} did not start in time')
    return server, thread


def stop(server, thread):
    """Stop a served middleware so its deregistration callbacks run."""
    server.should_exit = True
    thread.join(timeout=20)


db = graphdb_for(DEMO_REPOSITORY)
print(f"Connected to GraphDB repository: {db.repository}")

Connected to GraphDB repository: kapps-demo

Step 1 — Seed a Clean Repository

reset_door() puts the in-memory door back to closed, so that the notebook behaves the same on a second run as on the first. seed_scenario2 then clears the repository, loads the ontology this scenario needs, and creates the door and mobile-robot individuals.

As in scenario 1, the classes the middleware registers against must exist before an instance starts. The middleware looks them up; it never creates them.

reset_door()
seed.seed_scenario2(db)
print('Door resource: ', db.triple_exists((seed.DOOR_RESOURCE, RDF.type, seed.DOOR_RESOURCE_CLASS)))
print('Robot resource:', db.triple_exists((seed.MOBILE_ROBOT, RDF.type, seed.MOBILE_ROBOT_RESOURCE_CLASS)))

Door resource:  True

Robot resource: True

Step 2 — Start the Door Middleware

The door declares three things, and the difference between them is the point of this scenario.

workflow(...) registers door_open and door_close: actions that change the world and return when they are done. state(...) registers door_status: a reading that answers a question and changes nothing. Both become callable endpoints and both are advertised in the graph, but they are different kinds of thing and are typed differently in the ontology.

Nothing here creates a queue. These endpoints execute synchronously when they are called, which is what makes the robot’s logic in step 4 a straight line.

Starting the server is again what performs registration, and step 6 is again what undoes it.

door = SemanticMiddleware(
    mode='resource', resource_iri=seed.DOOR_RESOURCE,
    service_class=seed.DOOR_SERVICE_CLASS, ogm=OGM(db=db),
    host='127.0.0.1', port=8997,
)
door.workflow(capability_class=seed.DOOR_OPEN_CAPABILITY_CLASS, workflow_class=seed.DOOR_OPEN_WORKFLOW_CLASS)(door_open)
door.workflow(capability_class=seed.DOOR_CLOSE_CAPABILITY_CLASS, workflow_class=seed.DOOR_CLOSE_WORKFLOW_CLASS)(door_close)
door.state(capability_class=seed.DOOR_STATUS_CAPABILITY_CLASS, state_property_class=seed.DOOR_STATUS_STATE_CLASS)(door_status)

server, thread = serve(door, 8997)
print('Door middleware started on port 8997')

Door middleware started on port 8997

Step 3 — Inspect What Registration Wrote

The same structural triples as scenario 1, plus one more kind: a StateProperty belongs to its Service through isStatePropertyOf, exactly as a Workflow does through isWorkflowOf.

The endpoints printed here are the ones the robot discovers in the next step. Nothing hands them to the robot — they are in the graph, and the robot queries for them. This is what discovery through the graph means concretely: a URL written by whoever started the server, read by whoever needs to call it.

# Per-instance — read off the instance rather than rebuilt from the resource.
service_iri = door.service_iri
open_wf = mint_workflow_iri(service_iri, 'door_open')
status_sp = mint_state_property_iri(service_iri, 'door_status')
assert db.triple_exists((open_wf, SVC.isWorkflowOf, service_iri))
assert db.triple_exists((status_sp, SVC.isStatePropertyOf, service_iri))
print('open workflow endpoint:', list(db.triples_get(sub=open_wf, pred=SVC.endpoint)))
print('status state endpoint: ', list(db.triples_get(sub=status_sp, pred=SVC.endpoint)))

open workflow endpoint: [(IRI('https://example.org/kapps-demo#door_042_service_http_c__s__s_127_d_0_d_0_d_1_c_8997_workflow_door_open'), IRI('https://w3id.org/circularfactory/Service#endpoint'), 'http://127.0.0.1:8997/workflows/door_open/execute')]
status state endpoint:  [(IRI('https://example.org/kapps-demo#door_042_service_http_c__s__s_127_d_0_d_0_d_1_c_8997_state_door_status'), IRI('https://w3id.org/circularfactory/Service#endpoint'), 'http://127.0.0.1:8997/state/door_status')]

Step 4 — The Mobile Robot Discovers and Passes the Door

The robot is a served middleware instance in its own right, for the same reason the planner was in scenario 1: a resource-mode instance that never runs never registers, and a robot that drives a door is a peer the door could ring back.

The SPARQL query is the interesting part. Read it as a sentence: find the Service of this door, then the state property on that Service that is a door-status property, and the workflow on that Service that is a door-open workflow, and give me the URLs of both. The robot asks for the door by its identity and by the types of the things it needs — never by a hostname or a port. Any door modelled the same way answers the same query, which is what makes the behaviour reusable rather than wired to this one machine.

What follows is deliberately dull: read the status, and if it is closed, POST to the open URL. Two HTTP calls, to addresses that were unknown when the cell started running.

The door closes itself again after 30 seconds. The robot’s round trip takes less than that, so on the way back it finds the door still open and does not open it twice.

robot = SemanticMiddleware(
    mode='resource', resource_iri=seed.MOBILE_ROBOT,
    service_class=seed.MOBILE_ROBOT_SERVICE_CLASS, ogm=OGM(db=graphdb_for(DEMO_REPOSITORY)),
    host='127.0.0.1', port=8998,
)
# The robot is served, not merely constructed. on_start_up is what writes its Service
# individual and its svc:address, so an unserved resource-mode instance makes mode='resource' a
# label with no runtime consequence -- and this notebook is meant to be copied.
robot_server, robot_thread = serve(robot, 8998)
robot_service = robot.service_iri
print('Mobile robot served on port 8998')
print('  robot address in graph:', list(db.triples_get(sub=robot_service, pred=SVC.address)))

sparql = f'''
SELECT ?status_url ?open_url WHERE {{
    ?svc <{SVC.isServiceOf}> <{seed.DOOR_RESOURCE}> .
    ?sp <{SVC.isStatePropertyOf}> ?svc .
    ?sp a <{seed.DOOR_STATUS_STATE_CLASS}> .
    ?sp <{SVC.endpoint}> ?status_url .
    ?wf <{SVC.isWorkflowOf}> ?svc .
    ?wf a <{seed.DOOR_OPEN_WORKFLOW_CLASS}> .
    ?wf <{SVC.endpoint}> ?open_url .
}}'''
b = robot.ogm.db.query(sparql, convert_bindings=True)['results']['bindings'][0]
status_url, open_url = str(b['status_url']), str(b['open_url'])
print('discovered via SPARQL ->', status_url, '|', open_url)

def ensure_open(phase):
    state = httpx.get(status_url).json()
    print(f'  {phase}: door is {state}')
    if state == 'closed':
        httpx.post(open_url)  # invoke the open workflow directly at its execute URL
        state = httpx.get(status_url).json()
        print(f'  {phase}: opened it -> {state}')
    assert state == 'opened'

ensure_open('approach')
print('  drove through the door')
ensure_open('return')
print('  drove back through the door')

Mobile robot served on port 8998
  robot address in graph: [(IRI('https://example.org/kapps-demo#mobile_robot_007_service_http_c__s__s_127_d_0_d_0_d_1_c_8998'), IRI('https://w3id.org/circularfactory/Service#address'), 'http://127.0.0.1:8998')]
discovered via SPARQL -> http://127.0.0.1:8997/state/door_status | http://127.0.0.1:8997/workflows/door_open/execute
  approach: door is closed
  approach: opened it -> opened
  drove through the door
  return: door is opened
  drove back through the door

Step 5 — The Live Status is Never Persisted

This cell asserts a negative: nowhere on the state property is there a literal opened or closed.

The graph holds the state property’s identity, its type, and the URL to read it at. The value itself lives in the door’s own memory and is served over REST on request. Writing it to the graph would create a second copy that is stale the moment the door moves, and a consumer would have no way to tell which copy was true.

The rule this follows: the graph describes what exists and where to ask; the resource answers what is true right now.

for _, _, obj in db.triples_get(sub=status_sp):
    assert str(obj) not in ('opened', 'closed'), 'state value must not be persisted'
print("Confirmed: no 'opened'/'closed' literal on the state property.")

Confirmed: no 'opened'/'closed' literal on the state property.

Step 6 — Shutdown and Deregistration

Stopping the door and the robot removes their reachability triples — including the state property’s endpoint, which is what stops another peer from trying to read a status from a process that has exited.

The individuals stay. After this cell the door is still fully described in the graph, and the query the robot ran in step 4 returns nothing, because there is no endpoint left to return. That is the intended answer: the door exists, and right now there is nowhere to ask.

stop(server, thread)
stop(robot_server, robot_thread)
time.sleep(0.5)
reset_door()
print('state endpoint removed: ', not list(db.triples_get(sub=status_sp, pred=SVC.endpoint)))
print('robot address removed:  ', not list(db.triples_get(sub=robot_service, pred=SVC.address)))
print('state property preserved:', db.triple_exists((status_sp, RDF.type, seed.DOOR_STATUS_STATE_CLASS)))
print('robot Service preserved: ', db.triple_exists((robot_service, SVC.isServiceOf, seed.MOBILE_ROBOT)))

state endpoint removed:  True
robot address removed:   True
state property preserved: True
robot Service preserved:  True