1"""Lab operator using the connector's public JSON-lines interface."""23import json4import os5from pathlib import Path6import selectors7import subprocess8import time91011def public(name):12 return Path(f"/public/{name}.pub").read_text().strip()131415def tag(event, name):16 return next((row[1] for row in event.get("tags", []) if len(row) > 1 and row[0] == name), None)171819class Client:20 def __init__(self, actor="owner"):21 key = json.loads(Path(f"/operator/{actor}.json").read_text())["secret"]22 env = {**os.environ, "TINY_PRIVATE_KEY": key}23 self.process = subprocess.Popen(24 ["tinyagent", "rpc", "--relay", "http://tiny:7447"],25 stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, env=env,26 )27 self.sequence = 028 self.selector = selectors.DefaultSelector()29 self.selector.register(self.process.stdout, selectors.EVENT_READ)3031 def call(self, method, params=None, timeout=30):32 self.sequence += 133 self.process.stdin.write(json.dumps({"id": self.sequence, "method": method, "params": params or {}}) + "\n")34 self.process.stdin.flush()35 deadline = time.monotonic() + timeout36 while time.monotonic() < deadline:37 if not self.selector.select(max(0, deadline - time.monotonic())):38 break39 line = self.process.stdout.readline()40 if not line:41 raise RuntimeError("tinyagent exited before replying")42 message = json.loads(line)43 if message.get("id") != self.sequence:44 continue45 if "error" in message:46 raise RuntimeError(message["error"]["message"])47 return message.get("result")48 raise TimeoutError(f"tinyagent {method} timed out")4950 def publish(self, kind, content, tags):51 return self.call("publish", {"event": {"kind": kind, "content": content, "tags": tags}})5253 def query(self, filter):54 return self.call("query", {"filter": filter}) or []5556 def browse(self, method, params):57 response = self.call("request", {"path": "/manage/rpc", "method": "POST", "body": {58 "jsonrpc": "2.0", "id": 1, "method": method, "params": [params],59 }})60 if response.get("error"):61 raise RuntimeError(str(response["error"]))62 return response.get("result")6364 def close(self):65 self.selector.close()66 self.process.stdin.close()67 try:68 self.process.wait(timeout=5)69 except subprocess.TimeoutExpired:70 self.process.kill()71 self.process.wait()7273 def __enter__(self):74 return self7576 def __exit__(self, *_args):77 self.close()787980def answer(client, request, content):81 request_kind = str(request["kind"])82 return client.publish(1111, content, [83 ["h", tag(request, "h")], ["E", request["id"]], ["K", request_kind], ["P", request["pubkey"]],84 ["e", request["id"]], ["k", request_kind], ["p", request["pubkey"]],85 ])868788def wait_event(client, filter, predicate=lambda _event: True, timeout=90):89 deadline = time.monotonic() + timeout90 while time.monotonic() < deadline:91 events = client.query(filter)92 match = next((event for event in events if predicate(event)), None)93 if match:94 return match95 time.sleep(0.3)96 raise TimeoutError(f"No matching event for {filter}")