"""Persistent MCSJ Java Access Bridge daemon. Run with 32-bit Python: C:\MCSJAgent\Python311-32-embed\python.exe C:\MCSJAgent\mcsj-jab-daemon.py HTTP API on 127.0.0.1:8765: GET /state POST /refresh {"max_depth": 10, "max_nodes": 1500} GET /find?text=P.O.%20Listing POST /click {"text": "P.O. Listing"} or {"id": 108} POST /set {"text": "Description", "value": "Supplies"} or {"id": 123, "value": "Supplies"} GET /cache GET /health Purpose: keep the JAB connection and component cache alive so Michel does not pay the Python startup + Access Bridge attach + full Java tree walk cost on every action. """ import json import time import threading import traceback from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse, parse_qs from pathlib import Path from pyjab.jabdriver import JABDriver HOST = "127.0.0.1" PORT = 8765 TITLE_PREFIX = "MCSJ - 2026.1" DLL = r"C:\Edmunds\MCSJ26.1\jre\bin\WindowsAccessBridge-32.dll" CACHE_PATH = Path(r"C:\MCSJAgent\cache\mcsj-jab-cache.json") DEFAULT_MAX_DEPTH = 10 DEFAULT_MAX_NODES = 1500 class MCSJBridge: def __init__(self): self.lock = threading.RLock() self.driver = None self.root = None self.cache = [] self.elements = {} self.last_refresh = None self.last_error = None self.title = None def discover_title(self): import win32gui titles = [] def cb(hwnd, _): t = win32gui.GetWindowText(hwnd) if t.startswith(TITLE_PREFIX): titles.append(t) win32gui.EnumWindows(cb, None) return titles[0] if titles else TITLE_PREFIX def connect(self): with self.lock: self.title = self.discover_title() self.driver = JABDriver(title=self.title, bridge_dll=DLL, timeout=10) self.root = self.driver.root_element self.last_error = None return self.state() def ensure(self): if self.driver is None or self.root is None: self.connect() return self.driver def info(self, el, n=None, depth=0, parent=None): try: raw = el.get_element_information() except Exception: raw = {} states = raw.get("states") or [] bounds = raw.get("bounds") or {} return { "id": n, "parent": parent, "depth": depth, "name": raw.get("name", "") or "", "role": raw.get("role", "") or "", "role_en_us": raw.get("role_en_us") or raw.get("role_en_US") or raw.get("role", "") or "", "description": raw.get("description", "") or "", "states": states, "children_count": int(raw.get("children_count", 0) or 0), "bounds": bounds, "visible": "visible" in states, "showing": "showing" in states, "enabled": "enabled" in states, "editable": "editable" in states, "selectable": "selectable" in states, "focusable": "focusable" in states, } def refresh(self, max_depth=DEFAULT_MAX_DEPTH, max_nodes=DEFAULT_MAX_NODES): start = time.time() with self.lock: self.ensure() rows = [] elements = {} stack = [(self.root, 0, None)] while stack and len(rows) < max_nodes: el, depth, parent = stack.pop() eid = len(rows) + 1 row = self.info(el, eid, depth, parent) rows.append(row) elements[eid] = el if depth < max_depth: try: kids = list(el._generate_childs_from_element(el, visible=False)) except Exception: kids = [] # push reversed so display order is natural for child in reversed(kids): stack.append((child, depth + 1, eid)) self.cache = rows self.elements = elements self.last_refresh = time.time() self.last_error = None CACHE_PATH.parent.mkdir(parents=True, exist_ok=True) CACHE_PATH.write_text(json.dumps({ "ok": True, "title": self.title, "refreshed_at": self.last_refresh, "elapsed_sec": round(self.last_refresh - start, 3), "max_depth": max_depth, "max_nodes": max_nodes, "count": len(rows), "rows": rows, }, indent=2), encoding="utf-8") return {"ok": True, "count": len(rows), "elapsed_sec": round(time.time() - start, 3), "cache_path": str(CACHE_PATH)} def state(self): d = self.driver root_info = None if self.root is not None: try: root_info = self.info(self.root) except Exception as e: root_info = {"error": repr(e)} return { "ok": d is not None, "title": self.title, "hwnd": int(d.hwnd) if d is not None else None, "pid": d.pid if d is not None else None, "vmid": d.vmid if d is not None else None, "version": d.get_version_info() if d is not None else None, "cache_count": len(self.cache), "last_refresh": self.last_refresh, "last_error": self.last_error, "root": root_info, } def hay(self, row): return " | ".join(str(row.get(k, "")) for k in ("name", "description", "role", "role_en_us")).lower() def find(self, text, refresh_if_empty=True): with self.lock: if refresh_if_empty and not self.cache: self.refresh() needle = text.lower() hits = [r for r in self.cache if needle in self.hay(r)] hits.sort(key=lambda r: ( not r.get("showing"), not r.get("enabled"), 0 if r.get("role") in ("menu item", "push button", "text", "combo box") else 1, r.get("depth", 99), len(r.get("name", "")), )) return hits def resolve(self, payload): if "id" in payload and payload["id"] is not None: eid = int(payload["id"]) if eid not in self.elements: raise KeyError(f"element id {eid} is not in current cache; refresh and retry") return eid, self.elements[eid], self.cache[eid-1] text = payload.get("text") or payload.get("name") if not text: raise ValueError("payload must include id or text") hits = self.find(text) if not hits: raise KeyError(f"no element matching {text!r}") row = hits[0] return row["id"], self.elements[row["id"]], row def click(self, payload): with self.lock: eid, el, row = self.resolve(payload) el.click() return {"ok": True, "clicked": row} def set_text(self, payload): value = payload.get("value") if value is None: raise ValueError("payload must include value") with self.lock: eid, el, row = self.resolve(payload) try: el.clear() except Exception: pass el.send_text(str(value)) return {"ok": True, "set": row, "value": str(value)} def click_at(self, x, y): """Find and click element at physical screen coordinates using getAccessibleContextAt.""" from ctypes import byref from pyjab.common.types import JOBJECT64 from pyjab.jabelement import JABElement with self.lock: self.ensure() root_ac = self.root._accessible_context vmid = self.driver.vmid bridge = self.driver.bridge hwnd = self.driver.hwnd out_ac = JOBJECT64() ok = bridge.getAccessibleContextAt(vmid, root_ac, int(x), int(y), byref(out_ac)) if not ok: raise RuntimeError(f"getAccessibleContextAt({x},{y}) returned False — no element at that position") el = JABElement(bridge=bridge, hwnd=hwnd, vmid=vmid, accessible_context=out_ac) row = self.info(el, n=None, depth=99) el.click() return {"ok": True, "x": x, "y": y, "found": row} def info_at(self, x, y): """Return info about element at physical screen coordinates without clicking.""" from ctypes import byref from pyjab.common.types import JOBJECT64 from pyjab.jabelement import JABElement with self.lock: self.ensure() root_ac = self.root._accessible_context vmid = self.driver.vmid bridge = self.driver.bridge hwnd = self.driver.hwnd out_ac = JOBJECT64() ok = bridge.getAccessibleContextAt(vmid, root_ac, int(x), int(y), byref(out_ac)) if not ok: return {"ok": False, "error": f"no element at ({x},{y})"} el = JABElement(bridge=bridge, hwnd=hwnd, vmid=vmid, accessible_context=out_ac) row = self.info(el, n=None, depth=99) return {"ok": True, "x": x, "y": y, "found": row} def probe_children(self, node_id, max_children=60): """Force-enumerate children of a cached node using getAccessibleChildFromContext, even if children_count==0. Useful for JToolBar etc. that hide from normal tree walk.""" from pyjab.jabelement import JABElement with self.lock: self.ensure() if node_id not in self.elements: raise KeyError(f"node {node_id} not in cache; refresh first") parent_el = self.elements[node_id] parent_row = self.cache[node_id - 1] vmid = self.driver.vmid bridge = self.driver.bridge hwnd = self.driver.hwnd results = [] for i in range(max_children): try: child_ac = bridge.getAccessibleChildFromContext(vmid, parent_el._accessible_context, i) if child_ac is None or (hasattr(child_ac, 'value') and child_ac.value == 0): break child_el = JABElement(bridge=bridge, hwnd=hwnd, vmid=vmid, accessible_context=child_ac) row = self.info(child_el, n=i, depth=parent_row.get("depth", 0) + 1, parent=node_id) results.append(row) except Exception as e: results.append({"index": i, "error": str(e)}) break return {"ok": True, "node_id": node_id, "children_found": len(results), "children": results} def request_focus(self, node_id): """Call requestFocus on a cached node.""" with self.lock: self.ensure() if node_id not in self.elements: raise KeyError(f"node {node_id} not in cache; refresh first") el = self.elements[node_id] vmid = self.driver.vmid bridge = self.driver.bridge ok = bridge.requestFocus(vmid, el._accessible_context) row = self.info(el) return {"ok": bool(ok), "focused": row} def do_action(self, node_id, action_name="click"): """Try to perform an accessible action on a cached node.""" from ctypes import byref, create_string_buffer, c_int with self.lock: self.ensure() if node_id not in self.elements: raise KeyError(f"node {node_id} not in cache; refresh first") el = self.elements[node_id] row = self.info(el) try: el.click() return {"ok": True, "method": "el.click()", "node": row} except Exception as e1: try: el.do_action(action_name) return {"ok": True, "method": f"el.do_action({action_name!r})", "node": row} except Exception as e2: return {"ok": False, "errors": [str(e1), str(e2)], "node": row} bridge = MCSJBridge() class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *args): print("%s - - [%s] %s" % (self.client_address[0], self.log_date_time_string(), fmt % args), flush=True) def read_json(self): n = int(self.headers.get("Content-Length", "0") or 0) if not n: return {} return json.loads(self.rfile.read(n).decode("utf-8")) def send_obj(self, obj, code=200): data = json.dumps(obj, indent=2, default=str).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def handle_error(self, e): bridge.last_error = f"{type(e).__name__}: {e}" self.send_obj({"ok": False, "error_type": type(e).__name__, "error": str(e), "traceback": traceback.format_exc(limit=6)}, 500) def do_GET(self): try: u = urlparse(self.path) qs = parse_qs(u.query) if u.path == "/health": self.send_obj({"ok": True, "service": "mcsj-jab-daemon", "port": PORT}) elif u.path == "/state": bridge.ensure() self.send_obj(bridge.state()) elif u.path == "/find": text = (qs.get("text") or [""])[0] if not text: self.send_obj({"ok": False, "error": "missing text"}, 400) else: self.send_obj({"ok": True, "query": text, "count": len(bridge.find(text)), "hits": bridge.find(text)}) elif u.path == "/cache": self.send_obj({"ok": True, "count": len(bridge.cache), "rows": bridge.cache}) else: self.send_obj({"ok": False, "error": "not found"}, 404) except Exception as e: self.handle_error(e) def do_POST(self): try: u = urlparse(self.path) payload = self.read_json() if u.path == "/refresh": self.send_obj(bridge.refresh(int(payload.get("max_depth", DEFAULT_MAX_DEPTH)), int(payload.get("max_nodes", DEFAULT_MAX_NODES)))) elif u.path == "/click": self.send_obj(bridge.click(payload)) elif u.path == "/click-at": self.send_obj(bridge.click_at(payload.get("x"), payload.get("y"))) elif u.path == "/info-at": self.send_obj(bridge.info_at(payload.get("x"), payload.get("y"))) elif u.path == "/set": self.send_obj(bridge.set_text(payload)) elif u.path == "/probe-children": self.send_obj(bridge.probe_children(int(payload.get("node_id", 0)), int(payload.get("max_children", 60)))) elif u.path == "/request-focus": self.send_obj(bridge.request_focus(int(payload.get("node_id", 0)))) elif u.path == "/do-action": self.send_obj(bridge.do_action(int(payload.get("node_id", 0)), payload.get("action", "click"))) else: self.send_obj({"ok": False, "error": "not found"}, 404) except Exception as e: self.handle_error(e) if __name__ == "__main__": print(f"Starting MCSJ JAB daemon on http://{HOST}:{PORT}", flush=True) try: print(json.dumps(bridge.connect(), indent=2), flush=True) except Exception as e: print("Initial connect failed:", repr(e), flush=True) ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()