109 lines
4.4 KiB
Python
109 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""LOCAL RUNNER — runs on the operator's own machine (Windows), set up ONCE, to make Altium 365
|
|
Part Request submission the automatic tail of a skill run.
|
|
|
|
Why this exists: the library-manager skill runs in the cloud, but the Part Request web form must
|
|
be driven from the operator's signed-in Chrome on their desktop — the cloud can't reach it. So at
|
|
the end of a run the skill drops the finished files + a `part_requests.json` manifest into an
|
|
**inbox folder on this machine** (via the device bridge). This runner watches that inbox and, for
|
|
each manifest: makes sure Chrome is running with a debugging port on the operator's profile
|
|
(launching it if needed — the persisted login is reused), runs `altium365_part_request.py`, then
|
|
moves the manifest to a `processed/` folder.
|
|
|
|
Set it up once (Task Scheduler at logon, or a startup shortcut) and every future skill run
|
|
auto-submits its Part Requests with no further action — as long as this machine is on and the
|
|
Chrome profile has been signed into the Workspace at least once.
|
|
|
|
Config: a `runner_config.json` next to this file, e.g.
|
|
{
|
|
"base": "https://vecmocon-technologies-pvt-ltd.365.altium.com",
|
|
"inbox": "D:/User Data/Desktop/altium-library-master/_part_request_inbox",
|
|
"processed": "D:/User Data/Desktop/altium-library-master/_part_request_inbox/processed",
|
|
"chrome": "C:/Program Files/Google/Chrome/Application/chrome.exe",
|
|
"user_data_dir": "%LOCALAPPDATA%/Google/Chrome/User Data",
|
|
"debug_port": 9222,
|
|
"chromedriver": null,
|
|
"review_first": false,
|
|
"poll_seconds": 30
|
|
}
|
|
|
|
Run:
|
|
python local_runner.py # watch the inbox forever (use for Task Scheduler / startup)
|
|
python local_runner.py --once # process whatever is in the inbox now, then exit
|
|
"""
|
|
import glob, json, os, shutil, socket, subprocess, sys, time
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
def load_cfg():
|
|
with open(os.path.join(HERE, "runner_config.json"), encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _port_open(port):
|
|
s = socket.socket(); s.settimeout(1)
|
|
try:
|
|
s.connect(("127.0.0.1", int(port))); return True
|
|
except Exception:
|
|
return False
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def ensure_chrome(cfg):
|
|
"""Make sure a Chrome with the debug port is up (reusing the operator's logged-in profile)."""
|
|
if _port_open(cfg["debug_port"]):
|
|
return
|
|
udd = os.path.expandvars(cfg["user_data_dir"])
|
|
subprocess.Popen([cfg["chrome"], f"--remote-debugging-port={cfg['debug_port']}",
|
|
f"--user-data-dir={udd}"])
|
|
for _ in range(40):
|
|
if _port_open(cfg["debug_port"]):
|
|
time.sleep(2) # give the profile a moment to restore the session
|
|
return
|
|
time.sleep(1)
|
|
raise RuntimeError("Chrome did not expose a debug port — check 'chrome' path and 'debug_port'.")
|
|
|
|
|
|
def process(cfg, manifest):
|
|
ensure_chrome(cfg)
|
|
cmd = [sys.executable, os.path.join(HERE, "altium365_part_request.py"),
|
|
"--manifest", manifest, "--base", cfg["base"],
|
|
"--debugger-address", f"127.0.0.1:{cfg['debug_port']}",
|
|
"--out", manifest + ".result.json"]
|
|
if cfg.get("chromedriver"):
|
|
cmd += ["--chromedriver", cfg["chromedriver"]]
|
|
# Default is REVIEW mode: fill the form completely (fields + parameters + attachments) and
|
|
# leave it on screen WITHOUT saving, so the operator reviews and clicks Save themselves.
|
|
# This is the review-first gate and it works even when the runner runs unattended (unlike a
|
|
# console "press Enter"). Set "auto_submit": true in the config only to submit without review.
|
|
if not cfg.get("auto_submit", False):
|
|
cmd += ["--no-submit"]
|
|
subprocess.run(cmd, check=False)
|
|
|
|
|
|
def main():
|
|
cfg = load_cfg()
|
|
os.makedirs(cfg["inbox"], exist_ok=True)
|
|
os.makedirs(cfg["processed"], exist_ok=True)
|
|
once = "--once" in sys.argv
|
|
print(f"watching {cfg['inbox']} (base {cfg['base']})")
|
|
while True:
|
|
for m in sorted(glob.glob(os.path.join(cfg["inbox"], "*.json"))):
|
|
if m.endswith(".result.json"):
|
|
continue
|
|
print("processing", m)
|
|
try:
|
|
process(cfg, m)
|
|
except Exception as e:
|
|
print(" error:", e)
|
|
shutil.move(m, os.path.join(cfg["processed"], os.path.basename(m)))
|
|
if once:
|
|
break
|
|
time.sleep(int(cfg.get("poll_seconds", 30)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|