302 lines
9.1 KiB
Python
302 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Windows-side CODESYS agent with async job processing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import queue
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
from fastapi import FastAPI, Header, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
API_TOKEN = os.environ.get("CODESYS_AGENT_TOKEN", "")
|
|
SCRIPTS_DIR = Path(
|
|
os.environ.get("CODESYS_AGENT_SCRIPTS_DIR", r"C:\codesys-agent\scripts")
|
|
).resolve()
|
|
RUNTIME_DIR = Path(
|
|
os.environ.get("CODESYS_AGENT_RUNTIME_DIR", r"C:\codesys-agent\runtime")
|
|
).resolve()
|
|
PYTHON_BIN = os.environ.get("CODESYS_AGENT_PYTHON_BIN", "python")
|
|
DEFAULT_TIMEOUT_SEC = int(os.environ.get("CODESYS_AGENT_TIMEOUT_SEC", "180"))
|
|
ALLOW_DOWNLOAD = os.environ.get("CODESYS_AGENT_ALLOW_DOWNLOAD", "false").lower() == "true"
|
|
CODESYS_EXE_PATH = os.environ.get("CODESYS_EXE_PATH", "")
|
|
CODESYS_PROFILE = os.environ.get("CODESYS_PROFILE", "")
|
|
CODESYS_NO_UI = os.environ.get("CODESYS_NO_UI", "true").lower() == "true"
|
|
CODESYS_SCRIPT_TIMEOUT_SEC = int(os.environ.get("CODESYS_SCRIPT_TIMEOUT_SEC", "600"))
|
|
|
|
|
|
@dataclass
|
|
class Job:
|
|
id: str
|
|
action: str
|
|
payload: Dict[str, Any]
|
|
dry_run: bool = False
|
|
status: str = "queued"
|
|
created_at: float = field(default_factory=time.time)
|
|
started_at: Optional[float] = None
|
|
finished_at: Optional[float] = None
|
|
result: Optional[Dict[str, Any]] = None
|
|
error: Optional[str] = None
|
|
|
|
|
|
class SubmitJobRequest(BaseModel):
|
|
action: str = Field(..., description="Action name")
|
|
payload: Dict[str, Any] = Field(default_factory=dict)
|
|
dry_run: bool = False
|
|
|
|
|
|
class JobStore:
|
|
def __init__(self) -> None:
|
|
self._jobs: Dict[str, Job] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def create(self, action: str, payload: Dict[str, Any], dry_run: bool) -> Job:
|
|
job = Job(
|
|
id=str(uuid.uuid4()),
|
|
action=action,
|
|
payload=payload,
|
|
dry_run=dry_run,
|
|
)
|
|
with self._lock:
|
|
self._jobs[job.id] = job
|
|
return job
|
|
|
|
def get(self, job_id: str) -> Job:
|
|
with self._lock:
|
|
job = self._jobs.get(job_id)
|
|
if not job:
|
|
raise KeyError(job_id)
|
|
return job
|
|
|
|
def update(self, job: Job) -> None:
|
|
with self._lock:
|
|
self._jobs[job.id] = job
|
|
|
|
|
|
app = FastAPI(title="CODESYS VM Agent", version="0.2.0")
|
|
store = JobStore()
|
|
job_queue: "queue.Queue[str]" = queue.Queue()
|
|
|
|
|
|
def _check_auth(token: Optional[str]) -> None:
|
|
if not API_TOKEN:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="CODESYS_AGENT_TOKEN is not configured.",
|
|
)
|
|
if token != API_TOKEN:
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
|
|
def _action_to_script(action: str) -> str:
|
|
mapping = {
|
|
"create_project": "create_project.py",
|
|
"read_project_inventory": "read_project_inventory.py",
|
|
"read_device_io": "read_device_io.py",
|
|
"write_pou": "write_pou.py",
|
|
"write_io_mapping": "write_io_mapping.py",
|
|
"manage_device": "manage_device.py",
|
|
"sync_project": "sync_project.py",
|
|
"stage_existing_project": "stage_existing_project.py",
|
|
"build_project": "build_project.py",
|
|
"download_to_device": "download_to_device.py",
|
|
}
|
|
script = mapping.get(action)
|
|
if not script:
|
|
raise RuntimeError(f"Unsupported action: {action}")
|
|
if action == "download_to_device" and not ALLOW_DOWNLOAD:
|
|
raise RuntimeError("download_to_device is disabled by policy.")
|
|
return script
|
|
|
|
|
|
def _run_script_python(script_name: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
script_path = (SCRIPTS_DIR / script_name).resolve()
|
|
if not script_path.is_file():
|
|
raise RuntimeError(f"Script not found: {script_path}")
|
|
|
|
cmd = [PYTHON_BIN, str(script_path), json.dumps(payload)]
|
|
proc = subprocess.run(
|
|
cmd,
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=DEFAULT_TIMEOUT_SEC,
|
|
check=False,
|
|
)
|
|
return {
|
|
"command": cmd,
|
|
"exit_code": proc.returncode,
|
|
"stdout": proc.stdout,
|
|
"stderr": proc.stderr,
|
|
}
|
|
|
|
|
|
def _run_script_codesys(script_name: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
if not CODESYS_EXE_PATH or not CODESYS_PROFILE:
|
|
raise RuntimeError(
|
|
"CODESYS_EXE_PATH and CODESYS_PROFILE must be configured for ScriptEngine actions."
|
|
)
|
|
script_path = (SCRIPTS_DIR / script_name).resolve()
|
|
if not script_path.is_file():
|
|
raise RuntimeError(f"Script not found: {script_path}")
|
|
|
|
RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
|
|
payload_fd, payload_name = tempfile.mkstemp(
|
|
prefix="codesys_payload_",
|
|
suffix=".json",
|
|
dir=str(RUNTIME_DIR),
|
|
)
|
|
result_fd, result_name = tempfile.mkstemp(
|
|
prefix="codesys_result_",
|
|
suffix=".json",
|
|
dir=str(RUNTIME_DIR),
|
|
)
|
|
os.close(payload_fd)
|
|
os.close(result_fd)
|
|
payload_file = Path(payload_name)
|
|
result_file = Path(result_name)
|
|
payload_file.write_text(json.dumps(payload), encoding="utf-8")
|
|
|
|
script_args = f'--scriptargs:"{payload_file}"'
|
|
cmd = (
|
|
f'"{CODESYS_EXE_PATH}" '
|
|
f'--profile="{CODESYS_PROFILE}" '
|
|
f'--runscript="{script_path}" '
|
|
f"{script_args}"
|
|
)
|
|
if CODESYS_NO_UI:
|
|
cmd += " --noUI"
|
|
|
|
proc_env = os.environ.copy()
|
|
proc_env["CODESYS_SCRIPT_RESULT_PATH"] = str(result_file)
|
|
|
|
proc = subprocess.run(
|
|
cmd,
|
|
shell=True,
|
|
env=proc_env,
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=CODESYS_SCRIPT_TIMEOUT_SEC,
|
|
check=False,
|
|
)
|
|
parsed_result: Dict[str, Any] | None = None
|
|
if result_file.exists():
|
|
try:
|
|
parsed_result = json.loads(result_file.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError:
|
|
parsed_result = {"raw_result_text": result_file.read_text(encoding="utf-8", errors="ignore")}
|
|
|
|
return {
|
|
"command": cmd,
|
|
"exit_code": proc.returncode,
|
|
"stdout": proc.stdout,
|
|
"stderr": proc.stderr,
|
|
"script_result": parsed_result,
|
|
"payload_file": str(payload_file),
|
|
"result_file": str(result_file),
|
|
}
|
|
|
|
|
|
def _execute(job: Job) -> Dict[str, Any]:
|
|
if job.dry_run:
|
|
return {
|
|
"dry_run": True,
|
|
"action": job.action,
|
|
"payload": job.payload,
|
|
"message": "Dry-run accepted. No changes made.",
|
|
}
|
|
|
|
script_name = _action_to_script(job.action)
|
|
# Project operations use CODESYS ScriptEngine (no manual IDE opening required).
|
|
codesys_actions = {
|
|
"create_project",
|
|
"read_project_inventory",
|
|
"read_device_io",
|
|
"write_pou",
|
|
"write_io_mapping",
|
|
"manage_device",
|
|
"sync_project",
|
|
"build_project",
|
|
"download_to_device",
|
|
}
|
|
if job.action in codesys_actions:
|
|
result = _run_script_codesys(script_name, job.payload)
|
|
else:
|
|
result = _run_script_python(script_name, job.payload)
|
|
if result["exit_code"] != 0:
|
|
raise RuntimeError(
|
|
f"{job.action} failed with exit_code={result['exit_code']} stderr={result['stderr']}"
|
|
)
|
|
if result.get("script_result"):
|
|
return result["script_result"]
|
|
return result
|
|
|
|
|
|
def _worker() -> None:
|
|
while True:
|
|
job_id = job_queue.get()
|
|
try:
|
|
job = store.get(job_id)
|
|
job.status = "running"
|
|
job.started_at = time.time()
|
|
store.update(job)
|
|
|
|
result = _execute(job)
|
|
job.status = "succeeded"
|
|
job.result = result
|
|
job.finished_at = time.time()
|
|
store.update(job)
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
job = store.get(job_id)
|
|
job.status = "failed"
|
|
job.error = str(exc)
|
|
job.finished_at = time.time()
|
|
store.update(job)
|
|
finally:
|
|
job_queue.task_done()
|
|
|
|
|
|
threading.Thread(target=_worker, daemon=True).start()
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> Dict[str, Any]:
|
|
return {
|
|
"ok": True,
|
|
"service": "codesys-vm-agent",
|
|
"version": "0.2.0",
|
|
"scripts_dir": str(SCRIPTS_DIR),
|
|
"runtime_dir": str(RUNTIME_DIR),
|
|
"download_enabled": ALLOW_DOWNLOAD,
|
|
"codesys_exe_configured": bool(CODESYS_EXE_PATH),
|
|
"codesys_profile_configured": bool(CODESYS_PROFILE),
|
|
"codesys_no_ui": CODESYS_NO_UI,
|
|
}
|
|
|
|
|
|
@app.post("/jobs")
|
|
def submit_job(req: SubmitJobRequest, x_api_token: Optional[str] = Header(None)) -> Dict[str, Any]:
|
|
_check_auth(x_api_token)
|
|
job = store.create(req.action, req.payload, req.dry_run)
|
|
job_queue.put(job.id)
|
|
return {"job_id": job.id, "status": job.status}
|
|
|
|
|
|
@app.get("/jobs/{job_id}")
|
|
def get_job(job_id: str, x_api_token: Optional[str] = Header(None)) -> Dict[str, Any]:
|
|
_check_auth(x_api_token)
|
|
try:
|
|
job = store.get(job_id)
|
|
except KeyError as exc:
|
|
raise HTTPException(status_code=404, detail="Job not found") from exc
|
|
return asdict(job)
|