0
0
Fork 0

Initial release: MCP server for CODESYS with persistent UI instance

v0.3.0 returning-watcher architecture — background thread polls for commands
and marshals execution onto the CODESYS UI thread, keeping the IDE fully
responsive between operations. File-based IPC with atomic writes, async mutex
command serialization, headless fallback, and 35 passing tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Luke 2026-02-08 18:59:16 +10:00
commit e374519fe1
37 changed files with 7668 additions and 0 deletions

37
.github/workflows/ci.yml vendored Normal file
View file

@ -0,0 +1,37 @@
name: CI
on:
push:
branches: [main]
pull_request:
branches: [main]
jobs:
build-and-test:
runs-on: windows-latest
strategy:
matrix:
node-version: [18.x, 20.x, 22.x]
steps:
- uses: actions/checkout@v4
- name: Use Node.js ${{ matrix.node-version }}
uses: actions/setup-node@v4
with:
node-version: ${{ matrix.node-version }}
- name: Install dependencies
run: npm ci
- name: Type check
run: npx tsc --noEmit
- name: Build
run: npm run build
- name: Unit tests
run: npx vitest --run tests/unit/
- name: Integration tests (mock watcher)
run: npx vitest --run tests/integration/

25
.gitignore vendored Normal file
View file

@ -0,0 +1,25 @@
node_modules/
dist/
coverage/
*.tgz
*.log
.DS_Store
# Test artifacts
test_output/
projects/
project_examples/
# Python bytecode
__pycache__/
*.pyc
# CODESYS project binaries
*.project
*.precompilecache
*.opt
# Local config
.mcp.json
.claude/
CLAUDE.md

257
ARCHITECTURE.md Normal file
View file

@ -0,0 +1,257 @@
# Architecture
## Problem Statement
The original `@codesys/mcp-toolkit` spawns a new headless CODESYS process (`--noUI`) for every MCP tool call. This has two limitations:
1. **No UI visibility** — the user cannot see what the AI is doing to their project
2. **Project locking** — if the user opens CODESYS manually, the project file is locked and MCP tools fail
The desired workflow: a single CODESYS instance with its UI open, where MCP tool commands execute in the same process and changes appear in real-time.
## Architecture Overview
```
+-------------------------------------+
| MCP Client (Claude Code) |
+------------------+------------------+
| MCP Protocol (stdio)
+------------------v------------------+
| Node.js MCP Server |
| |
| bin.ts -> CLI entry point |
| server.ts -> MCP tools/resources |
| launcher.ts -> Process management |
| ipc.ts -> File-based IPC |
| headless.ts -> Fallback mode |
| script-manager.ts -> Templates |
+------------------+------------------+
| File-based IPC (persistent)
| OR spawn-per-command (headless)
+------------------v------------------+
| CODESYS.exe |
| watcher.py running inside via |
| --runscript (persistent mode) |
+-------------------------------------+
```
## IPC Protocol
### Directory Layout
Each session creates a unique directory under `os.tmpdir()`:
```
%TEMP%/codesys-mcp-persistent/<sessionId>/
commands/ Node.js writes here
<requestId>.py Script to execute
<requestId>.command.json Command trigger file
results/ Watcher writes here
<requestId>.result.json Execution result
watcher.py Interpolated watcher script
ready.signal Written by watcher on startup
terminate.signal Written by Node.js for shutdown
```
### Command File Format
`<requestId>.command.json`:
```json
{
"requestId": "uuid-v4",
"scriptPath": "/path/to/commands/<requestId>.py",
"timestamp": 1700000000000
}
```
### Result File Format
`<requestId>.result.json`:
```json
{
"requestId": "uuid-v4",
"success": true,
"output": "captured stdout from script execution",
"error": "",
"timestamp": 1700000000.123
}
```
### Write Ordering (Atomicity)
All files use atomic writes: write to `.tmp`, `fsync`, then `rename`.
Command submission order:
1. Write `<requestId>.py` (script content) -> fsync -> rename
2. Write `<requestId>.command.json.tmp` -> fsync -> rename to `.command.json`
The watcher triggers on `.command.json` appearance. Since the `.py` file is written and renamed first, it is guaranteed to exist when the watcher reads the command.
### Progressive Polling
Node.js polls for result files with exponential backoff:
- Initial interval: 100ms
- Doubles each poll: 100, 200, 400, 800, 1000ms
- Capped at 1000ms
- Default timeout: 60s (120s for compile)
## Watcher Script
The watcher (`src/scripts/watcher.py`) runs inside CODESYS via `--runscript` and provides the bridge between Node.js IPC and the CODESYS scripting API.
### Polling Loop
```python
while True:
if check_terminate():
break
command_files = scan_commands_dir()
if command_files:
process_command(command_files[0]) # one per iteration
time.sleep(0.05) # 50ms yield to UI thread
```
The 50ms sleep interval balances responsiveness (commands processed within ~50ms) against UI thread availability (CODESYS UI stays responsive).
### Script Execution via exec()
Each command script is executed with `exec(script_code, exec_globals)` where `exec_globals` is a fresh dictionary:
```python
exec_globals = {
'__builtins__': __builtins__,
'sys': sys,
'os': os,
'time': time,
'traceback': traceback,
'shutil': __import__('shutil'),
}
```
This provides:
- **Namespace isolation** — variables from script A are not visible to script B
- **CODESYS API access**`scriptengine` is available via `import scriptengine` because the watcher runs within the CODESYS scripting context (it's already in `sys.modules`)
- **Standard library access** — common modules pre-loaded in globals
### SystemExit Handling
CODESYS scripts use `sys.exit(0)` for success and `sys.exit(1)` for failure. The watcher catches `SystemExit` to prevent CODESYS from closing:
| Exit code | Mapping |
|-----------|---------|
| `None` or `0` | Success |
| Non-zero int | Failure |
| String | Failure (string is the error message) |
Output markers (`SCRIPT_SUCCESS` / `SCRIPT_ERROR`) take priority over exit codes when both are present.
### Output Capture
The `OutputCapture` class redirects `sys.stdout` and `sys.stderr` during script execution:
```python
class OutputCapture:
def __init__(self):
self._buffer = []
def write(self, s):
self._buffer.append(str(s))
def getvalue(self):
return ''.join(self._buffer)
```
Original stdout/stderr are saved and restored in a `try/finally` block, guaranteeing restoration even on unexpected exceptions. This class works across CPython and IronPython (CODESYS uses IronPython).
## Script Template System
Python scripts are stored as templates in `src/scripts/` with `{PLACEHOLDER}` tokens. The `ScriptManager` handles:
1. **Loading** — reads `.py` files from disk with caching
2. **Interpolation** — replaces `{KEY}` with escaped values
3. **Escaping** — backslashes doubled for Python string embedding (`C:\Users` -> `C:\\Users`)
4. **Triple-quote escaping**`"""` in values escaped to `\"\"\"` for Python triple-quoted strings
5. **Helper prepending** — shared functions (`ensure_project_open`, `find_object_by_path`) prepended before the main script
### Helper Scripts
Two helper scripts are prepended to most tool scripts:
- **`ensure_project_open.py`** — opens a project file if not already open, with retry logic (3 attempts, 2s delay)
- **`find_object_by_path.py`** — navigates the CODESYS project tree to find objects by path (e.g., `Application/MyPOU`)
## Lifecycle Management
### Launch Sequence
1. Validate CODESYS executable exists
2. Generate session UUID
3. Create IPC directory with `commands/` and `results/` subdirectories
4. Load `watcher.py` template, interpolate `{IPC_BASE_DIR}`
5. Write interpolated watcher to session directory
6. Spawn: `CODESYS.exe --profile="..." --runscript="watcher.py"` (detached, UI visible)
7. `process.unref()` so Node.js doesn't wait for CODESYS
8. Poll for `ready.signal` (max 60s, every 500ms)
9. Start health monitor (5s interval PID check)
### Shutdown Sequence
1. Write `terminate.signal`
2. Wait up to 5s for process exit (poll every 500ms)
3. If still alive: `SIGTERM`, wait 2s, then `SIGKILL`
4. Clean up IPC directory
### Health Monitoring
A `setInterval` runs every 5 seconds checking if the CODESYS process is still alive (`process.kill(pid, 0)`). On process death:
- State transitions to `error`
- `lastError` is set with a descriptive message
- Registered `onStateChange` callbacks are invoked
- Monitor stops itself
## Concurrency Model
### Async Mutex
The `IpcClient` uses an async mutex to serialize commands. Only one command can be in-flight at a time. This prevents:
- Race conditions in the CODESYS scripting API (not thread-safe)
- File system conflicts in the IPC directory
- Interleaved script output
When multiple tool calls arrive concurrently, they queue and execute sequentially.
### Watcher Single-Threaded Processing
The watcher processes one command per polling iteration. If multiple `.command.json` files exist, they're sorted alphabetically and processed in order.
## Headless Fallback
When persistent mode is unavailable, the `HeadlessExecutor` provides the same `ScriptExecutor` interface using spawn-per-command:
1. Write script to temp file
2. Spawn `CODESYS.exe --profile="..." --noUI --runscript="script.py"` with `windowsHide: true`
3. Capture stdout/stderr
4. Parse `SCRIPT_SUCCESS` / `SCRIPT_ERROR` markers
5. Return `IpcResult`
Fallback activates when:
- `--mode headless` is specified
- Persistent launch fails and `--fallback-headless` is enabled
- Server starts with `--no-auto-launch` before `launch_codesys` is called
## Differences from Original Toolkit
| Aspect | @codesys/mcp-toolkit | codesys-mcp-persistent |
|--------|---------------------|----------------------|
| CODESYS UI | Hidden (`--noUI`) | Visible (persistent) or hidden (headless) |
| Process lifetime | New process per command | Single long-running process |
| IPC mechanism | Spawn + stdout | File-based polling |
| Project locking | Blocks if user opens CODESYS | Shares the same instance |
| Real-time feedback | None | Changes visible in UI |
| Startup overhead | ~10-30s per command | ~10-30s once, then <100ms per command |
| Management tools | None | `launch_codesys`, `shutdown_codesys`, `get_codesys_status` |
## Security Considerations
- **Temp directory** — IPC files are created in the user's temp directory with default permissions. No sensitive data (credentials, keys) is written to IPC files.
- **Script injection** — tool parameters are escaped for Python string embedding (backslashes doubled, triple quotes escaped). The `exec()` context has access to the full CODESYS scripting API, which is the intended design.
- **Localhost only** — IPC is file-based with no network exposure. The MCP server communicates via stdio only.
- **Process isolation** — CODESYS is spawned as a detached process. The Node.js server can crash and restart without affecting CODESYS (though a new session would be created).

21
LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

207
README.md Normal file
View file

@ -0,0 +1,207 @@
# codesys-mcp-persistent
MCP server for CODESYS with a persistent UI instance and file-based IPC.
Unlike headless-only approaches that spawn a new CODESYS process per command, this server launches CODESYS **with its UI visible** and keeps it running. MCP tool calls are sent to the same instance via a file-based IPC watcher, so changes appear in real-time and the user can interact with the IDE alongside AI-driven automation.
## Features
- **Persistent mode** — CODESYS UI stays open; commands execute in the running instance
- **Headless fallback** — automatic fallback to `--noUI` spawn-per-command if persistent mode fails
- **File-based IPC** — proven approach using atomic file writes and a Python watcher script
- **Command serialization** — async mutex ensures one command at a time
- **Health monitoring** — detects CODESYS crashes and reports state
- **Drop-in replacement** — same MCP tool names and parameters as `@codesys/mcp-toolkit`
## Installation
```bash
npm install -g codesys-mcp-persistent
```
Or install from the repository:
```bash
git clone https://github.com/luke-harriman/Codesys-MCP.git
cd Codesys-MCP
npm install
npm run build
npm link
```
**Requirements:** Node.js 18+, Windows, CODESYS 3.5 SP19 or SP21 installed.
## Quick Start
Add to your `.mcp.json` (Claude Code configuration):
```json
{
"mcpServers": {
"codesys": {
"command": "codesys-mcp-persistent",
"args": [
"--codesys-path", "C:\\Program Files\\CODESYS 3.5.21.0\\CODESYS\\Common\\CODESYS.exe",
"--codesys-profile", "CODESYS V3.5 SP21 Patch 3",
"--mode", "persistent"
]
}
}
}
```
Or run directly:
```bash
codesys-mcp-persistent \
--codesys-path "C:\Program Files\CODESYS 3.5.21.0\CODESYS\Common\CODESYS.exe" \
--codesys-profile "CODESYS V3.5 SP21 Patch 3"
```
## CLI Reference
| Flag | Description | Default |
|------|-------------|---------|
| `-p, --codesys-path <path>` | Path to CODESYS executable | `$CODESYS_PATH` or auto-detected |
| `-f, --codesys-profile <name>` | CODESYS profile name | `$CODESYS_PROFILE` or `CODESYS V3.5 SP21` |
| `-w, --workspace <dir>` | Workspace directory for relative paths | Current directory |
| `-m, --mode <mode>` | `persistent` (UI) or `headless` (--noUI) | `persistent` |
| `--no-auto-launch` | Don't launch CODESYS on startup | Auto-launch enabled |
| `--fallback-headless` | Fall back to headless if persistent fails | `true` |
| `--keep-alive` | Keep CODESYS running after server stops | `false` |
| `--timeout <ms>` | Default command timeout | `60000` |
| `--detect` | List installed CODESYS versions and exit | — |
| `--verbose` | Enable verbose logging | — |
| `--debug` | Enable debug logging | — |
| `-V, --version` | Show version number | — |
| `-h, --help` | Show help | — |
Environment variables `CODESYS_PATH` and `CODESYS_PROFILE` are used as defaults when the corresponding flags are not provided.
## MCP Tools
### Management Tools
| Tool | Description |
|------|-------------|
| `launch_codesys` | Manually launch CODESYS (use with `--no-auto-launch`) |
| `shutdown_codesys` | Shut down the persistent CODESYS instance |
| `get_codesys_status` | Get current state, PID, execution mode |
### Project Tools
| Tool | Description |
|------|-------------|
| `open_project` | Open an existing CODESYS project file |
| `create_project` | Create a new project from the standard template |
| `save_project` | Save the currently open project |
| `compile_project` | Build the primary application (120s timeout) |
### POU Tools
| Tool | Description |
|------|-------------|
| `create_pou` | Create a Program, Function Block, or Function |
| `set_pou_code` | Set declaration and/or implementation code |
| `create_property` | Create a property within a Function Block |
| `create_method` | Create a method within a Function Block |
## MCP Resources
| Resource URI | Description |
|--------------|-------------|
| `codesys://project/status` | CODESYS scripting status and open project info |
| `codesys://project/{path}/structure` | Project tree structure |
| `codesys://project/{path}/pou/{pou}/code` | POU declaration and implementation code |
## Execution Modes
### Persistent Mode (default)
1. Server launches `CODESYS.exe` with `--runscript=watcher.py` (no `--noUI`)
2. CODESYS UI opens — user can see and interact with the IDE
3. The watcher script starts a .NET background thread that polls a `commands/` directory, then **returns control to CODESYS** so the UI stays fully responsive
4. When a tool is called, the server writes a `.py` script + `.command.json` to `commands/`
5. The background thread detects the command and marshals execution onto the CODESYS UI thread via `system.execute_on_primary_thread()`
6. Results are written atomically to `results/`
7. Changes made by tools appear in the CODESYS UI in real-time
8. The UI remains interactive between commands — only briefly paused during synchronous API calls (compile, open)
### Headless Mode
Falls back to the original approach: each tool call spawns a new CODESYS process with `--noUI`, runs the script, and exits. No UI is shown. Used when:
- `--mode headless` is specified
- Persistent mode fails to launch and `--fallback-headless` is enabled
- CODESYS is launched with `--no-auto-launch` and `launch_codesys` hasn't been called yet
## Detect Installed Versions
```bash
codesys-mcp-persistent --detect
```
Scans `Program Files` and `Program Files (x86)` for CODESYS installations.
## Troubleshooting
**CODESYS not found**
Verify the path with `--detect`. The executable is typically at:
`C:\Program Files\CODESYS 3.5.XX.X\CODESYS\Common\CODESYS.exe`
**Project file locked**
Another CODESYS instance may have the project open. Close it first or use persistent mode so there's only one instance.
**Watcher timeout (persistent mode)**
If the watcher doesn't signal ready within 60 seconds, check:
- CODESYS path and profile are correct
- No modal dialogs are blocking CODESYS startup
- Try `--verbose` for detailed logging
**UI briefly pauses during commands (persistent mode)**
The v0.3.0 watcher uses a background thread that marshals work onto the UI thread, so the UI stays responsive between commands. During synchronous CODESYS API calls (compile, project open), the UI may briefly pause — this is expected and normal. If a command hangs, check the CODESYS messages window for modal dialogs or errors.
**Command timeout**
Default is 60s (120s for compile). Increase with `--timeout <ms>`. Check CODESYS messages window for errors.
## Development
```bash
# Install dependencies
npm install
# Build (compiles TypeScript + copies Python scripts)
npm run build
# Run all tests
npm test
# Type check only
npm run typecheck
# Run tests in watch mode
npm run test:watch
```
### Project Structure
```
src/
bin.ts CLI entry point
server.ts MCP tool/resource registration
launcher.ts CODESYS process management
ipc.ts File-based IPC transport
headless.ts Headless fallback executor
script-manager.ts Python template loading + interpolation
types.ts Shared TypeScript types
logger.ts Structured stderr logging
scripts/ Python scripts (watcher + 13 tool scripts)
tests/
unit/ Unit tests (IPC, script manager, launcher)
integration/ Integration tests (script pipeline, manual CODESYS tests)
mock_watcher.py Standalone watcher for testing without CODESYS
```
## License
MIT

3451
package-lock.json generated Normal file

File diff suppressed because it is too large Load diff

52
package.json Normal file
View file

@ -0,0 +1,52 @@
{
"name": "codesys-mcp-persistent",
"version": "0.3.0",
"description": "MCP server for CODESYS with persistent UI instance and file-based IPC",
"main": "dist/server.js",
"bin": {
"codesys-mcp-persistent": "dist/bin.js"
},
"files": [
"dist/",
"README.md",
"LICENSE"
],
"scripts": {
"build": "tsc && node -e \"require('fs').cpSync('src/scripts','dist/scripts',{recursive:true})\"",
"test": "vitest --run",
"test:watch": "vitest",
"typecheck": "tsc --noEmit",
"prepublishOnly": "npm run build && npm test"
},
"keywords": [
"codesys",
"mcp",
"plc",
"automation",
"persistent",
"ui"
],
"author": "luke-harriman",
"repository": {
"type": "git",
"url": "https://github.com/luke-harriman/Codesys-MCP.git"
},
"homepage": "https://github.com/luke-harriman/Codesys-MCP#readme",
"license": "MIT",
"engines": {
"node": ">=18"
},
"dependencies": {
"@modelcontextprotocol/sdk": "^1.26.0",
"commander": "^11.1.0",
"uuid": "^9.0.0",
"zod": "^3.24.3"
},
"devDependencies": {
"@types/node": "^20.14.1",
"@types/uuid": "^9.0.0",
"tsx": "^4.21.0",
"typescript": "^5.5.3",
"vitest": "^1.6.0"
}
}

111
src/bin.ts Normal file
View file

@ -0,0 +1,111 @@
#!/usr/bin/env node
/**
* CLI entry point for codesys-mcp-persistent.
*/
import { program } from 'commander';
import { startMcpServer } from './server';
import { ServerConfig, ExecutionMode } from './types';
let version = '0.1.0';
try {
// eslint-disable-next-line @typescript-eslint/no-var-requires
const pkg = require('../package.json');
version = pkg.version;
} catch {
// ignore
}
program
.name('codesys-mcp-persistent')
.description('MCP server for CODESYS with persistent UI instance')
.version(version)
.option(
'-p, --codesys-path <path>',
'Path to CODESYS executable',
process.env.CODESYS_PATH || 'C:\\Program Files\\CODESYS 3.5.21.0\\CODESYS\\Common\\CODESYS.exe'
)
.option(
'-f, --codesys-profile <profile>',
'CODESYS profile name',
process.env.CODESYS_PROFILE || 'CODESYS V3.5 SP21'
)
.option(
'-w, --workspace <dir>',
'Workspace directory for relative project paths',
process.cwd()
)
.option(
'-m, --mode <mode>',
'Execution mode: persistent (UI) or headless (--noUI)',
'persistent'
)
.option('--no-auto-launch', 'Do not auto-launch CODESYS on startup')
.option('--fallback-headless', 'Fall back to headless if persistent fails', true)
.option('--keep-alive', 'Keep CODESYS running after server stops', false)
.option('--timeout <ms>', 'Default command timeout in ms', '60000')
.option('--verbose', 'Enable verbose logging')
.option('--debug', 'Enable debug logging (more verbose)')
.option('--detect', 'Detect installed CODESYS versions and exit')
.parse(process.argv);
const opts = program.opts();
// Handle --detect flag
if (opts.detect) {
import('fs').then((fs) => {
import('path').then((pathMod) => {
const dirs = [
'C:\\Program Files',
'C:\\Program Files (x86)',
];
process.stderr.write('Scanning for CODESYS installations...\n\n');
let found = 0;
for (const base of dirs) {
try {
const entries = fs.readdirSync(base);
for (const entry of entries) {
if (entry.toLowerCase().includes('codesys')) {
const commonExe = pathMod.join(base, entry, 'CODESYS', 'Common', 'CODESYS.exe');
const exists = fs.existsSync(commonExe);
process.stderr.write(` ${exists ? '[OK]' : '[--]'} ${pathMod.join(base, entry)}\n`);
if (exists) {
process.stderr.write(` Exe: ${commonExe}\n`);
found++;
}
}
}
} catch {
// dir doesn't exist
}
}
process.stderr.write(`\nFound ${found} CODESYS installation(s).\n`);
process.exit(0);
});
});
} else {
// Build server config
const config: ServerConfig = {
codesysPath: opts.codesysPath.trim(),
profileName: opts.codesysProfile.trim(),
workspaceDir: opts.workspace.trim(),
autoLaunch: opts.autoLaunch !== false,
keepAlive: opts.keepAlive || false,
timeoutMs: parseInt(opts.timeout, 10) || 60000,
fallbackHeadless: opts.fallbackHeadless !== false,
verbose: opts.verbose || false,
debug: opts.debug || false,
mode: (opts.mode === 'headless' ? 'headless' : 'persistent') as ExecutionMode,
};
process.stderr.write(`Starting CODESYS MCP Server v${version}\n`);
process.stderr.write(` CODESYS Path: ${config.codesysPath}\n`);
process.stderr.write(` Profile: ${config.profileName}\n`);
process.stderr.write(` Mode: ${config.mode}\n`);
process.stderr.write(` Auto-launch: ${config.autoLaunch}\n`);
startMcpServer(config).catch((err) => {
process.stderr.write(`FATAL: ${err.message}\n`);
process.exit(1);
});
}

167
src/headless.ts Normal file
View file

@ -0,0 +1,167 @@
/**
* Headless fallback mode spawns CODESYS with --noUI per command.
* Direct port of the original codesys_interop.js approach.
*/
import * as fs from 'fs';
import * as path from 'path';
import * as os from 'os';
import { spawn } from 'child_process';
import { LauncherConfig, IpcResult, ScriptExecutor } from './types';
import { headlessLog } from './logger';
const SCRIPT_SUCCESS_MARKER = 'SCRIPT_SUCCESS';
const SCRIPT_ERROR_MARKER = 'SCRIPT_ERROR';
const DEFAULT_TIMEOUT_MS = 60_000;
export class HeadlessExecutor implements ScriptExecutor {
private config: LauncherConfig;
constructor(config: LauncherConfig) {
this.config = config;
// Validate CODESYS exe exists
if (!fs.existsSync(config.codesysPath)) {
throw new Error(
`CODESYS executable not found: ${config.codesysPath}`
);
}
}
/** Execute a script by spawning CODESYS with --noUI */
async executeScript(
scriptContent: string,
timeoutMs?: number
): Promise<IpcResult> {
const timeout = timeoutMs ?? DEFAULT_TIMEOUT_MS;
const tempDir = os.tmpdir();
const tempFileName = `codesys_script_${Date.now()}_${Math.random()
.toString(36)
.substring(2, 9)}.py`;
const tempFilePath = path.join(tempDir, tempFileName);
const requestId = `headless_${Date.now()}`;
const codesysDir = path.dirname(this.config.codesysPath);
try {
// Write script to temp file
const normalized = scriptContent.replace(/\r\n/g, '\n');
fs.writeFileSync(tempFilePath, normalized, 'latin1');
headlessLog.debug(`Temp script written: ${tempFilePath}`);
// Build command
const quotedExe = `"${this.config.codesysPath}"`;
const profileArg = `--profile="${this.config.profileName}"`;
const scriptArg = `--runscript="${tempFilePath}"`;
const fullCommand = `${quotedExe} ${profileArg} --noUI ${scriptArg}`;
headlessLog.debug(`Spawning: ${fullCommand}`);
// Spawn and collect output
const result = await new Promise<{
code: number | null;
stdout: string;
stderr: string;
error?: Error;
}>((resolve) => {
let stdoutData = '';
let stderrData = '';
const controller = new AbortController();
// Prepend CODESYS dir to PATH
const spawnEnv = { ...process.env };
const originalPath = spawnEnv.PATH || '';
spawnEnv.PATH = `${codesysDir};${originalPath}`;
const child = spawn(fullCommand, [], {
windowsHide: true,
signal: controller.signal,
cwd: codesysDir,
env: spawnEnv,
shell: true,
});
const timeoutId = setTimeout(() => {
headlessLog.warn('Process timeout reached');
controller.abort();
}, timeout);
child.stdout.on('data', (data: Buffer) => {
stdoutData += data.toString();
});
child.stderr.on('data', (data: Buffer) => {
stderrData += data.toString();
});
child.on('error', (err: Error) => {
clearTimeout(timeoutId);
resolve({ code: 1, stdout: stdoutData, stderr: stderrData, error: err });
});
child.on('close', (code: number | null) => {
clearTimeout(timeoutId);
resolve({ code, stdout: stdoutData, stderr: stderrData });
});
controller.signal.addEventListener(
'abort',
() => {
if (!child.killed) {
child.kill('SIGTERM');
setTimeout(() => {
if (!child.killed) child.kill('SIGKILL');
}, 2_000);
}
resolve({
code: null,
stdout: stdoutData,
stderr: stderrData + '\nTIMEOUT: Process aborted.',
});
},
{ once: true }
);
});
// Determine success
let success = false;
const combinedOutput = result.stdout;
const stderrOutput = result.stderr;
if (result.error) {
success = false;
} else if (
combinedOutput.includes(SCRIPT_SUCCESS_MARKER) ||
stderrOutput.includes(SCRIPT_SUCCESS_MARKER)
) {
success = true;
} else if (
combinedOutput.includes(SCRIPT_ERROR_MARKER) ||
stderrOutput.includes(SCRIPT_ERROR_MARKER)
) {
success = false;
} else {
success = result.code === 0;
}
const finalOutput = success
? combinedOutput
: `${stderrOutput}\n${combinedOutput}`.trim();
return {
requestId,
success,
output: finalOutput,
error: success ? '' : (result.error?.message || stderrOutput || `Exit code ${result.code}`),
timestamp: Date.now(),
};
} finally {
// Clean up temp file
try {
fs.unlinkSync(tempFilePath);
} catch {
headlessLog.debug(`Failed to delete temp file: ${tempFilePath}`);
}
}
}
}

239
src/ipc.ts Normal file
View file

@ -0,0 +1,239 @@
/**
* File-based IPC transport layer.
* Writes command files, polls for result files, with command serialization.
*/
import * as fs from 'fs';
import * as path from 'path';
import { v4 as uuidv4 } from 'uuid';
import { IpcConfig, IpcResult, IpcCommand, RequestId } from './types';
import { ipcLog } from './logger';
/** Default IPC configuration */
export const DEFAULT_IPC_CONFIG: Omit<IpcConfig, 'baseDir'> = {
commandTimeoutMs: 60_000,
pollIntervalMs: 100,
maxPollIntervalMs: 1_000,
deleteResultAfterRead: true,
};
/**
* Async mutex for serializing commands.
* Ensures only one command is in-flight at a time.
*/
class AsyncMutex {
private _queue: Array<() => void> = [];
private _locked = false;
async acquire(): Promise<void> {
if (!this._locked) {
this._locked = true;
return;
}
return new Promise<void>((resolve) => {
this._queue.push(resolve);
});
}
release(): void {
if (this._queue.length > 0) {
const next = this._queue.shift()!;
next();
} else {
this._locked = false;
}
}
}
/**
* Atomic file write: write to .tmp then rename.
* Uses fsync to ensure data is flushed to disk before rename.
*/
async function atomicWrite(filePath: string, content: string): Promise<void> {
const tmpPath = filePath + '.tmp';
const fd = fs.openSync(tmpPath, 'w');
try {
fs.writeSync(fd, content, undefined, 'utf-8');
fs.fsyncSync(fd);
} finally {
fs.closeSync(fd);
}
fs.renameSync(tmpPath, filePath);
}
export class IpcClient {
private config: IpcConfig;
private mutex = new AsyncMutex();
private commandsDir: string;
private resultsDir: string;
constructor(config: IpcConfig) {
this.config = config;
this.commandsDir = path.join(config.baseDir, 'commands');
this.resultsDir = path.join(config.baseDir, 'results');
}
/** Create commands/ and results/ directories */
async ensureDirectories(): Promise<void> {
fs.mkdirSync(this.commandsDir, { recursive: true });
fs.mkdirSync(this.resultsDir, { recursive: true });
ipcLog.debug(`IPC directories created at ${this.config.baseDir}`);
}
/** Check if the watcher has written ready.signal */
async isReady(): Promise<boolean> {
const signalPath = path.join(this.config.baseDir, 'ready.signal');
return fs.existsSync(signalPath);
}
/** Write terminate.signal to request watcher shutdown */
async sendTerminate(): Promise<void> {
const signalPath = path.join(this.config.baseDir, 'terminate.signal');
await atomicWrite(signalPath, JSON.stringify({ timestamp: Date.now() }));
ipcLog.info('Terminate signal sent');
}
/**
* Send a command to the watcher and wait for result.
* Serialized via async mutex only one command in-flight at a time.
*/
async sendCommand(scriptContent: string, timeoutMs?: number): Promise<IpcResult> {
await this.mutex.acquire();
try {
return await this._sendCommandInternal(scriptContent, timeoutMs);
} finally {
this.mutex.release();
}
}
private async _sendCommandInternal(
scriptContent: string,
timeoutMs?: number
): Promise<IpcResult> {
const requestId: RequestId = uuidv4();
const timeout = timeoutMs ?? this.config.commandTimeoutMs;
const scriptFileName = `${requestId}.py`;
const commandFileName = `${requestId}.command.json`;
const resultFileName = `${requestId}.result.json`;
const scriptPath = path.join(this.commandsDir, scriptFileName);
const commandPath = path.join(this.commandsDir, commandFileName);
const resultPath = path.join(this.resultsDir, resultFileName);
ipcLog.debug(`Sending command ${requestId}`);
// Step 1: Write .py script file with fsync
await atomicWrite(scriptPath, scriptContent);
// Step 2: Write .command.json (triggers watcher)
const command: IpcCommand = {
requestId,
scriptPath: scriptPath,
timestamp: Date.now(),
};
await atomicWrite(commandPath, JSON.stringify(command));
ipcLog.debug(`Command ${requestId} written, polling for result...`);
// Step 3: Poll for result with progressive backoff
const startTime = Date.now();
let pollInterval = this.config.pollIntervalMs;
while (Date.now() - startTime < timeout) {
if (fs.existsSync(resultPath)) {
// Try to read result with retry for partial writes
const result = await this._readResultWithRetry(resultPath, requestId);
if (result) {
// Clean up result file if configured
if (this.config.deleteResultAfterRead) {
try {
fs.unlinkSync(resultPath);
} catch {
// Ignore cleanup errors
}
}
ipcLog.debug(
`Command ${requestId} completed: success=${result.success}`
);
return result;
}
}
// Progressive backoff: double interval, cap at max
await this._sleep(pollInterval);
pollInterval = Math.min(pollInterval * 2, this.config.maxPollIntervalMs);
}
// Timeout — clean up command files
this._cleanupCommandFiles(requestId);
throw new Error(
`Command ${requestId} timed out after ${timeout}ms waiting for result`
);
}
/**
* Read result file with retry for corrupted/partial JSON.
* Up to 3 attempts with 100ms delay between.
*/
private async _readResultWithRetry(
resultPath: string,
requestId: string
): Promise<IpcResult | null> {
for (let attempt = 0; attempt < 3; attempt++) {
try {
const content = fs.readFileSync(resultPath, 'utf-8');
const result: IpcResult = JSON.parse(content);
if (result.requestId === requestId) {
return result;
}
ipcLog.warn(
`Result file requestId mismatch: expected ${requestId}, got ${result.requestId}`
);
return null;
} catch (err) {
if (attempt < 2) {
ipcLog.debug(
`Result read attempt ${attempt + 1} failed, retrying in 100ms...`
);
await this._sleep(100);
} else {
ipcLog.warn(
`Failed to read result after 3 attempts for ${requestId}: ${err}`
);
return null;
}
}
}
return null;
}
/** Clean up command files for a given request */
private _cleanupCommandFiles(requestId: string): void {
const scriptPath = path.join(this.commandsDir, `${requestId}.py`);
const commandPath = path.join(
this.commandsDir,
`${requestId}.command.json`
);
try {
if (fs.existsSync(scriptPath)) fs.unlinkSync(scriptPath);
} catch { /* ignore */ }
try {
if (fs.existsSync(commandPath)) fs.unlinkSync(commandPath);
} catch { /* ignore */ }
}
/** Remove the entire session directory */
async cleanup(): Promise<void> {
try {
fs.rmSync(this.config.baseDir, { recursive: true, force: true });
ipcLog.info(`Session directory cleaned up: ${this.config.baseDir}`);
} catch (err) {
ipcLog.warn(`Failed to clean up session directory: ${err}`);
}
}
private _sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
}

300
src/launcher.ts Normal file
View file

@ -0,0 +1,300 @@
/**
* CODESYS launcher spawns CODESYS with UI and watcher script,
* tracks process lifecycle, delegates to IPC for command execution.
*/
import * as fs from 'fs';
import * as path from 'path';
import * as os from 'os';
import { spawn, ChildProcess } from 'child_process';
import { v4 as uuidv4 } from 'uuid';
import { LauncherConfig, LauncherStatus, CodesysState, IpcResult, ScriptExecutor } from './types';
import { IpcClient, DEFAULT_IPC_CONFIG } from './ipc';
import { ScriptManager } from './script-manager';
import { launcherLog } from './logger';
const SESSION_DIR_PREFIX = 'codesys-mcp-persistent';
const READY_TIMEOUT_MS = 60_000;
const READY_POLL_MS = 500;
const SHUTDOWN_WAIT_MS = 5_000;
const HEALTH_CHECK_INTERVAL_MS = 5_000;
export class CodesysLauncher implements ScriptExecutor {
private config: LauncherConfig;
private state: CodesysState = 'stopped';
private pid: number | null = null;
private sessionId: string | null = null;
private ipcDir: string | null = null;
private ipcClient: IpcClient | null = null;
private process: ChildProcess | null = null;
private startedAt: number | null = null;
private lastError: string | null = null;
private healthInterval: ReturnType<typeof setInterval> | null = null;
private stateChangeCallbacks: Array<(state: CodesysState) => void> = [];
constructor(config: LauncherConfig) {
this.config = config;
}
/** Launch CODESYS with UI and watcher script */
async launch(): Promise<void> {
if (this.state === 'ready' || this.state === 'launching') {
launcherLog.warn(`Cannot launch: state is ${this.state}`);
return;
}
// Validate CODESYS exe exists
if (!fs.existsSync(this.config.codesysPath)) {
const err = `CODESYS executable not found: ${this.config.codesysPath}`;
this.setState('error');
this.lastError = err;
throw new Error(err);
}
this.setState('launching');
this.sessionId = uuidv4();
this.ipcDir = path.join(os.tmpdir(), SESSION_DIR_PREFIX, this.sessionId);
launcherLog.info(`Session ${this.sessionId} — IPC dir: ${this.ipcDir}`);
// Create IPC client and directories
this.ipcClient = new IpcClient({
baseDir: this.ipcDir,
...DEFAULT_IPC_CONFIG,
});
await this.ipcClient.ensureDirectories();
// Prepare watcher script with interpolated IPC path
const scriptManager = new ScriptManager();
const watcherTemplate = scriptManager.loadTemplate('watcher');
const ipcPathEscaped = this.ipcDir.replace(/\\/g, '\\\\');
const watcherContent = scriptManager.interpolate(watcherTemplate, {
IPC_BASE_DIR: ipcPathEscaped,
});
// Write interpolated watcher to IPC directory
const watcherPath = path.join(this.ipcDir, 'watcher.py');
fs.writeFileSync(watcherPath, watcherContent, 'utf-8');
// Build CODESYS command
const quotedExe = `"${this.config.codesysPath}"`;
const profileArg = `--profile="${this.config.profileName}"`;
const scriptArg = `--runscript="${watcherPath}"`;
const fullCommand = `${quotedExe} ${profileArg} ${scriptArg}`;
launcherLog.info(`Spawning: ${fullCommand}`);
// Spawn CODESYS detached with UI visible
const codesysDir = path.dirname(this.config.codesysPath);
this.process = spawn(fullCommand, [], {
detached: true,
shell: true,
windowsHide: false,
stdio: 'ignore',
cwd: codesysDir,
});
this.pid = this.process.pid ?? null;
this.process.unref();
launcherLog.info(`CODESYS spawned with PID ${this.pid}`);
// Handle process exit
this.process.on('exit', (code) => {
launcherLog.warn(`CODESYS process exited with code ${code}`);
if (this.state !== 'stopping') {
this.lastError = `CODESYS exited unexpectedly (code ${code})`;
this.setState('error');
}
this.pid = null;
this.process = null;
});
// Poll for ready.signal
const readyStart = Date.now();
while (Date.now() - readyStart < READY_TIMEOUT_MS) {
if (await this.ipcClient.isReady()) {
this.setState('ready');
this.startedAt = Date.now();
launcherLog.info('CODESYS watcher is ready');
this.startHealthMonitor();
return;
}
await this.sleep(READY_POLL_MS);
}
// Timeout — watcher never signaled ready
this.lastError = `Watcher did not signal ready within ${READY_TIMEOUT_MS}ms`;
this.setState('error');
throw new Error(this.lastError);
}
/** Graceful shutdown */
async shutdown(): Promise<void> {
if (this.state === 'stopped' || this.state === 'stopping') return;
this.setState('stopping');
this.stopHealthMonitor();
// Try to close projects and quit CODESYS gracefully via script
if (this.ipcClient && this.state !== 'error') {
try {
launcherLog.info('Sending quit script to close projects and exit CODESYS...');
await this.ipcClient.sendCommand(`
import sys
try:
import scriptengine as se
# Close all open projects without saving (save should be done before shutdown)
for p in list(se.projects):
try:
p.close()
except:
pass
print("Projects closed")
except:
pass
# Request CODESYS to quit
try:
import scriptengine as se
se.system.exit()
except:
pass
print("SCRIPT_SUCCESS")
sys.exit(0)
`, 10_000);
} catch {
launcherLog.debug('Quit script timed out or failed (expected if CODESYS exits)');
}
}
// Send terminate signal to watcher
if (this.ipcClient) {
try {
await this.ipcClient.sendTerminate();
} catch {
launcherLog.warn('Failed to send terminate signal');
}
}
// Wait for process exit
if (this.pid !== null) {
const waitStart = Date.now();
while (Date.now() - waitStart < SHUTDOWN_WAIT_MS) {
if (!this.isRunning()) break;
await this.sleep(500);
}
// Force kill if still alive
if (this.isRunning() && this.pid !== null) {
launcherLog.warn('Force-killing CODESYS process');
try {
// On Windows, use taskkill for reliable process termination
if (process.platform === 'win32') {
const { execSync } = require('child_process');
try {
// First try graceful close (WM_CLOSE)
execSync(`taskkill /PID ${this.pid}`, { timeout: 5000, stdio: 'ignore' });
await this.sleep(3_000);
} catch { /* ignore */ }
if (this.isRunning()) {
// Force kill
try {
execSync(`taskkill /F /PID ${this.pid}`, { timeout: 5000, stdio: 'ignore' });
} catch { /* ignore */ }
}
} else if (this.process) {
this.process.kill('SIGTERM');
await this.sleep(2_000);
if (this.isRunning() && this.process) {
this.process.kill('SIGKILL');
}
}
} catch {
launcherLog.warn('Failed to kill CODESYS process');
}
}
}
// Clean up IPC directory
if (this.ipcClient) {
await this.ipcClient.cleanup();
}
this.pid = null;
this.process = null;
this.ipcClient = null;
this.setState('stopped');
launcherLog.info('Shutdown complete');
}
/** Execute a script through the IPC channel */
async executeScript(content: string, timeoutMs?: number): Promise<IpcResult> {
if (this.state !== 'ready' || !this.ipcClient) {
throw new Error(`Cannot execute script: launcher state is '${this.state}'`);
}
return this.ipcClient.sendCommand(content, timeoutMs);
}
/** Get current launcher status */
getStatus(): LauncherStatus {
return {
state: this.state,
pid: this.pid,
sessionId: this.sessionId,
ipcDir: this.ipcDir,
startedAt: this.startedAt,
lastError: this.lastError,
};
}
/** Check if the CODESYS process is still alive */
isRunning(): boolean {
if (this.pid === null) return false;
try {
process.kill(this.pid, 0); // Signal 0 = test if process exists
return true;
} catch {
return false;
}
}
/** Register callback for state changes */
onStateChange(callback: (state: CodesysState) => void): void {
this.stateChangeCallbacks.push(callback);
}
private setState(state: CodesysState): void {
const prev = this.state;
this.state = state;
if (prev !== state) {
launcherLog.info(`State: ${prev} -> ${state}`);
for (const cb of this.stateChangeCallbacks) {
try { cb(state); } catch { /* ignore callback errors */ }
}
}
}
private startHealthMonitor(): void {
this.healthInterval = setInterval(() => {
if (this.state === 'ready' && !this.isRunning()) {
launcherLog.error('Health check: CODESYS process died');
this.lastError = 'CODESYS process died unexpectedly';
this.pid = null;
this.process = null;
this.setState('error');
this.stopHealthMonitor();
}
}, HEALTH_CHECK_INTERVAL_MS);
}
private stopHealthMonitor(): void {
if (this.healthInterval) {
clearInterval(this.healthInterval);
this.healthInterval = null;
}
}
private sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
}

63
src/logger.ts Normal file
View file

@ -0,0 +1,63 @@
/**
* Structured stderr logging with levels.
* MCP uses stdout for protocol, so all logs go to stderr.
*/
export type LogLevel = 'debug' | 'info' | 'warn' | 'error';
const LOG_LEVELS: Record<LogLevel, number> = {
debug: 0,
info: 1,
warn: 2,
error: 3,
};
let currentLevel: LogLevel = 'info';
export function setLogLevel(level: LogLevel): void {
currentLevel = level;
}
export function getLogLevel(): LogLevel {
return currentLevel;
}
function shouldLog(level: LogLevel): boolean {
return LOG_LEVELS[level] >= LOG_LEVELS[currentLevel];
}
function formatMessage(prefix: string, level: LogLevel, message: string): string {
const timestamp = new Date().toISOString();
return `${timestamp} [${prefix}] ${level.toUpperCase()}: ${message}`;
}
function createLogger(prefix: string) {
return {
debug(message: string): void {
if (shouldLog('debug')) {
process.stderr.write(formatMessage(prefix, 'debug', message) + '\n');
}
},
info(message: string): void {
if (shouldLog('info')) {
process.stderr.write(formatMessage(prefix, 'info', message) + '\n');
}
},
warn(message: string): void {
if (shouldLog('warn')) {
process.stderr.write(formatMessage(prefix, 'warn', message) + '\n');
}
},
error(message: string): void {
if (shouldLog('error')) {
process.stderr.write(formatMessage(prefix, 'error', message) + '\n');
}
},
};
}
export const ipcLog = createLogger('IPC');
export const launcherLog = createLogger('LAUNCHER');
export const serverLog = createLogger('SERVER');
export const watcherLog = createLogger('WATCHER');
export const headlessLog = createLogger('HEADLESS');

72
src/script-manager.ts Normal file
View file

@ -0,0 +1,72 @@
/**
* Python script template loading and interpolation.
* Loads .py templates from src/scripts/, caches them, and performs {PARAM} replacement.
*/
import * as fs from 'fs';
import * as path from 'path';
import { ScriptParams } from './types';
export class ScriptManager {
private scriptsDir: string;
private cache: Map<string, string> = new Map();
constructor(scriptsDir?: string) {
this.scriptsDir = scriptsDir ?? path.join(__dirname, 'scripts');
}
/** Synchronously load a template file and cache it */
loadTemplate(name: string): string {
const fileName = name.endsWith('.py') ? name : `${name}.py`;
const cached = this.cache.get(fileName);
if (cached !== undefined) {
return cached;
}
const filePath = path.join(this.scriptsDir, fileName);
if (!fs.existsSync(filePath)) {
throw new Error(`Script template not found: ${filePath}`);
}
const content = fs.readFileSync(filePath, 'utf-8');
this.cache.set(fileName, content);
return content;
}
/**
* Replace {KEY} placeholders with values.
* No automatic escaping callers are responsible for escaping values
* appropriate to their Python context (raw strings, triple-quoted strings, etc.).
*/
interpolate(template: string, params: ScriptParams): string {
let result = template;
for (const [key, value] of Object.entries(params)) {
const pattern = new RegExp(`\\{${key}\\}`, 'g');
result = result.replace(pattern, String(value));
}
return result;
}
/** Concatenate multiple script fragments with double newlines */
combineScripts(...scripts: string[]): string {
return scripts.join('\n\n');
}
/** Load a template and interpolate parameters */
prepareScript(name: string, params: ScriptParams): string {
const template = this.loadTemplate(name);
return this.interpolate(template, params);
}
/** Prepend helper scripts before the main script, then interpolate all */
prepareScriptWithHelpers(
name: string,
params: ScriptParams,
helpers: string[]
): string {
const helperContents = helpers.map((h) => this.loadTemplate(h));
const mainTemplate = this.loadTemplate(name);
const combined = this.combineScripts(...helperContents, mainTemplate);
return this.interpolate(combined, params);
}
}

View file

@ -0,0 +1,21 @@
import sys, scriptengine as script_engine, os, traceback
project_open = False; project_name = "No project open"; project_path = "N/A"; scripting_ok = False
try:
scripting_ok = True; primary_project = script_engine.projects.primary
if primary_project:
project_open = True
try:
project_path = os.path.normcase(os.path.abspath(primary_project.path))
try:
project_name = primary_project.get_name() # Might fail
if not project_name: project_name = "Unnamed (path: %s)" % os.path.basename(project_path)
except: project_name = "Unnamed (path: %s)" % os.path.basename(project_path)
except Exception as e_path: project_path = "N/A (Error: %s)" % e_path; project_name = "Unnamed (Path Error)"
print("Project Open: %s" % project_open); print("Project Name: %s" % project_name)
print("Project Path: %s" % project_path); print("Scripting OK: %s" % scripting_ok)
print("SCRIPT_SUCCESS: Status check complete."); sys.exit(0)
except Exception as e:
error_message = "Error during status check: %s" % e
print(error_message); print("Scripting OK: False")
# traceback.print_exc() # Optional traceback
print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)

View file

@ -0,0 +1,59 @@
import sys, scriptengine as script_engine, os, traceback
try:
print("DEBUG: compile_project script: Project='%s'" % PROJECT_FILE_PATH)
primary_project = ensure_project_open(PROJECT_FILE_PATH)
project_name = os.path.basename(PROJECT_FILE_PATH)
target_app = None
app_name = "N/A"
# Try getting active application first
try:
target_app = primary_project.active_application
if target_app:
app_name = getattr(target_app, 'get_name', lambda: "Unnamed App (Active)")()
print("DEBUG: Found active application: %s" % app_name)
except Exception as active_err:
print("WARN: Could not get active application: %s. Searching..." % active_err)
# If no active app, search for the first one
if not target_app:
print("DEBUG: Searching for first compilable application...")
apps = []
try:
# Search recursively through all project objects
all_children = primary_project.get_children(True)
for child in all_children:
# Check using the marker property and if build method exists
if hasattr(child, 'is_application') and child.is_application and hasattr(child, 'build'):
app_name_found = getattr(child, 'get_name', lambda: "Unnamed App")()
print("DEBUG: Found potential application object: %s" % app_name_found)
apps.append(child)
break # Take the first one found
except Exception as find_err: print("WARN: Error finding application object: %s" % find_err)
if not apps: raise RuntimeError("No compilable application found in project '%s'" % project_name)
target_app = apps[0]
app_name = getattr(target_app, 'get_name', lambda: "Unnamed App (First Found)")()
print("WARN: Compiling first found application: %s" % app_name)
print("DEBUG: Calling build() on app '%s'..." % app_name)
if not hasattr(target_app, 'build'):
raise TypeError("Selected object '%s' is not an application or doesn't support build()." % app_name)
# Execute the build
target_app.build();
print("DEBUG: Build command executed for application '%s'." % app_name)
# Check messages is harder without direct access to message store from script.
# Rely on CODESYS UI or log output for now.
print("Compile Initiated For Application: %s" % app_name)
print("In Project: %s" % project_name)
print("SCRIPT_SUCCESS: Application compilation initiated.")
sys.exit(0)
except Exception as e:
detailed_error = traceback.format_exc()
error_message = "Error initiating compilation for project %s: %s\\n%s" % (PROJECT_FILE_PATH, e, detailed_error)
print(error_message)
print("SCRIPT_ERROR: %s" % error_message)
sys.exit(1)

View file

@ -0,0 +1,68 @@
import sys, scriptengine as script_engine, os, traceback
PARENT_POU_FULL_PATH = "{PARENT_POU_FULL_PATH}" # e.g., "Application/MyFB"
METHOD_NAME = "{METHOD_NAME}"
RETURN_TYPE = "{RETURN_TYPE}" # Can be empty string for no return type
# Optional: Language
# LANG_GUID_STR = "{LANG_GUID_STR}" # Example if needed
try:
print("DEBUG: create_method script: ParentPOU='%s', Name='%s', ReturnType='%s', Project='%s'" % (PARENT_POU_FULL_PATH, METHOD_NAME, RETURN_TYPE, PROJECT_FILE_PATH))
primary_project = ensure_project_open(PROJECT_FILE_PATH)
if not PARENT_POU_FULL_PATH: raise ValueError("Parent POU full path empty.")
if not METHOD_NAME: raise ValueError("Method name empty.")
# RETURN_TYPE can be empty
# Find the parent POU object
parent_pou_object = find_object_by_path_robust(primary_project, PARENT_POU_FULL_PATH, "parent POU")
if not parent_pou_object: raise ValueError("Parent POU object not found: %s" % PARENT_POU_FULL_PATH)
parent_pou_name = getattr(parent_pou_object, 'get_name', lambda: PARENT_POU_FULL_PATH)()
print("DEBUG: Found Parent POU object: %s" % parent_pou_name)
# Check if parent object supports creating methods (should implement ScriptIecLanguageMemberContainer)
if not hasattr(parent_pou_object, 'create_method'):
raise TypeError("Parent object '%s' of type %s does not support create_method." % (parent_pou_name, type(parent_pou_object).__name__))
# Default language to None (usually ST)
lang_guid = None
# Use None if RETURN_TYPE is empty string, otherwise use the string
actual_return_type = RETURN_TYPE if RETURN_TYPE else None
print("DEBUG: Calling create_method: Name='%s', ReturnType=%s, Lang=%s" % (METHOD_NAME, actual_return_type, lang_guid))
# Call the create_method method ON THE PARENT POU
new_method_object = parent_pou_object.create_method(
name=METHOD_NAME,
return_type=actual_return_type,
language=lang_guid # Pass None to use default
)
if new_method_object:
new_meth_name = getattr(new_method_object, 'get_name', lambda: METHOD_NAME)()
print("DEBUG: Method object created: %s" % new_meth_name)
# --- SAVE THE PROJECT TO PERSIST THE NEW METHOD OBJECT ---
try:
print("DEBUG: Saving Project (after method creation)...")
primary_project.save()
print("DEBUG: Project saved successfully after method creation.")
except Exception as save_err:
print("ERROR: Failed to save Project after creating method: %s" % save_err)
detailed_error = traceback.format_exc()
error_message = "Error saving Project after creating method '%s': %s\\n%s" % (METHOD_NAME, save_err, detailed_error)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)
# --- END SAVING ---
print("Method Created: %s" % new_meth_name)
print("Parent POU: %s" % PARENT_POU_FULL_PATH)
print("Return Type: %s" % (RETURN_TYPE if RETURN_TYPE else "(None)"))
print("SCRIPT_SUCCESS: Method created successfully.")
sys.exit(0)
else:
error_message = "Failed to create method '%s' under '%s'. create_method returned None." % (METHOD_NAME, parent_pou_name)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)
except Exception as e:
detailed_error = traceback.format_exc()
error_message = "Error creating method '%s' under POU '%s' in project '%s': %s\\n%s" % (METHOD_NAME, PARENT_POU_FULL_PATH, PROJECT_FILE_PATH, e, detailed_error)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)

129
src/scripts/create_pou.py Normal file
View file

@ -0,0 +1,129 @@
import sys, scriptengine as script_engine, os, traceback
POU_NAME = "{POU_NAME}"
POU_TYPE_STR = "{POU_TYPE_STR}"
IMPL_LANGUAGE_STR = "{IMPL_LANGUAGE_STR}"
PARENT_PATH_REL = "{PARENT_PATH}"
pou_type_map = {
"Program": script_engine.PouType.Program,
"FunctionBlock": script_engine.PouType.FunctionBlock,
"Function": script_engine.PouType.Function
}
# Map common language names to ImplementationLanguages attributes if needed (optional, None usually works)
# lang_map = { "ST": script_engine.ImplementationLanguage.st, ... }
try:
print("DEBUG: create_pou script: Name='%s', Type='%s', Lang='%s', ParentPath='%s', Project='%s'" % (POU_NAME, POU_TYPE_STR, IMPL_LANGUAGE_STR, PARENT_PATH_REL, PROJECT_FILE_PATH))
primary_project = ensure_project_open(PROJECT_FILE_PATH)
if not POU_NAME: raise ValueError("POU name empty.")
if not PARENT_PATH_REL: raise ValueError("Parent path empty.")
# Resolve POU Type Enum
pou_type_enum = pou_type_map.get(POU_TYPE_STR)
if not pou_type_enum: raise ValueError("Invalid POU type string: %s. Use Program, FunctionBlock, or Function." % POU_TYPE_STR)
# For common case where user just specified "Application", automatically try to find it
if PARENT_PATH_REL == "Application":
# Get project name from file path to build the likely full path
project_name = os.path.splitext(os.path.basename(PROJECT_FILE_PATH))[0]
potential_paths = [
PARENT_PATH_REL, # Original "Application"
"%s.%s" % (project_name, PARENT_PATH_REL), # "projectName.Application"
"%s/%s" % (project_name, PARENT_PATH_REL), # "projectName/Application"
"PLCWinNT/Plc Logic/Application", # Common CODESYS structure
"PLCWinNT.Plc Logic.Application", # Using dots instead
project_name # Just the project name itself might work
]
print("DEBUG: Parent path is simply 'Application', trying several variants to find it")
# Try each potential path until one works
parent_object = None
for path in potential_paths:
print("DEBUG: Attempting to find parent with path: '%s'" % path)
parent_candidate = find_object_by_path_robust(primary_project, path, "parent container")
if parent_candidate:
parent_object = parent_candidate
print("DEBUG: Successfully found parent using path: '%s'" % path)
break
if not parent_object:
# For diagnostics, try to get the application object directly as a fallback
print("DEBUG: All path attempts failed. Trying to access application directly...")
try:
if hasattr(primary_project, 'active_application'):
app = primary_project.active_application
if app:
parent_object = app
print("DEBUG: Found application object directly: %s" % app.get_name())
if not parent_object and hasattr(primary_project, 'find'):
apps = primary_project.find("Application", True)
if apps:
parent_object = apps[0]
print("DEBUG: Found application via search: %s" % parent_object.get_name())
except Exception as e:
print("ERROR: Direct application access also failed: %s" % e)
else:
# Use the provided path normally
parent_object = find_object_by_path_robust(primary_project, PARENT_PATH_REL, "parent container")
# Final check if parent was found
if not parent_object:
raise ValueError("Parent object not found for path: %s. Try using the full path like 'ProjectName.Application' or run get_project_structure first to see the correct structure." % PARENT_PATH_REL)
parent_name = getattr(parent_object, 'get_name', lambda: str(parent_object))()
print("DEBUG: Using parent object: %s (Type: %s)" % (parent_name, type(parent_object).__name__))
# Check if parent object supports creating POUs (should implement ScriptIecLanguageObjectContainer)
if not hasattr(parent_object, 'create_pou'):
raise TypeError("Parent object '%s' of type %s does not support create_pou." % (parent_name, type(parent_object).__name__))
# Set language GUID to None (let CODESYS default based on parent/settings)
lang_guid = None
print("DEBUG: Setting language to None (will use default).")
# Example if mapping language string: lang_guid = lang_map.get(IMPL_LANGUAGE_STR, None)
print("DEBUG: Calling parent_object.create_pou: Name='%s', Type=%s, Lang=%s" % (POU_NAME, pou_type_enum, lang_guid))
# Call create_pou using keyword arguments
new_pou = parent_object.create_pou(
name=POU_NAME,
type=pou_type_enum,
language=lang_guid # Pass None
)
print("DEBUG: parent_object.create_pou returned: %s" % new_pou)
if new_pou:
new_pou_name = getattr(new_pou, 'get_name', lambda: POU_NAME)()
print("DEBUG: POU object created: %s" % new_pou_name)
# --- SAVE THE PROJECT TO PERSIST THE NEW POU ---
try:
print("DEBUG: Saving Project...")
primary_project.save() # Save the overall project file
print("DEBUG: Project saved successfully after POU creation.")
except Exception as save_err:
print("ERROR: Failed to save Project after POU creation: %s" % save_err)
detailed_error = traceback.format_exc()
error_message = "Error saving Project after creating POU '%s': %s\\n%s" % (new_pou_name, save_err, detailed_error)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)
# --- END SAVING ---
print("POU Created: %s" % new_pou_name)
print("Type: %s" % POU_TYPE_STR)
print("Language: %s (Defaulted)" % IMPL_LANGUAGE_STR)
print("Parent Path: %s" % PARENT_PATH_REL)
print("SCRIPT_SUCCESS: POU created successfully.")
sys.exit(0)
else:
error_message = "Failed to create POU '%s'. create_pou returned None." % POU_NAME
print(error_message)
print("SCRIPT_ERROR: %s" % error_message)
sys.exit(1)
except Exception as e:
detailed_error = traceback.format_exc()
error_message = "Error creating POU '%s' in project '%s': %s\\n%s" % (POU_NAME, PROJECT_FILE_PATH, e, detailed_error)
print(error_message)
print("SCRIPT_ERROR: Error creating POU '%s': %s" % (POU_NAME, e))
sys.exit(1)

View file

@ -0,0 +1,54 @@
import sys, scriptengine as script_engine, os, shutil, time, traceback
# Placeholders
TEMPLATE_PROJECT_PATH = r'{TEMPLATE_PROJECT_PATH}' # Path to Standard.project
PROJECT_FILE_PATH = r'{PROJECT_FILE_PATH}' # Path for the new project (Target Path)
try:
print("DEBUG: Python script create_project (copy from template):")
print("DEBUG: Template Source = %s" % TEMPLATE_PROJECT_PATH)
print("DEBUG: Target Path = %s" % PROJECT_FILE_PATH)
if not PROJECT_FILE_PATH: raise ValueError("Target project file path empty.")
if not TEMPLATE_PROJECT_PATH: raise ValueError("Template project file path empty.")
if not os.path.exists(TEMPLATE_PROJECT_PATH): raise IOError("Template project file not found: %s" % TEMPLATE_PROJECT_PATH)
# 1. Copy the template project file to the new location
target_dir = os.path.dirname(PROJECT_FILE_PATH)
if not os.path.exists(target_dir): print("DEBUG: Creating target directory: %s" % target_dir); os.makedirs(target_dir)
# Check if target file already exists
if os.path.exists(PROJECT_FILE_PATH): print("WARN: Target project file already exists, overwriting: %s" % PROJECT_FILE_PATH)
print("DEBUG: Copying '%s' to '%s'..." % (TEMPLATE_PROJECT_PATH, PROJECT_FILE_PATH))
shutil.copy2(TEMPLATE_PROJECT_PATH, PROJECT_FILE_PATH) # copy2 preserves metadata
print("DEBUG: File copy complete.")
# 2. Open the newly copied project file
print("DEBUG: Opening the copied project: %s" % PROJECT_FILE_PATH)
# Set flags for silent opening
update_mode = script_engine.VersionUpdateFlags.NoUpdates | script_engine.VersionUpdateFlags.SilentMode
# try:
# update_mode = script_engine.VersionUpdateFlags.NoUpdates | script_engine.VersionUpdateFlags.SilentMode
# except AttributeError:
# print("WARN: VersionUpdateFlags not found, using integer flags for open (1 | 2 = 3).")
# update_mode = 3
project = script_engine.projects.open(PROJECT_FILE_PATH, update_flags=update_mode)
print("DEBUG: script_engine.projects.open returned: %s" % project)
if project:
print("DEBUG: Pausing briefly after open...")
time.sleep(1.0)
try:
print("DEBUG: Explicitly saving project after opening copy...")
project.save();
print("DEBUG: Project save after opening copy succeeded.")
except Exception as save_err:
print("WARN: Explicit save after opening copy failed: %s" % save_err)
# Decide if this is critical - maybe not, but good to know.
print("Project Created from Template Copy at: %s" % PROJECT_FILE_PATH)
print("SCRIPT_SUCCESS: Project copied from template and opened successfully.")
sys.exit(0)
else:
error_message = "Failed to open project copy %s after copying template %s. projects.open returned None." % (PROJECT_FILE_PATH, TEMPLATE_PROJECT_PATH)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)
except Exception as e:
detailed_error = traceback.format_exc()
error_message = "Error creating project '%s' from template '%s': %s\\n%s" % (PROJECT_FILE_PATH, TEMPLATE_PROJECT_PATH, e, detailed_error)
print(error_message); print("SCRIPT_ERROR: Error copying/opening template: %s" % e); sys.exit(1)

View file

@ -0,0 +1,66 @@
import sys, scriptengine as script_engine, os, traceback
PARENT_POU_FULL_PATH = "{PARENT_POU_FULL_PATH}" # e.g., "Application/MyFB"
PROPERTY_NAME = "{PROPERTY_NAME}"
PROPERTY_TYPE = "{PROPERTY_TYPE}"
# Optional: Language for Getter/Setter (usually defaults to ST)
# LANG_GUID_STR = "{LANG_GUID_STR}" # Example if needed
try:
print("DEBUG: create_property script: ParentPOU='%s', Name='%s', Type='%s', Project='%s'" % (PARENT_POU_FULL_PATH, PROPERTY_NAME, PROPERTY_TYPE, PROJECT_FILE_PATH))
primary_project = ensure_project_open(PROJECT_FILE_PATH)
if not PARENT_POU_FULL_PATH: raise ValueError("Parent POU full path empty.")
if not PROPERTY_NAME: raise ValueError("Property name empty.")
if not PROPERTY_TYPE: raise ValueError("Property type empty.")
# Find the parent POU object
parent_pou_object = find_object_by_path_robust(primary_project, PARENT_POU_FULL_PATH, "parent POU")
if not parent_pou_object: raise ValueError("Parent POU object not found: %s" % PARENT_POU_FULL_PATH)
parent_pou_name = getattr(parent_pou_object, 'get_name', lambda: PARENT_POU_FULL_PATH)()
print("DEBUG: Found Parent POU object: %s" % parent_pou_name)
# Check if parent object supports creating properties (should implement ScriptIecLanguageMemberContainer)
if not hasattr(parent_pou_object, 'create_property'):
raise TypeError("Parent object '%s' of type %s does not support create_property." % (parent_pou_name, type(parent_pou_object).__name__))
# Default language to None (usually ST)
lang_guid = None
print("DEBUG: Calling create_property: Name='%s', Type='%s', Lang=%s" % (PROPERTY_NAME, PROPERTY_TYPE, lang_guid))
# Call the create_property method ON THE PARENT POU
new_property_object = parent_pou_object.create_property(
name=PROPERTY_NAME,
return_type=PROPERTY_TYPE,
language=lang_guid # Pass None to use default
)
if new_property_object:
new_prop_name = getattr(new_property_object, 'get_name', lambda: PROPERTY_NAME)()
print("DEBUG: Property object created: %s" % new_prop_name)
# --- SAVE THE PROJECT TO PERSIST THE NEW PROPERTY OBJECT ---
try:
print("DEBUG: Saving Project (after property creation)...")
primary_project.save()
print("DEBUG: Project saved successfully after property creation.")
except Exception as save_err:
print("ERROR: Failed to save Project after creating property: %s" % save_err)
detailed_error = traceback.format_exc()
error_message = "Error saving Project after creating property '%s': %s\\n%s" % (PROPERTY_NAME, save_err, detailed_error)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)
# --- END SAVING ---
print("Property Created: %s" % new_prop_name)
print("Parent POU: %s" % PARENT_POU_FULL_PATH)
print("Type: %s" % PROPERTY_TYPE)
print("SCRIPT_SUCCESS: Property created successfully.")
sys.exit(0)
else:
error_message = "Failed to create property '%s' under '%s'. create_property returned None." % (PROPERTY_NAME, parent_pou_name)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)
except Exception as e:
detailed_error = traceback.format_exc()
error_message = "Error creating property '%s' under POU '%s' in project '%s': %s\\n%s" % (PROPERTY_NAME, PARENT_POU_FULL_PATH, PROJECT_FILE_PATH, e, detailed_error)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)

View file

@ -0,0 +1,170 @@
import sys
import scriptengine as script_engine
import os
import time
import traceback
# --- Function to ensure the correct project is open ---
MAX_RETRIES = 3
RETRY_DELAY = 2.0 # seconds (use float for time.sleep)
# Basic path cleanup without excessive escaping
def clean_path(path_str):
"""Clean up a path for CODESYS scripting without excessive escaping"""
# Simply remove any extraneous quotes
cleaned = path_str.strip('"\'')
print("DEBUG: Cleaned path: '%s'" % cleaned)
return cleaned
def ensure_project_open(target_project_path):
print("DEBUG: Ensuring project is open: %s" % target_project_path)
# Just clean the path without adding escapes
path_to_use = clean_path(target_project_path)
# Normalize target path once (must match normcase+abspath used on primary_project.path)
normalized_target_path = os.path.normcase(os.path.abspath(path_to_use))
for attempt in range(MAX_RETRIES):
print("DEBUG: Ensure project attempt %d/%d for %s" % (attempt + 1, MAX_RETRIES, normalized_target_path))
primary_project = None
try:
# Getting primary project might fail if CODESYS instance is unstable
primary_project = script_engine.projects.primary
except Exception as primary_err:
print("WARN: Error getting primary project: %s. Assuming none." % primary_err)
# traceback.print_exc() # Optional: Print stack trace for this error
primary_project = None
current_project_path = ""
project_ok = False # Flag to check if target is confirmed primary and accessible
if primary_project:
try:
# Getting path should be relatively safe if primary_project object exists
current_project_path = os.path.normcase(os.path.abspath(primary_project.path))
print("DEBUG: Current primary project path: %s" % current_project_path)
if current_project_path == normalized_target_path:
# Found the right project as primary, now check if it's usable
print("DEBUG: Target project path matches primary. Checking access...")
try:
# Try a relatively safe operation to confirm object usability
# Getting children count is a reasonable check
_ = len(primary_project.get_children(False))
print("DEBUG: Target project '%s' is primary and accessible." % target_project_path)
project_ok = True
return primary_project # SUCCESS CASE 1: Already open and accessible
except Exception as access_err:
# Project found, but accessing it failed. Might be unstable.
print("WARN: Primary project access check failed for '%s': %s. Will attempt reopen." % (current_project_path, access_err))
# traceback.print_exc() # Optional: Print stack trace
primary_project = None # Force reopen by falling through
else:
# A *different* project is primary
print("DEBUG: Primary project is '%s', not the target '%s'." % (current_project_path, normalized_target_path))
# Consider closing the wrong project if causing issues, but for now, just open target
# try:
# print("DEBUG: Closing incorrect primary project '%s'..." % current_project_path)
# primary_project.close() # Be careful with unsaved changes
# except Exception as close_err:
# print("WARN: Failed to close incorrect primary project: %s" % close_err)
primary_project = None # Force open target project
except Exception as path_err:
# Failed even to get the path of the supposed primary project
print("WARN: Could not get path of current primary project: %s. Assuming not the target." % path_err)
# traceback.print_exc() # Optional: Print stack trace
primary_project = None # Force open target project
# If target project not confirmed as primary and accessible, attempt to open/reopen
if not project_ok:
# Log clearly whether we are opening initially or reopening
if primary_project is None and current_project_path == "":
print("DEBUG: No primary project detected. Attempting to open target: %s" % target_project_path)
elif primary_project is None and current_project_path != "":
print("DEBUG: Primary project was '%s' but failed access check or needed close. Attempting to open target: %s" % (current_project_path, target_project_path))
else: # Includes cases where wrong project was open
print("DEBUG: Target project not primary or initial check failed. Attempting to open/reopen: %s" % target_project_path)
try:
# Set flags for silent opening, handle potential attribute errors
update_mode = script_engine.VersionUpdateFlags.NoUpdates | script_engine.VersionUpdateFlags.SilentMode
# try:
# update_mode = script_engine.VersionUpdateFlags.NoUpdates | script_engine.VersionUpdateFlags.SilentMode
# except AttributeError:
# print("WARN: VersionUpdateFlags not found, using integer flags for open (1 | 2 = 3).")
# update_mode = 3 # 1=NoUpdates, 2=SilentMode
opened_project = None
try:
# The actual open call
print("DEBUG: Calling script_engine.projects.open('%s', update_flags=%s)..." % (target_project_path, update_mode))
opened_project = script_engine.projects.open(target_project_path, update_flags=update_mode)
if not opened_project:
# This is a critical failure if open returns None without exception
print("ERROR: projects.open returned None for %s on attempt %d" % (target_project_path, attempt + 1))
# Allow retry loop to continue
else:
# Open call returned *something*, let's verify
print("DEBUG: projects.open call returned an object for: %s" % target_project_path)
print("DEBUG: Pausing for stabilization after open...")
time.sleep(RETRY_DELAY)
# Re-verify: Is the project now primary and accessible?
recheck_primary = None
try:
recheck_primary = script_engine.projects.primary
print("DEBUG: Recheck primary project type: %s" % type(recheck_primary).__name__)
except Exception as recheck_primary_err:
print("WARN: Error getting primary project after reopen: %s" % recheck_primary_err)
traceback.print_exc() # Print full trace for primary project access issue
if recheck_primary:
recheck_path = ""
try: # Getting path might fail
recheck_path = os.path.normcase(os.path.abspath(recheck_primary.path))
except Exception as recheck_path_err:
print("WARN: Failed to get path after reopen: %s" % recheck_path_err)
if recheck_path == normalized_target_path:
print("DEBUG: Target project confirmed as primary after reopening.")
try: # Final sanity check
_ = len(recheck_primary.get_children(False))
print("DEBUG: Reopened project basic access confirmed.")
return recheck_primary # SUCCESS CASE 2: Successfully opened/reopened
except Exception as access_err_reopen:
print("WARN: Reopened project (%s) basic access check failed: %s." % (normalized_target_path, access_err_reopen))
# traceback.print_exc() # Optional
# Allow retry loop to continue
else:
print("WARN: Different project is primary after reopening! Expected '%s', got '%s'." % (normalized_target_path, recheck_path))
# Allow retry loop to continue, maybe it fixes itself
else:
print("WARN: No primary project found after reopening attempt %d!" % (attempt+1))
# Allow retry loop to continue
except Exception as open_err:
# Catch errors during the open call itself
print("ERROR: Exception during projects.open call on attempt %d: %s" % (attempt + 1, open_err))
traceback.print_exc() # Crucial for diagnosing open failures
# Allow retry loop to continue
except Exception as outer_open_err:
# Catch errors in the flag setup etc.
print("ERROR: Unexpected error during open setup/logic attempt %d: %s" % (attempt + 1, outer_open_err))
traceback.print_exc()
# If we didn't return successfully in this attempt, wait before retrying
if attempt < MAX_RETRIES - 1:
print("DEBUG: Ensure project attempt %d did not succeed. Waiting %f seconds..." % (attempt + 1, RETRY_DELAY))
time.sleep(RETRY_DELAY)
else: # Last attempt failed
print("ERROR: Failed all ensure_project_open attempts for %s." % normalized_target_path)
# If all retries fail after the loop
raise RuntimeError("Failed to ensure project '%s' is open and accessible after %d attempts." % (target_project_path, MAX_RETRIES))
# --- End of function ---
# Placeholder for the project file path (must be set in scripts using this snippet)
PROJECT_FILE_PATH = r"{PROJECT_FILE_PATH}"

View file

@ -0,0 +1,116 @@
import traceback
# --- Find object by path function ---
def find_object_by_path_robust(start_node, full_path, target_type_name="object"):
print("DEBUG: Finding %s by path: '%s'" % (target_type_name, full_path))
# Handle both dot and slash separators
# First replace backslashes with forward slashes
path_with_slashes = full_path.replace('\\', '/').strip('/')
# Then replace dots with slashes (CODESYS sometimes uses dots)
normalized_path = path_with_slashes.replace('.', '/')
# Split into parts
path_parts = normalized_path.split('/')
if not path_parts:
print("ERROR: Path is empty.")
return None
# Determine the actual starting node (project or application)
project = start_node # Assume start_node is project initially
if not hasattr(start_node, 'active_application') and hasattr(start_node, 'project'):
# If start_node is not project but has project ref (e.g., an application), get the project
try: project = start_node.project
except Exception as proj_ref_err:
print("WARN: Could not get project reference from start_node: %s" % proj_ref_err)
# Proceed assuming start_node might be the project anyway or search fails
# Try to get the application object robustly if we think we have the project
app = None
if hasattr(project, 'active_application'):
try: app = project.active_application
except Exception: pass # Ignore errors getting active app
if not app:
try:
apps = project.find("Application", True) # Search recursively
if apps: app = apps[0]
except Exception: pass
# Check if the first path part matches the application name
app_name_lower = ""
if app:
try: app_name_lower = (app.get_name() or "application").lower()
except Exception: app_name_lower = "application" # Fallback
# Decide where to start the traversal
current_obj = start_node # Default to the node passed in
if hasattr(project, 'active_application'): # Only adjust if start_node was likely the project
if app and path_parts[0].lower() == app_name_lower:
print("DEBUG: Path starts with Application name '%s'. Beginning search there." % path_parts[0])
current_obj = app
path_parts = path_parts[1:] # Consume the app name part
# If path was *only* the application name
if not path_parts:
print("DEBUG: Target path is the Application object itself.")
return current_obj
else:
print("DEBUG: Path does not start with Application name. Starting search from project root.")
current_obj = project # Start search from the project root
else:
print("DEBUG: Starting search from originally provided node.")
# Traverse the remaining path parts
parent_path_str = getattr(current_obj, 'get_name', lambda: str(current_obj))() # Safer name getting
for i, part_name in enumerate(path_parts):
is_last_part = (i == len(path_parts) - 1)
print("DEBUG: Searching for part [%d/%d]: '%s' under '%s'" % (i+1, len(path_parts), part_name, parent_path_str))
found_in_parent = None
try:
# Prioritize non-recursive find for direct children
children_of_current = current_obj.get_children(False)
print("DEBUG: Found %d direct children under '%s'." % (len(children_of_current), parent_path_str))
for child in children_of_current:
child_name = getattr(child, 'get_name', lambda: None)() # Safer name getting
# print("DEBUG: Checking child: '%s'" % child_name) # Verbose
if child_name == part_name:
found_in_parent = child
print("DEBUG: Found direct child matching '%s'." % part_name)
break # Found direct child, stop searching children
# If not found directly, AND it's the last part, try recursive find from current parent
if not found_in_parent and is_last_part:
print("DEBUG: Direct find failed for last part '%s'. Trying recursive find under '%s'." % (part_name, parent_path_str))
found_recursive_list = current_obj.find(part_name, True) # Recursive find
if found_recursive_list:
# Maybe add a check here if multiple are found?
found_in_parent = found_recursive_list[0] # Take the first match
print("DEBUG: Found last part '%s' recursively." % part_name)
else:
print("DEBUG: Recursive find also failed for last part '%s'." % part_name)
# Update current object if found
if found_in_parent:
current_obj = found_in_parent
parent_path_str = getattr(current_obj, 'get_name', lambda: part_name)() # Safer name getting
print("DEBUG: Stepped into '%s'." % parent_path_str)
else:
# If not found at any point, the path is invalid from this parent
print("ERROR: Path part '%s' not found under '%s'." % (part_name, parent_path_str))
return None # Path broken
except Exception as find_err:
print("ERROR: Exception while searching for '%s' under '%s': %s" % (part_name, parent_path_str, find_err))
traceback.print_exc()
return None # Error during search
# Final verification (optional but recommended): Check if the found object's name matches the last part
final_expected_name = full_path.split('/')[-1]
found_final_name = getattr(current_obj, 'get_name', lambda: None)() # Safer name getting
if found_final_name == final_expected_name:
print("DEBUG: Final %s found and name verified for path '%s': %s" % (target_type_name, full_path, found_final_name))
return current_obj
else:
print("ERROR: Traversal ended on object '%s' but expected final name was '%s'." % (found_final_name, final_expected_name))
return None # Name mismatch implies target not found as expected
# --- End of find object function ---

View file

@ -0,0 +1,79 @@
import sys, scriptengine as script_engine, os, traceback
POU_FULL_PATH = "{POU_FULL_PATH}"
CODE_START_MARKER = "### POU CODE START ###"
CODE_END_MARKER = "### POU CODE END ###"
DECL_START_MARKER = "### POU DECLARATION START ###"
DECL_END_MARKER = "### POU DECLARATION END ###"
IMPL_START_MARKER = "### POU IMPLEMENTATION START ###"
IMPL_END_MARKER = "### POU IMPLEMENTATION END ###"
try:
print("DEBUG: Getting code: POU_FULL_PATH='%s', Project='%s'" % (POU_FULL_PATH, PROJECT_FILE_PATH))
primary_project = ensure_project_open(PROJECT_FILE_PATH)
if not POU_FULL_PATH: raise ValueError("POU full path empty.")
# Find the target POU/Method/Property object
target_object = find_object_by_path_robust(primary_project, POU_FULL_PATH, "target object")
if not target_object: raise ValueError("Target object not found using path: %s" % POU_FULL_PATH)
target_name = getattr(target_object, 'get_name', lambda: POU_FULL_PATH)()
print("DEBUG: Found target object: %s" % target_name)
declaration_code = ""; implementation_code = ""
# --- Get Declaration Part ---
if hasattr(target_object, 'textual_declaration'):
decl_obj = target_object.textual_declaration
if decl_obj and hasattr(decl_obj, 'text'):
try:
declaration_code = decl_obj.text
print("DEBUG: Got declaration text.")
except Exception as decl_read_err:
print("ERROR: Failed to read declaration text: %s" % decl_read_err)
declaration_code = "/* ERROR reading declaration: %s */" % decl_read_err
else:
print("WARN: textual_declaration exists but is None or has no 'text' attribute.")
else:
print("WARN: No textual_declaration attribute.")
# --- Get Implementation Part ---
if hasattr(target_object, 'textual_implementation'):
impl_obj = target_object.textual_implementation
if impl_obj and hasattr(impl_obj, 'text'):
try:
implementation_code = impl_obj.text
print("DEBUG: Got implementation text.")
except Exception as impl_read_err:
print("ERROR: Failed to read implementation text: %s" % impl_read_err)
implementation_code = "/* ERROR reading implementation: %s */" % impl_read_err
else:
print("WARN: textual_implementation exists but is None or has no 'text' attribute.")
else:
print("WARN: No textual_implementation attribute.")
print("Code retrieved for: %s" % target_name)
# Print declaration between markers, ensuring markers are on separate lines
print("\\n" + DECL_START_MARKER)
print(declaration_code)
print(DECL_END_MARKER + "\\n")
# Print implementation between markers
print(IMPL_START_MARKER)
print(implementation_code)
print(IMPL_END_MARKER + "\\n")
# --- LEGACY MARKERS for backward compatibility if needed ---
# Combine both for old marker format, adding a separator line
# legacy_combined_code = declaration_code + "\\n\\n// Implementation\\n" + implementation_code
# print(CODE_START_MARKER); print(legacy_combined_code); print(CODE_END_MARKER)
# --- END LEGACY ---
print("SCRIPT_SUCCESS: Code retrieved.")
sys.exit(0)
except Exception as e:
detailed_error = traceback.format_exc()
error_message = "Error getting code for object '%s' in project '%s': %s\\n%s" % (POU_FULL_PATH, PROJECT_FILE_PATH, e, detailed_error)
print(error_message)
print("SCRIPT_ERROR: %s" % error_message)
sys.exit(1)

View file

@ -0,0 +1,81 @@
import sys, scriptengine as script_engine, os, traceback
def get_object_structure(obj, indent=0, max_depth=10):
"""Recursively traverse a CODESYS project object and build a text tree."""
lines = []
indent_str = " " * indent
if obj is None:
lines.append("%s- ERROR: Null object received" % indent_str)
return lines
if indent > max_depth:
lines.append("%s- (max depth reached)" % indent_str)
return lines
try:
name = "Unnamed"
obj_type = type(obj).__name__
guid_str = ""
folder_str = ""
try:
name = getattr(obj, 'get_name', lambda: "Unnamed")() or "Unnamed"
if hasattr(obj, 'guid'):
guid_str = " {%s}" % obj.guid
if hasattr(obj, 'is_folder') and obj.is_folder:
folder_str = " [Folder]"
except Exception as name_err:
print("WARN: Error getting name for object: %s" % name_err)
name = "!Error!"
lines.append("%s- %s (%s)%s%s" % (indent_str, name, obj_type, folder_str, guid_str))
# Try to get children - catch errors for objects that don't support it
children = []
if hasattr(obj, 'get_children'):
try:
children = obj.get_children(False)
except Exception as child_err:
lines.append("%s (error getting children: %s)" % (indent_str, child_err))
for child in children:
lines.extend(get_object_structure(child, indent + 1, max_depth))
except Exception as e:
lines.append("%s- Error processing node: %s" % (indent_str, e))
return lines
# --- Main script ---
PROJECT_FILE_PATH = r"{PROJECT_FILE_PATH}"
try:
project_path = PROJECT_FILE_PATH.strip('"\'')
print("DEBUG: Getting structure for project: '%s'" % project_path)
# Use the prepended ensure_project_open helper (same as all other scripts)
primary_project = ensure_project_open(project_path)
if primary_project is None:
raise ValueError("Failed to open project: %s" % project_path)
print("DEBUG: Project object type: %s" % type(primary_project).__name__)
# Build structure tree
print("DEBUG: Starting recursion for project structure (max_depth=15)")
structure_list = get_object_structure(primary_project, max_depth=15)
structure_output = "\n".join(structure_list)
# Output with markers
print("")
print("--- PROJECT STRUCTURE START ---")
print(structure_output)
print("--- PROJECT STRUCTURE END ---")
print("")
print("SCRIPT_SUCCESS: Project structure retrieved.")
sys.exit(0)
except Exception as e:
detailed_error = traceback.format_exc()
print("Error getting structure for %s: %s" % (PROJECT_FILE_PATH, e))
print(detailed_error)
print("SCRIPT_ERROR: %s" % e)
sys.exit(1)

View file

@ -0,0 +1,19 @@
import sys, scriptengine as script_engine, os, traceback
try:
project = ensure_project_open(PROJECT_FILE_PATH)
# Get name from object if possible, otherwise use path basename
proj_name = "Unknown"
try:
if project: proj_name = project.get_name() or os.path.basename(PROJECT_FILE_PATH)
else: proj_name = os.path.basename(PROJECT_FILE_PATH) + " (ensure_project_open returned None?)"
except Exception:
proj_name = os.path.basename(PROJECT_FILE_PATH) + " (name retrieval failed)"
print("Project Opened: %s" % proj_name)
print("SCRIPT_SUCCESS: Project opened successfully.")
sys.exit(0)
except Exception as e:
error_message = "Error opening project %s: %s" % (PROJECT_FILE_PATH, e)
print(error_message)
traceback.print_exc()
print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)

View file

@ -0,0 +1,23 @@
import sys, scriptengine as script_engine, os, traceback
try:
primary_project = ensure_project_open(PROJECT_FILE_PATH)
# Get name from object if possible, otherwise use path basename
project_name = "Unknown"
try:
if primary_project: project_name = primary_project.get_name() or os.path.basename(PROJECT_FILE_PATH)
else: project_name = os.path.basename(PROJECT_FILE_PATH) + " (ensure_project_open returned None?)"
except Exception:
project_name = os.path.basename(PROJECT_FILE_PATH) + " (name retrieval failed)"
print("DEBUG: Saving project: %s (%s)" % (project_name, PROJECT_FILE_PATH))
primary_project.save()
print("DEBUG: project.save() executed.")
print("Project Saved: %s" % project_name)
print("SCRIPT_SUCCESS: Project saved successfully.")
sys.exit(0)
except Exception as e:
error_message = "Error saving project %s: %s" % (PROJECT_FILE_PATH, e)
print(error_message)
traceback.print_exc()
print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)

View file

@ -0,0 +1,92 @@
import sys, scriptengine as script_engine, os, traceback
POU_FULL_PATH = "{POU_FULL_PATH}" # Expecting format like "Application/MyPOU" or "Folder/SubFolder/MyPOU"
DECLARATION_CONTENT = """{DECLARATION_CONTENT}"""
IMPLEMENTATION_CONTENT = """{IMPLEMENTATION_CONTENT}"""
try:
print("DEBUG: set_pou_code script: POU_FULL_PATH='%s', Project='%s'" % (POU_FULL_PATH, PROJECT_FILE_PATH))
primary_project = ensure_project_open(PROJECT_FILE_PATH)
if not POU_FULL_PATH: raise ValueError("POU full path empty.")
# Find the target POU/Method/Property object
target_object = find_object_by_path_robust(primary_project, POU_FULL_PATH, "target object")
if not target_object: raise ValueError("Target object not found using path: %s" % POU_FULL_PATH)
target_name = getattr(target_object, 'get_name', lambda: POU_FULL_PATH)()
print("DEBUG: Found target object: %s" % target_name)
# --- Set Declaration Part ---
declaration_updated = False
# Check if the content is actually provided (might be None/empty if only impl is set)
has_declaration_content = 'DECLARATION_CONTENT' in locals() or 'DECLARATION_CONTENT' in globals()
if has_declaration_content and DECLARATION_CONTENT is not None: # Check not None
if hasattr(target_object, 'textual_declaration'):
decl_obj = target_object.textual_declaration
if decl_obj and hasattr(decl_obj, 'replace'):
try:
print("DEBUG: Accessing textual_declaration...")
decl_obj.replace(DECLARATION_CONTENT)
print("DEBUG: Set declaration text using replace().")
declaration_updated = True
except Exception as decl_err:
print("ERROR: Failed to set declaration text: %s" % decl_err)
traceback.print_exc() # Print stack trace for detailed error
else:
print("WARN: Target '%s' textual_declaration attribute is None or does not have replace(). Skipping declaration update." % target_name)
else:
print("WARN: Target '%s' does not have textual_declaration attribute. Skipping declaration update." % target_name)
else:
print("DEBUG: Declaration content not provided or is None. Skipping declaration update.")
# --- Set Implementation Part ---
implementation_updated = False
has_implementation_content = 'IMPLEMENTATION_CONTENT' in locals() or 'IMPLEMENTATION_CONTENT' in globals()
if has_implementation_content and IMPLEMENTATION_CONTENT is not None: # Check not None
if hasattr(target_object, 'textual_implementation'):
impl_obj = target_object.textual_implementation
if impl_obj and hasattr(impl_obj, 'replace'):
try:
print("DEBUG: Accessing textual_implementation...")
impl_obj.replace(IMPLEMENTATION_CONTENT)
print("DEBUG: Set implementation text using replace().")
implementation_updated = True
except Exception as impl_err:
print("ERROR: Failed to set implementation text: %s" % impl_err)
traceback.print_exc() # Print stack trace for detailed error
else:
print("WARN: Target '%s' textual_implementation attribute is None or does not have replace(). Skipping implementation update." % target_name)
else:
print("WARN: Target '%s' does not have textual_implementation attribute. Skipping implementation update." % target_name)
else:
print("DEBUG: Implementation content not provided or is None. Skipping implementation update.")
# --- SAVE THE PROJECT TO PERSIST THE CODE CHANGE ---
# Only save if something was actually updated to avoid unnecessary saves
if declaration_updated or implementation_updated:
try:
print("DEBUG: Saving Project (after code change)...")
primary_project.save() # Save the overall project file
print("DEBUG: Project saved successfully after code change.")
except Exception as save_err:
print("ERROR: Failed to save Project after setting code: %s" % save_err)
detailed_error = traceback.format_exc()
error_message = "Error saving Project after code change for '%s': %s\\n%s" % (target_name, save_err, detailed_error)
print(error_message); print("SCRIPT_ERROR: %s" % error_message); sys.exit(1)
else:
print("DEBUG: No code parts were updated, skipping project save.")
# --- END SAVING ---
print("Code Set For: %s" % target_name)
print("Path: %s" % POU_FULL_PATH)
print("SCRIPT_SUCCESS: Declaration and/or implementation set successfully.")
sys.exit(0)
except Exception as e:
detailed_error = traceback.format_exc()
error_message = "Error setting code for object '%s' in project '%s': %s\\n%s" % (POU_FULL_PATH, PROJECT_FILE_PATH, e, detailed_error)
print(error_message)
print("SCRIPT_ERROR: %s" % error_message)
sys.exit(1)

271
src/scripts/watcher.py Normal file
View file

@ -0,0 +1,271 @@
"""
Persistent watcher script for CODESYS IPC.
Runs inside CODESYS via --runscript, starts a background polling thread,
then RETURNS so the CODESYS UI stays interactive.
Commands are marshaled to the primary thread via execute_on_primary_thread.
{IPC_BASE_DIR} is interpolated by Node.js before launch.
"""
import sys
import os
import time
import traceback
import json
# --- Configuration ---
IPC_BASE_DIR = r"{IPC_BASE_DIR}"
COMMANDS_DIR = os.path.join(IPC_BASE_DIR, "commands")
RESULTS_DIR = os.path.join(IPC_BASE_DIR, "results")
POLL_INTERVAL = 50 # milliseconds
WATCHER_VERSION = "0.3.0"
# --- Error capture file (written before anything else can fail) ---
_ERROR_FILE = os.path.join(IPC_BASE_DIR, "watcher_error.txt")
def _write_error(msg):
try:
with open(_ERROR_FILE, "a") as f:
f.write("[%f] %s\n" % (time.time(), msg))
except:
pass
try:
# --- Ensure directories exist ---
if not os.path.exists(COMMANDS_DIR):
os.makedirs(COMMANDS_DIR)
if not os.path.exists(RESULTS_DIR):
os.makedirs(RESULTS_DIR)
# --- Atomic file write helper ---
def atomic_write(file_path, content):
tmp_path = file_path + ".tmp"
with open(tmp_path, "w") as f:
f.write(content)
f.flush()
os.fsync(f.fileno())
if os.path.exists(file_path):
os.remove(file_path)
os.rename(tmp_path, file_path)
# --- Write ready signal EARLY (before .NET imports) ---
ready_path = os.path.join(IPC_BASE_DIR, "ready.signal")
info = {
"version": WATCHER_VERSION,
"python_version": sys.version,
"platform": sys.platform,
"ipc_dir": IPC_BASE_DIR,
"timestamp": time.time(),
"pid": os.getpid(),
}
atomic_write(ready_path, json.dumps(info, indent=2))
print("[WATCHER] Ready signal written to %s" % ready_path)
# --- Import .NET threading (after ready signal) ---
_write_error("About to import clr")
import clr
_write_error("clr imported OK")
from System.Threading import Thread, ThreadStart, ManualResetEvent
_write_error("System.Threading imported OK")
import scriptengine as se
_write_error("scriptengine imported OK")
# --- File-based logging (print from bg thread crashes CODESYS) ---
_LOG_FILE = os.path.join(IPC_BASE_DIR, "watcher.log")
def _log(msg):
try:
with open(_LOG_FILE, "a") as f:
f.write("[%f] %s\n" % (time.time(), msg))
except:
pass
# --- Output Capture ---
class OutputCapture:
def __init__(self):
self._buffer = []
def write(self, s):
self._buffer.append(str(s))
def writelines(self, lines):
self._buffer.extend([str(l) for l in lines])
def flush(self):
pass
def getvalue(self):
return ''.join(self._buffer)
# --- Stop event ---
_stop_event = ManualResetEvent(False)
def process_command(command_file):
"""Process a single command. File I/O on bg thread, exec on primary thread."""
command_path = os.path.join(COMMANDS_DIR, command_file)
request_id = command_file.replace(".command.json", "")
result_path = os.path.join(RESULTS_DIR, "%s.result.json" % request_id)
_log("Processing command: %s" % request_id)
# Read command and script (file I/O - safe from bg thread)
try:
with open(command_path, "r") as f:
command_data = json.loads(f.read())
script_path = command_data.get("scriptPath", "")
if not os.path.exists(script_path):
raise IOError("Script file not found: %s" % script_path)
with open(script_path, "r") as f:
script_code = f.read()
except Exception as read_err:
_log("Error reading command: %s" % read_err)
atomic_write(result_path, json.dumps({
"requestId": request_id,
"success": False,
"output": "",
"error": "Read error: %s" % read_err,
"timestamp": time.time(),
}))
return
# Cross-thread communication
shared_result = [None]
done_event = ManualResetEvent(False)
def execute_on_ui():
success = False
output = ""
error = ""
old_stdout = sys.stdout
old_stderr = sys.stderr
capture = OutputCapture()
sys.stdout = capture
sys.stderr = capture
try:
exec_globals = {
'__builtins__': __builtins__,
'sys': sys,
'os': os,
'time': time,
'traceback': traceback,
'shutil': __import__('shutil'),
}
exec(script_code, exec_globals)
output = capture.getvalue()
if "SCRIPT_ERROR" in output:
success = False
error = "Script reported error via SCRIPT_ERROR marker"
elif "SCRIPT_SUCCESS" in output:
success = True
else:
success = True
except SystemExit as e:
output = capture.getvalue()
exit_code = e.code
if exit_code is None or exit_code == 0:
success = True
if "SCRIPT_ERROR" in output:
success = False
error = "Script reported error via SCRIPT_ERROR marker"
elif isinstance(exit_code, int):
if "SCRIPT_SUCCESS" in output and "SCRIPT_ERROR" not in output:
success = True
else:
success = False
error = "Script exited with code %s" % exit_code
elif isinstance(exit_code, str):
success = False
error = exit_code
except Exception as e:
output = capture.getvalue()
error = "%s: %s\n%s" % (type(e).__name__, str(e), traceback.format_exc())
success = False
finally:
sys.stdout = old_stdout
sys.stderr = old_stderr
shared_result[0] = {
"requestId": request_id,
"success": success,
"output": output,
"error": error,
"timestamp": time.time(),
}
done_event.Set()
# Marshal execution to the primary thread
_log("Marshaling to primary thread...")
try:
se.system.execute_on_primary_thread(execute_on_ui)
except Exception as marshal_err:
_log("Marshal error: %s" % marshal_err)
shared_result[0] = {
"requestId": request_id,
"success": False,
"output": "",
"error": "Marshal error: %s" % marshal_err,
"timestamp": time.time(),
}
done_event.Set()
# Wait for completion (2 min timeout)
done_event.WaitOne(120000)
# Write result (file I/O - safe from bg thread)
if shared_result[0]:
atomic_write(result_path, json.dumps(shared_result[0]))
_log("Result written: success=%s" % shared_result[0].get("success"))
else:
_log("ERROR: No result after timeout")
atomic_write(result_path, json.dumps({
"requestId": request_id,
"success": False,
"output": "",
"error": "Timeout waiting for primary thread execution",
"timestamp": time.time(),
}))
# Cleanup command and script files
try:
if os.path.exists(command_path):
os.remove(command_path)
sp = os.path.join(COMMANDS_DIR, "%s.py" % request_id)
if os.path.exists(sp):
os.remove(sp)
except:
pass
def worker():
_log("Background worker started")
while not _stop_event.WaitOne(POLL_INTERVAL):
try:
if os.path.exists(os.path.join(IPC_BASE_DIR, "terminate.signal")):
_log("Terminate signal received")
break
cmd_files = sorted([
f for f in os.listdir(COMMANDS_DIR)
if f.endswith(".command.json")
])
if cmd_files:
process_command(cmd_files[0])
except Exception as e:
_log("Worker error: %s" % e)
_log("Background worker stopped")
# --- Start background worker thread and RETURN ---
print("[WATCHER] Starting background watcher v%s" % WATCHER_VERSION)
print("[WATCHER] IPC directory: %s" % IPC_BASE_DIR)
print("[WATCHER] Python version: %s" % sys.version)
t = Thread(ThreadStart(worker))
t.IsBackground = True
t.Start()
import System
System.GC.KeepAlive(t)
print("[WATCHER] Background thread started, script returning - UI is free")
except Exception as _fatal:
_write_error("FATAL: %s\n%s" % (_fatal, traceback.format_exc()))
print("[WATCHER] FATAL ERROR: %s" % _fatal)
traceback.print_exc()
# Script returns here. CODESYS UI thread is freed.

659
src/server.ts Normal file
View file

@ -0,0 +1,659 @@
/**
* MCP Server registers tools and resources for CODESYS automation.
* Supports persistent (watcher-based) and headless (spawn-per-command) modes.
*/
import * as path from 'path';
import * as fs from 'fs';
import { McpServer, ResourceTemplate } from '@modelcontextprotocol/sdk/server/mcp.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { z } from 'zod';
import { ServerConfig, IpcResult, ScriptExecutor, ExecutionMode } from './types';
import { CodesysLauncher } from './launcher';
import { HeadlessExecutor } from './headless';
import { ScriptManager } from './script-manager';
import { serverLog, setLogLevel } from './logger';
// Zod enums for POU tools
const PouTypeEnum = z.enum(['Program', 'FunctionBlock', 'Function']);
const ImplementationLanguageEnum = z.enum([
'ST', 'LD', 'FBD', 'SFC', 'IL', 'CFC',
'StructuredText', 'LadderDiagram', 'FunctionBlockDiagram',
'SequentialFunctionChart', 'InstructionList', 'ContinuousFunctionChart',
]);
/** Resolve a file path to an absolute normalized path */
function resolvePath(filePath: string, workspaceDir: string): string {
return path.normalize(
path.isAbsolute(filePath) ? filePath : path.join(workspaceDir, filePath)
);
}
/** Sanitize a POU path (forward slashes, no leading/trailing slashes) */
function sanitizePouPath(pouPath: string): string {
return pouPath.replace(/\\/g, '/').replace(/^\/+|\/+$/g, '');
}
/** Format an IpcResult into an MCP tool response */
function formatToolResponse(
result: IpcResult,
successMessage: string
): { content: Array<{ type: 'text'; text: string }>; isError: boolean } {
const success = result.success && result.output.includes('SCRIPT_SUCCESS');
return {
content: [
{
type: 'text' as const,
text: success
? successMessage
: `Operation failed. Output:\n${result.output}${result.error ? '\nError: ' + result.error : ''}`,
},
],
isError: !success,
};
}
/** Check if a file exists (async) */
async function fileExists(filePath: string): Promise<boolean> {
try {
fs.statSync(filePath);
return true;
} catch {
return false;
}
}
export async function startMcpServer(config: ServerConfig): Promise<void> {
// Set log level
if (config.debug) setLogLevel('debug');
else if (config.verbose) setLogLevel('info');
serverLog.info(`Starting CODESYS Persistent MCP Server v0.1.0`);
serverLog.info(`Mode: ${config.mode}`);
serverLog.info(`CODESYS Path: ${config.codesysPath}`);
serverLog.info(`Profile: ${config.profileName}`);
serverLog.info(`Workspace: ${config.workspaceDir}`);
// Validate CODESYS path
if (!fs.existsSync(config.codesysPath)) {
throw new Error(`CODESYS executable not found: ${config.codesysPath}`);
}
// Initialize executor based on mode
let executor: ScriptExecutor;
let launcher: CodesysLauncher | null = null;
let executionMode: ExecutionMode = config.mode;
if (config.mode === 'persistent') {
launcher = new CodesysLauncher(config);
if (config.autoLaunch) {
try {
await launcher.launch();
executor = launcher;
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
serverLog.error(`Persistent launch failed: ${errMsg}`);
if (config.fallbackHeadless) {
serverLog.warn('Falling back to headless mode');
executor = new HeadlessExecutor(config);
executionMode = 'headless';
} else {
throw err;
}
}
} else {
// Launcher exists but not yet launched — will use headless until manually launched
executor = new HeadlessExecutor(config);
executionMode = 'headless';
}
} else {
executor = new HeadlessExecutor(config);
}
const scriptManager = new ScriptManager();
const workspaceDir = config.workspaceDir;
// Create MCP server
const server = new McpServer(
{
name: 'CODESYS Persistent MCP Server',
version: '0.1.0',
},
{
capabilities: {
resources: { listChanged: true },
tools: { listChanged: true },
},
}
);
// Note: using 'as any' cast on server for tool() calls to work around
// TS2589 deep type instantiation with MCP SDK generics + Zod.
const s = server as any;
// ─── Management Tools ────────────────────────────────────────────────
s.tool(
'launch_codesys',
'Manually launch CODESYS with UI. Use when --no-auto-launch was set.',
async () => {
if (!launcher) {
return {
content: [{ type: 'text' as const, text: 'Persistent mode not configured. Use --mode persistent.' }],
isError: true,
};
}
try {
await launcher.launch();
executor = launcher;
executionMode = 'persistent';
return {
content: [{ type: 'text' as const, text: 'CODESYS launched successfully in persistent mode.' }],
isError: false,
};
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
return {
content: [{ type: 'text' as const, text: `Launch failed: ${msg}` }],
isError: true,
};
}
}
);
s.tool(
'shutdown_codesys',
'Shut down the persistent CODESYS instance.',
async () => {
if (!launcher) {
return {
content: [{ type: 'text' as const, text: 'No persistent CODESYS instance to shut down.' }],
isError: true,
};
}
try {
await launcher.shutdown();
executor = new HeadlessExecutor(config);
executionMode = 'headless';
return {
content: [{ type: 'text' as const, text: 'CODESYS shut down successfully.' }],
isError: false,
};
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
return {
content: [{ type: 'text' as const, text: `Shutdown failed: ${msg}` }],
isError: true,
};
}
}
);
s.tool(
'get_codesys_status',
'Get the current status of the CODESYS instance (state, PID, mode).',
async () => {
const status = launcher ? launcher.getStatus() : {
state: 'stopped',
pid: null,
sessionId: null,
ipcDir: null,
startedAt: null,
lastError: null,
};
const text = [
`State: ${status.state}`,
`Mode: ${executionMode}`,
`PID: ${status.pid ?? 'N/A'}`,
`Session: ${status.sessionId ?? 'N/A'}`,
`Started: ${status.startedAt ? new Date(status.startedAt).toISOString() : 'N/A'}`,
status.lastError ? `Last Error: ${status.lastError}` : null,
].filter(Boolean).join('\n');
return {
content: [{ type: 'text' as const, text }],
isError: false,
};
}
);
// ─── Project Tools ───────────────────────────────────────────────────
s.tool(
'open_project',
'Opens an existing CODESYS project file.',
{
filePath: z.string().describe("Path to the project file (e.g., 'C:/Projects/MyPLC.project')."),
},
async (args: { filePath: string }) => {
const escaped = resolvePath(args.filePath, workspaceDir);
const script = scriptManager.prepareScriptWithHelpers(
'open_project', { PROJECT_FILE_PATH: escaped }, ['ensure_project_open']
);
const result = await executor.executeScript(script);
return formatToolResponse(result, `Project opened: ${args.filePath}`);
}
);
s.tool(
'create_project',
'Creates a new CODESYS project from the standard template.',
{
filePath: z.string().describe("Path where the new project file should be created."),
},
async (args: { filePath: string }) => {
const absPath = path.normalize(
path.isAbsolute(args.filePath) ? args.filePath : path.join(workspaceDir, args.filePath)
);
// Find template project
let templatePath = '';
try {
const baseDir = path.dirname(path.dirname(config.codesysPath));
templatePath = path.normalize(path.join(baseDir, 'Templates', 'Standard.project'));
if (!(await fileExists(templatePath))) {
const programData = process.env.ALLUSERSPROFILE || process.env.ProgramData || 'C:\\ProgramData';
const pd1 = path.normalize(path.join(programData, 'CODESYS', 'CODESYS', config.profileName, 'Templates', 'Standard.project'));
if (await fileExists(pd1)) {
templatePath = pd1;
} else {
const pd2 = path.normalize(path.join(programData, 'CODESYS', 'Templates', 'Standard.project'));
if (await fileExists(pd2)) {
templatePath = pd2;
} else {
throw new Error('Standard template project file not found.');
}
}
}
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
return {
content: [{ type: 'text' as const, text: `Template Error: ${msg}` }],
isError: true,
};
}
const script = scriptManager.prepareScript('create_project', {
PROJECT_FILE_PATH: absPath,
TEMPLATE_PROJECT_PATH: templatePath,
});
const result = await executor.executeScript(script);
return formatToolResponse(result, `Project created from template: ${absPath}`);
}
);
s.tool(
'save_project',
'Saves the currently open CODESYS project.',
{
projectFilePath: z.string().describe("Path to the project file to ensure is open before saving."),
},
async (args: { projectFilePath: string }) => {
const escaped = resolvePath(args.projectFilePath, workspaceDir);
const script = scriptManager.prepareScriptWithHelpers(
'save_project', { PROJECT_FILE_PATH: escaped }, ['ensure_project_open']
);
const result = await executor.executeScript(script);
return formatToolResponse(result, `Project saved: ${args.projectFilePath}`);
}
);
// ─── POU Tools ───────────────────────────────────────────────────────
s.tool(
'create_pou',
'Creates a new Program, Function Block, or Function POU within the specified CODESYS project.',
{
projectFilePath: z.string().describe("Path to the project file."),
name: z.string().describe("Name for the new POU (must be a valid IEC identifier)."),
type: z.string().describe("Type of POU: Program, FunctionBlock, or Function."),
language: z.string().describe("Implementation language: ST, LD, FBD, SFC, IL, or CFC."),
parentPath: z.string().describe("Relative path under project root or application (e.g., 'Application')."),
},
async (args: { projectFilePath: string; name: string; type: string; language: string; parentPath: string }) => {
const escProjPath = resolvePath(args.projectFilePath, workspaceDir);
const sanParentPath = sanitizePouPath(args.parentPath);
const script = scriptManager.prepareScriptWithHelpers(
'create_pou',
{
PROJECT_FILE_PATH: escProjPath,
POU_NAME: args.name.trim(),
POU_TYPE_STR: args.type,
IMPL_LANGUAGE_STR: args.language,
PARENT_PATH: sanParentPath,
},
['ensure_project_open', 'find_object_by_path']
);
const result = await executor.executeScript(script);
return formatToolResponse(
result,
`POU '${args.name}' created in '${sanParentPath}' of ${args.projectFilePath}. Project saved.`
);
}
);
s.tool(
'set_pou_code',
'Sets the declaration and/or implementation code for a specific POU, Method, or Property.',
{
projectFilePath: z.string().describe("Path to the project file."),
pouPath: z.string().describe("Full relative path to the target object (e.g., 'Application/MyPOU')."),
declarationCode: z.string().optional().describe("Code for the declaration part (VAR...END_VAR). If omitted, not changed."),
implementationCode: z.string().optional().describe("Code for the implementation logic. If omitted, not changed."),
},
async (args: { projectFilePath: string; pouPath: string; declarationCode?: string; implementationCode?: string }) => {
if (args.declarationCode === undefined && args.implementationCode === undefined) {
return {
content: [{ type: 'text' as const, text: 'Error: At least one of declarationCode or implementationCode must be provided.' }],
isError: true,
};
}
const escProjPath = resolvePath(args.projectFilePath, workspaceDir);
const sanPouPath = sanitizePouPath(args.pouPath);
// Escape for triple-quoted Python strings
const sanDecl = (args.declarationCode ?? '').replace(/\\/g, '\\\\').replace(/"""/g, '\\"\\"\\"');
const sanImpl = (args.implementationCode ?? '').replace(/\\/g, '\\\\').replace(/"""/g, '\\"\\"\\"');
const script = scriptManager.prepareScriptWithHelpers(
'set_pou_code',
{
PROJECT_FILE_PATH: escProjPath,
POU_FULL_PATH: sanPouPath,
DECLARATION_CONTENT: sanDecl,
IMPLEMENTATION_CONTENT: sanImpl,
},
['ensure_project_open', 'find_object_by_path']
);
const result = await executor.executeScript(script);
return formatToolResponse(
result,
`Code set for '${sanPouPath}' in ${args.projectFilePath}. Project saved.`
);
}
);
s.tool(
'create_property',
'Creates a new Property within a specific Function Block POU.',
{
projectFilePath: z.string().describe("Path to the project file."),
parentPouPath: z.string().describe("Relative path to the parent Function Block POU (e.g., 'Application/MyFB')."),
propertyName: z.string().describe("Name for the new property (must be a valid IEC identifier)."),
propertyType: z.string().describe("Data type of the property (e.g., 'BOOL', 'INT', 'MyDUT')."),
},
async (args: { projectFilePath: string; parentPouPath: string; propertyName: string; propertyType: string }) => {
const escProjPath = resolvePath(args.projectFilePath, workspaceDir);
const sanParentPath = sanitizePouPath(args.parentPouPath);
const script = scriptManager.prepareScriptWithHelpers(
'create_property',
{
PROJECT_FILE_PATH: escProjPath,
PARENT_POU_FULL_PATH: sanParentPath,
PROPERTY_NAME: args.propertyName.trim(),
PROPERTY_TYPE: args.propertyType.trim(),
},
['ensure_project_open', 'find_object_by_path']
);
const result = await executor.executeScript(script);
return formatToolResponse(
result,
`Property '${args.propertyName}' created under '${sanParentPath}' in ${args.projectFilePath}. Project saved.`
);
}
);
s.tool(
'create_method',
'Creates a new Method within a specific Function Block POU.',
{
projectFilePath: z.string().describe("Path to the project file."),
parentPouPath: z.string().describe("Relative path to the parent Function Block POU (e.g., 'Application/MyFB')."),
methodName: z.string().describe("Name of the new method (must be a valid IEC identifier)."),
returnType: z.string().optional().describe("Return type (e.g., 'BOOL', 'INT'). Leave empty or omit for no return value."),
},
async (args: { projectFilePath: string; parentPouPath: string; methodName: string; returnType?: string }) => {
const escProjPath = resolvePath(args.projectFilePath, workspaceDir);
const sanParentPath = sanitizePouPath(args.parentPouPath);
const script = scriptManager.prepareScriptWithHelpers(
'create_method',
{
PROJECT_FILE_PATH: escProjPath,
PARENT_POU_FULL_PATH: sanParentPath,
METHOD_NAME: args.methodName.trim(),
RETURN_TYPE: (args.returnType ?? '').trim(),
},
['ensure_project_open', 'find_object_by_path']
);
const result = await executor.executeScript(script);
return formatToolResponse(
result,
`Method '${args.methodName}' created under '${sanParentPath}' in ${args.projectFilePath}. Project saved.`
);
}
);
s.tool(
'compile_project',
'Compiles (Builds) the primary application within a CODESYS project.',
{
projectFilePath: z.string().describe("Path to the project file containing the application to compile."),
},
async (args: { projectFilePath: string }) => {
const escaped = resolvePath(args.projectFilePath, workspaceDir);
const script = scriptManager.prepareScriptWithHelpers(
'compile_project', { PROJECT_FILE_PATH: escaped }, ['ensure_project_open']
);
const result = await executor.executeScript(script, 120_000); // 120s timeout for compile
const success = result.success && result.output.includes('SCRIPT_SUCCESS');
const hasCompileErrors =
result.output.includes('Compile complete --') &&
!/ 0 error\(s\),/.test(result.output);
let message = success
? `Compilation initiated for ${args.projectFilePath}. Check CODESYS messages for results.`
: `Failed initiating compilation for ${args.projectFilePath}. Output:\n${result.output}`;
let isError = !success;
if (success && hasCompileErrors) {
message += ' WARNING: Build command reported errors.';
isError = true;
}
return { content: [{ type: 'text' as const, text: message }], isError };
}
);
// ─── Resources ───────────────────────────────────────────────────────
server.resource(
'project-status',
'codesys://project/status',
async (uri) => {
try {
const script = scriptManager.loadTemplate('check_status');
const result = await executor.executeScript(script);
const outputLines = result.output.split(/[\r\n]+/).filter((l) => l.trim());
const statusData: Record<string, string> = {};
outputLines.forEach((line) => {
const match = line.match(/^([^:]+):\s*(.*)$/);
if (match) statusData[match[1].trim()] = match[2].trim();
});
const statusText = [
'CODESYS Status:',
` - Scripting OK: ${statusData['Scripting OK'] ?? 'Unknown'}`,
` - Project Open: ${statusData['Project Open'] ?? 'Unknown'}`,
` - Project Name: ${statusData['Project Name'] ?? 'Unknown'}`,
` - Project Path: ${statusData['Project Path'] ?? 'N/A'}`,
].join('\n');
const isError =
!result.success ||
statusData['Scripting OK']?.toLowerCase() !== 'true';
return {
contents: [{ uri: uri.href, text: statusText, contentType: 'text/plain' }],
isError,
};
} catch (error) {
const msg = error instanceof Error ? error.message : String(error);
return {
contents: [{ uri: uri.href, text: `Failed status check: ${msg}`, contentType: 'text/plain' }],
isError: true,
};
}
}
);
const projectStructureTemplate = new ResourceTemplate(
'codesys://project/{+project_path}/structure',
{ list: undefined }
);
server.resource(
'project-structure',
projectStructureTemplate,
async (uri, params) => {
const projectPath = params.project_path as string;
if (!projectPath) {
return {
contents: [{ uri: uri.href, text: 'Error: Project path missing.', contentType: 'text/plain' }],
isError: true,
};
}
try {
const escaped = resolvePath(projectPath, workspaceDir);
const script = scriptManager.prepareScriptWithHelpers(
'get_project_structure', { PROJECT_FILE_PATH: escaped }, ['ensure_project_open']
);
const result = await executor.executeScript(script);
let structureText = `Error retrieving structure.\n\n${result.output}`;
let isError = !result.success;
if (result.success && result.output.includes('SCRIPT_SUCCESS')) {
const startMarker = '--- PROJECT STRUCTURE START ---';
const endMarker = '--- PROJECT STRUCTURE END ---';
const startIdx = result.output.indexOf(startMarker);
const endIdx = result.output.indexOf(endMarker);
if (startIdx !== -1 && endIdx !== -1 && startIdx < endIdx) {
structureText = result.output
.substring(startIdx + startMarker.length, endIdx)
.replace(/\\n/g, '\n')
.trim();
} else {
structureText = `Could not parse structure markers.\n\nOutput:\n${result.output}`;
isError = true;
}
}
return {
contents: [{ uri: uri.href, text: structureText, contentType: 'text/plain' }],
isError,
};
} catch (error) {
const msg = error instanceof Error ? error.message : String(error);
return {
contents: [{ uri: uri.href, text: `Failed: ${msg}`, contentType: 'text/plain' }],
isError: true,
};
}
}
);
const pouCodeTemplate = new ResourceTemplate(
'codesys://project/{+project_path}/pou/{+pou_path}/code',
{ list: undefined }
);
server.resource(
'pou-code',
pouCodeTemplate,
async (uri, params) => {
const projectPath = params.project_path as string;
const pouPath = params.pou_path as string;
if (!projectPath || !pouPath) {
return {
contents: [{ uri: uri.href, text: 'Error: Project or POU path missing.', contentType: 'text/plain' }],
isError: true,
};
}
try {
const escProjPath = resolvePath(projectPath, workspaceDir);
const sanPouPath = sanitizePouPath(pouPath);
const script = scriptManager.prepareScriptWithHelpers(
'get_pou_code',
{ PROJECT_FILE_PATH: escProjPath, POU_FULL_PATH: sanPouPath },
['ensure_project_open', 'find_object_by_path']
);
const result = await executor.executeScript(script);
let codeText = `Error retrieving code.\n\n${result.output}`;
let isError = !result.success;
if (result.success && result.output.includes('SCRIPT_SUCCESS')) {
const declStart = '### POU DECLARATION START ###';
const declEnd = '### POU DECLARATION END ###';
const implStart = '### POU IMPLEMENTATION START ###';
const implEnd = '### POU IMPLEMENTATION END ###';
let declaration = '/* Declaration not found */';
let implementation = '/* Implementation not found */';
const ds = result.output.indexOf(declStart);
const de = result.output.indexOf(declEnd);
if (ds !== -1 && de !== -1 && ds < de) {
declaration = result.output.substring(ds + declStart.length, de).replace(/\\n/g, '\n').trim();
}
const is_ = result.output.indexOf(implStart);
const ie = result.output.indexOf(implEnd);
if (is_ !== -1 && ie !== -1 && is_ < ie) {
implementation = result.output.substring(is_ + implStart.length, ie).replace(/\\n/g, '\n').trim();
}
codeText = `// ----- Declaration -----\n${declaration}\n\n// ----- Implementation -----\n${implementation}`;
}
return {
contents: [{ uri: uri.href, text: codeText, contentType: 'text/plain' }],
isError,
};
} catch (error) {
const msg = error instanceof Error ? error.message : String(error);
return {
contents: [{ uri: uri.href, text: `Failed: ${msg}`, contentType: 'text/plain' }],
isError: true,
};
}
}
);
// ─── Connect ─────────────────────────────────────────────────────────
const transport = new StdioServerTransport();
serverLog.info('Connecting MCP server via stdio...');
server.connect(transport);
serverLog.info('MCP Server connected and listening.');
// ─── Graceful Shutdown ───────────────────────────────────────────────
const shutdown = async () => {
serverLog.info('Shutdown signal received');
if (launcher) {
try {
await launcher.shutdown();
} catch {
serverLog.warn('Launcher shutdown failed during signal handler');
}
}
process.exit(0);
};
process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);
process.on('unhandledRejection', (reason) => {
serverLog.error(`Unhandled rejection: ${reason}`);
});
}

73
src/types.ts Normal file
View file

@ -0,0 +1,73 @@
/**
* Shared TypeScript types for codesys-mcp-persistent
*/
export type RequestId = string;
export type SessionId = string;
/** Command file written by Node.js to commands/ directory */
export interface IpcCommand {
requestId: RequestId;
scriptPath: string;
timestamp: number;
}
/** Result file written by watcher to results/ directory */
export interface IpcResult {
requestId: RequestId;
success: boolean;
output: string;
error: string;
timestamp: number;
}
/** CODESYS process lifecycle state */
export type CodesysState = 'stopped' | 'launching' | 'ready' | 'stopping' | 'error';
/** Configuration for launching CODESYS */
export interface LauncherConfig {
codesysPath: string;
profileName: string;
workspaceDir: string;
}
/** Runtime status of the CODESYS launcher */
export interface LauncherStatus {
state: CodesysState;
pid: number | null;
sessionId: SessionId | null;
ipcDir: string | null;
startedAt: number | null;
lastError: string | null;
}
/** IPC transport configuration */
export interface IpcConfig {
baseDir: string;
commandTimeoutMs: number;
pollIntervalMs: number;
maxPollIntervalMs: number;
deleteResultAfterRead: boolean;
}
/** Full server configuration */
export interface ServerConfig extends LauncherConfig {
autoLaunch: boolean;
keepAlive: boolean;
timeoutMs: number;
fallbackHeadless: boolean;
verbose: boolean;
debug: boolean;
mode: ExecutionMode;
}
/** Script template parameters */
export type ScriptParams = Record<string, string>;
/** Execution mode */
export type ExecutionMode = 'persistent' | 'headless';
/** Interface for script executors (both persistent and headless) */
export interface ScriptExecutor {
executeScript(content: string, timeoutMs?: number): Promise<IpcResult>;
}

View file

@ -0,0 +1,48 @@
# Manual CODESYS Integration Tests
These tests require a real CODESYS installation and cannot run in CI.
## Prerequisites
- CODESYS 3.5 SP19 or SP21 installed
- No other CODESYS instances running
- Node.js 18+
## Steps
### 1. Build the package
```bash
npm run build
```
### 2. Test persistent mode
```bash
node dist/bin.js \
--codesys-path "C:\path\to\CODESYS.exe" \
--codesys-profile "CODESYS V3.5 SP21 Patch 3" \
--mode persistent \
--verbose
```
**Verify:**
- CODESYS UI opens
- Console shows "CODESYS watcher is ready"
- The MCP server accepts connections
### 3. Test headless fallback
```bash
node dist/bin.js \
--codesys-path "C:\path\to\CODESYS.exe" \
--codesys-profile "CODESYS V3.5 SP21 Patch 3" \
--mode headless
```
### 4. Test --detect flag
```bash
node dist/bin.js --detect
```
**Verify:** Lists installed CODESYS versions.
### 5. Ctrl+C shutdown
- Press Ctrl+C during persistent mode
- Verify CODESYS shuts down cleanly

View file

@ -0,0 +1,98 @@
import { describe, it, expect } from 'vitest';
import * as path from 'path';
import { ScriptManager } from '../../src/script-manager';
/**
* Integration tests that verify the full script preparation pipeline.
* These don't require CODESYS but verify the template system works end-to-end.
*/
describe('E2E Script Preparation', () => {
const scriptsDir = path.join(__dirname, '..', '..', 'src', 'scripts');
const mgr = new ScriptManager(scriptsDir);
it('open_project script prepares correctly with helpers', () => {
const script = mgr.prepareScriptWithHelpers(
'open_project',
{ PROJECT_FILE_PATH: 'C:\\Projects\\Test.project' },
['ensure_project_open']
);
// Should contain ensure_project_open function
expect(script).toContain('def ensure_project_open');
// Should contain the actual open logic
expect(script).toContain('Project Opened');
// Path should appear as-is (no escaping) since templates use r"..." raw strings
expect(script).toContain('C:\\Projects\\Test.project');
// Should contain success marker
expect(script).toContain('SCRIPT_SUCCESS');
});
it('create_pou script prepares with both helpers', () => {
const script = mgr.prepareScriptWithHelpers(
'create_pou',
{
PROJECT_FILE_PATH: 'C:\\test.project',
POU_NAME: 'MyProgram',
POU_TYPE_STR: 'Program',
IMPL_LANGUAGE_STR: 'ST',
PARENT_PATH: 'Application',
},
['ensure_project_open', 'find_object_by_path']
);
expect(script).toContain('def ensure_project_open');
expect(script).toContain('def find_object_by_path_robust');
expect(script).toContain('MyProgram');
expect(script).toContain('POU_TYPE_STR = "Program"');
});
it('set_pou_code script handles pre-escaped code content', () => {
// Simulate what server.ts does: manually escape code for triple-quoted strings
const declCode = 'VAR\\n x : INT;\\nEND_VAR';
const implCode = 'x := 42;';
const sanDecl = declCode.replace(/\\/g, '\\\\').replace(/"""/g, '\\"\\"\\"');
const sanImpl = implCode.replace(/\\/g, '\\\\').replace(/"""/g, '\\"\\"\\"');
const script = mgr.prepareScriptWithHelpers(
'set_pou_code',
{
PROJECT_FILE_PATH: 'C:\\test.project',
POU_FULL_PATH: 'Application/MyPOU',
DECLARATION_CONTENT: sanDecl,
IMPLEMENTATION_CONTENT: sanImpl,
},
['ensure_project_open', 'find_object_by_path']
);
expect(script).toContain('Application/MyPOU');
expect(script).toContain('x := 42;');
});
it('check_status script has no placeholders after load', () => {
const script = mgr.loadTemplate('check_status');
// check_status has no {PLACEHOLDER} params
expect(script).not.toMatch(/\{[A-Z_]+\}/);
expect(script).toContain('SCRIPT_SUCCESS');
});
it('compile_project script prepares with ensure_project_open', () => {
const script = mgr.prepareScriptWithHelpers(
'compile_project',
{ PROJECT_FILE_PATH: 'C:\\test.project' },
['ensure_project_open']
);
expect(script).toContain('def ensure_project_open');
expect(script).toContain('build()');
});
it('all scripts are loadable', () => {
const scriptNames = [
'check_status', 'compile_project', 'create_method', 'create_pou',
'create_project', 'create_property', 'ensure_project_open',
'find_object_by_path', 'get_pou_code', 'get_project_structure',
'open_project', 'save_project', 'set_pou_code', 'watcher',
];
for (const name of scriptNames) {
expect(() => mgr.loadTemplate(name)).not.toThrow();
const content = mgr.loadTemplate(name);
expect(content.length).toBeGreaterThan(0);
}
});
});

196
tests/mock_watcher.py Normal file
View file

@ -0,0 +1,196 @@
"""
Mock watcher for testing the IPC mechanism without CODESYS.
Simulates the persistent watcher by polling commands/ and executing scripts via exec().
"""
import sys
import os
import time
import json
import traceback
import argparse
class OutputCapture:
"""Capture stdout/stderr for script execution."""
def __init__(self):
self._buffer = []
def write(self, s):
self._buffer.append(str(s))
def writelines(self, lines):
self._buffer.extend([str(l) for l in lines])
def flush(self):
pass
def getvalue(self):
return ''.join(self._buffer)
def atomic_write(file_path, content):
"""Write to .tmp then rename for atomic file creation."""
tmp_path = file_path + ".tmp"
with open(tmp_path, "w") as f:
f.write(content)
f.flush()
os.fsync(f.fileno())
if os.path.exists(file_path):
os.remove(file_path)
os.rename(tmp_path, file_path)
def process_command(commands_dir, results_dir, command_file):
"""Process a single .command.json file."""
command_path = os.path.join(commands_dir, command_file)
request_id = command_file.replace(".command.json", "")
result_path = os.path.join(results_dir, "%s.result.json" % request_id)
success = False
output = ""
error = ""
try:
with open(command_path, "r") as f:
command_data = json.loads(f.read())
script_path = command_data.get("scriptPath", "")
if not os.path.exists(script_path):
raise IOError("Script file not found: %s" % script_path)
with open(script_path, "r") as f:
script_code = f.read()
# Capture stdout/stderr
old_stdout = sys.stdout
old_stderr = sys.stderr
capture = OutputCapture()
sys.stdout = capture
sys.stderr = capture
try:
# Fresh globals dict for namespace isolation
exec_globals = {
'__builtins__': __builtins__,
'sys': sys,
'os': os,
'time': time,
'traceback': traceback,
'shutil': __import__('shutil'),
}
exec(script_code, exec_globals)
output = capture.getvalue()
if "SCRIPT_ERROR" in output:
success = False
error = "Script reported error via SCRIPT_ERROR marker"
elif "SCRIPT_SUCCESS" in output:
success = True
else:
success = True
except SystemExit as e:
output = capture.getvalue()
exit_code = e.code
if exit_code is None or exit_code == 0:
success = True
if "SCRIPT_ERROR" in output:
success = False
error = "Script reported error via SCRIPT_ERROR marker"
elif isinstance(exit_code, int):
if "SCRIPT_SUCCESS" in output and "SCRIPT_ERROR" not in output:
success = True
else:
success = False
error = "Script exited with code %s" % exit_code
elif isinstance(exit_code, str):
success = False
error = exit_code
except Exception as e:
output = capture.getvalue()
error = "%s: %s\n%s" % (type(e).__name__, str(e), traceback.format_exc())
success = False
finally:
sys.stdout = old_stdout
sys.stderr = old_stderr
except Exception as outer_err:
error = "Mock watcher error: %s\n%s" % (str(outer_err), traceback.format_exc())
success = False
result_data = {
"requestId": request_id,
"success": success,
"output": output,
"error": error,
"timestamp": time.time(),
}
try:
atomic_write(result_path, json.dumps(result_data))
except Exception as write_err:
sys.stderr.write("ERROR writing result: %s\n" % write_err)
# Clean up
try:
if os.path.exists(command_path):
os.remove(command_path)
script_file = os.path.join(commands_dir, "%s.py" % request_id)
if os.path.exists(script_file):
os.remove(script_file)
except Exception:
pass
def main():
parser = argparse.ArgumentParser(description="Mock CODESYS watcher for testing")
parser.add_argument("--ipc-dir", required=True, help="IPC base directory")
args = parser.parse_args()
ipc_dir = args.ipc_dir
commands_dir = os.path.join(ipc_dir, "commands")
results_dir = os.path.join(ipc_dir, "results")
os.makedirs(commands_dir, exist_ok=True)
os.makedirs(results_dir, exist_ok=True)
# Write ready signal
ready_path = os.path.join(ipc_dir, "ready.signal")
atomic_write(ready_path, json.dumps({
"version": "mock-0.1.0",
"python_version": sys.version,
"platform": sys.platform,
"ipc_dir": ipc_dir,
"timestamp": time.time(),
"pid": os.getpid(),
}))
# Main loop
try:
while True:
# Check terminate
if os.path.exists(os.path.join(ipc_dir, "terminate.signal")):
break
try:
command_files = sorted([
f for f in os.listdir(commands_dir)
if f.endswith(".command.json")
])
if command_files:
process_command(commands_dir, results_dir, command_files[0])
except Exception as e:
sys.stderr.write("Scan error: %s\n" % e)
time.sleep(0.05)
except KeyboardInterrupt:
pass
if __name__ == "__main__":
main()

176
tests/unit/ipc.test.ts Normal file
View file

@ -0,0 +1,176 @@
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import * as fs from 'fs';
import * as path from 'path';
import * as os from 'os';
import { spawn, ChildProcess } from 'child_process';
import { IpcClient, DEFAULT_IPC_CONFIG } from '../../src/ipc';
const TEST_BASE = path.join(os.tmpdir(), 'codesys-mcp-test-ipc');
function createTestIpcDir(): string {
const dir = path.join(TEST_BASE, `test-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`);
fs.mkdirSync(dir, { recursive: true });
return dir;
}
function spawnMockWatcher(ipcDir: string): ChildProcess {
const mockWatcherPath = path.join(__dirname, '..', 'mock_watcher.py');
const child = spawn('python', [mockWatcherPath, '--ipc-dir', ipcDir], {
stdio: ['pipe', 'pipe', 'pipe'],
});
return child;
}
async function waitForReady(ipcDir: string, timeoutMs = 10_000): Promise<void> {
const readyPath = path.join(ipcDir, 'ready.signal');
const start = Date.now();
while (Date.now() - start < timeoutMs) {
if (fs.existsSync(readyPath)) return;
await new Promise((r) => setTimeout(r, 100));
}
throw new Error('Watcher did not become ready');
}
describe('IpcClient', () => {
let ipcDir: string;
let client: IpcClient;
beforeEach(() => {
ipcDir = createTestIpcDir();
client = new IpcClient({
baseDir: ipcDir,
...DEFAULT_IPC_CONFIG,
commandTimeoutMs: 10_000,
});
});
afterEach(async () => {
try {
fs.rmSync(ipcDir, { recursive: true, force: true });
} catch { /* ignore */ }
});
it('ensureDirectories creates commands/ and results/ dirs', async () => {
await client.ensureDirectories();
expect(fs.existsSync(path.join(ipcDir, 'commands'))).toBe(true);
expect(fs.existsSync(path.join(ipcDir, 'results'))).toBe(true);
});
it('isReady returns false before watcher, true after', async () => {
expect(await client.isReady()).toBe(false);
// Write ready signal manually
fs.mkdirSync(ipcDir, { recursive: true });
fs.writeFileSync(path.join(ipcDir, 'ready.signal'), '{}');
expect(await client.isReady()).toBe(true);
});
it('sendTerminate writes terminate.signal', async () => {
fs.mkdirSync(ipcDir, { recursive: true });
await client.sendTerminate();
expect(fs.existsSync(path.join(ipcDir, 'terminate.signal'))).toBe(true);
});
it('cleanup removes session directory', async () => {
await client.ensureDirectories();
expect(fs.existsSync(ipcDir)).toBe(true);
await client.cleanup();
expect(fs.existsSync(ipcDir)).toBe(false);
});
describe('with mock watcher', () => {
let watcher: ChildProcess;
beforeEach(async () => {
await client.ensureDirectories();
watcher = spawnMockWatcher(ipcDir);
await waitForReady(ipcDir);
});
afterEach(async () => {
// Send terminate signal and wait for watcher to exit
try {
await client.sendTerminate();
await new Promise((r) => setTimeout(r, 500));
} catch { /* ignore */ }
try {
watcher.kill();
} catch { /* ignore */ }
});
it('sendCommand roundtrip - print Hello World', async () => {
const result = await client.sendCommand('print("Hello World")');
expect(result.success).toBe(true);
expect(result.output).toContain('Hello World');
});
it('sendCommand handles script error', async () => {
const result = await client.sendCommand('raise Exception("test error")');
expect(result.success).toBe(false);
expect(result.error).toContain('test error');
});
it('sendCommand handles SystemExit(0) as success', async () => {
const result = await client.sendCommand('import sys; sys.exit(0)');
expect(result.success).toBe(true);
});
it('sendCommand handles SystemExit(1) as failure', async () => {
const result = await client.sendCommand('import sys; sys.exit(1)');
expect(result.success).toBe(false);
});
it('namespace isolation between commands', async () => {
const result1 = await client.sendCommand('x = 42\nprint("set x")');
expect(result1.success).toBe(true);
expect(result1.output).toContain('set x');
const result2 = await client.sendCommand('print(x)');
expect(result2.success).toBe(false); // x not defined in fresh namespace
});
it('large output - 100KB of text', async () => {
const script = 'print("A" * 102400)';
const result = await client.sendCommand(script);
expect(result.success).toBe(true);
expect(result.output.length).toBeGreaterThanOrEqual(102400);
});
it('serialization - concurrent commands execute sequentially', async () => {
// Send two commands concurrently
const p1 = client.sendCommand('import time; time.sleep(0.1); print("first")');
const p2 = client.sendCommand('print("second")');
const [r1, r2] = await Promise.all([p1, p2]);
expect(r1.success).toBe(true);
expect(r2.success).toBe(true);
expect(r1.output).toContain('first');
expect(r2.output).toContain('second');
// r1 should complete before r2 starts (due to mutex)
expect(r1.timestamp).toBeLessThanOrEqual(r2.timestamp);
});
it('SCRIPT_SUCCESS marker in output', async () => {
const result = await client.sendCommand('print("SCRIPT_SUCCESS: done")');
expect(result.success).toBe(true);
expect(result.output).toContain('SCRIPT_SUCCESS');
});
it('SCRIPT_ERROR marker in output', async () => {
const result = await client.sendCommand('print("SCRIPT_ERROR: something went wrong")');
expect(result.success).toBe(false);
});
});
it('sendCommand timeout - no watcher', async () => {
await client.ensureDirectories();
const shortTimeoutClient = new IpcClient({
baseDir: ipcDir,
...DEFAULT_IPC_CONFIG,
commandTimeoutMs: 500,
});
await expect(
shortTimeoutClient.sendCommand('print("hello")')
).rejects.toThrow(/timed out/);
});
});

View file

@ -0,0 +1,56 @@
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import * as fs from 'fs';
import * as path from 'path';
import * as os from 'os';
import { spawn, ChildProcess } from 'child_process';
import { CodesysLauncher } from '../../src/launcher';
// These tests use the mock watcher directly (not CODESYS)
// They validate the launcher's IPC integration behavior
describe('CodesysLauncher', () => {
it('rejects launch when CODESYS exe not found', async () => {
const launcher = new CodesysLauncher({
codesysPath: 'C:\\nonexistent\\CODESYS.exe',
profileName: 'Test Profile',
workspaceDir: os.tmpdir(),
});
await expect(launcher.launch()).rejects.toThrow(/not found/);
const status = launcher.getStatus();
expect(status.state).toBe('error');
});
it('getStatus reports stopped initially', () => {
const launcher = new CodesysLauncher({
codesysPath: 'C:\\nonexistent\\CODESYS.exe',
profileName: 'Test Profile',
workspaceDir: os.tmpdir(),
});
const status = launcher.getStatus();
expect(status.state).toBe('stopped');
expect(status.pid).toBeNull();
expect(status.sessionId).toBeNull();
});
it('isRunning returns false when not launched', () => {
const launcher = new CodesysLauncher({
codesysPath: 'C:\\nonexistent\\CODESYS.exe',
profileName: 'Test Profile',
workspaceDir: os.tmpdir(),
});
expect(launcher.isRunning()).toBe(false);
});
it('executeScript rejects when not ready', async () => {
const launcher = new CodesysLauncher({
codesysPath: 'C:\\nonexistent\\CODESYS.exe',
profileName: 'Test Profile',
workspaceDir: os.tmpdir(),
});
await expect(launcher.executeScript('print("hi")')).rejects.toThrow(/state is 'stopped'/);
});
});

View file

@ -0,0 +1,86 @@
import { describe, it, expect } from 'vitest';
import * as path from 'path';
import { ScriptManager } from '../../src/script-manager';
const SCRIPTS_DIR = path.join(__dirname, '..', '..', 'src', 'scripts');
describe('ScriptManager', () => {
const mgr = new ScriptManager(SCRIPTS_DIR);
it('loads an existing template', () => {
const content = mgr.loadTemplate('check_status');
expect(content).toContain('scriptengine');
expect(content).toContain('SCRIPT_SUCCESS');
});
it('throws for non-existent template', () => {
expect(() => mgr.loadTemplate('nonexistent_script')).toThrow(/not found/);
});
it('interpolates a single param', () => {
const result = mgr.interpolate('hello {FOO}', { FOO: 'bar' });
expect(result).toBe('hello bar');
});
it('passes backslashes through unchanged (raw string templates)', () => {
const result = mgr.interpolate('path = r"{PATH}"', {
PATH: 'C:\\Users\\Test',
});
expect(result).toBe('path = r"C:\\Users\\Test"');
});
it('passes triple quotes through unchanged (callers handle escaping)', () => {
const result = mgr.interpolate('code = """{CODE}"""', {
CODE: 'a """ b',
});
expect(result).toBe('code = """a """ b"""');
});
it('interpolates multiple params', () => {
const result = mgr.interpolate('{A} and {B}', { A: 'x', B: 'y' });
expect(result).toBe('x and y');
});
it('cache hit - second load returns same content', () => {
const first = mgr.loadTemplate('check_status');
const second = mgr.loadTemplate('check_status');
expect(first).toBe(second); // Same reference from cache
});
it('combineScripts concatenates with double newlines', () => {
const result = mgr.combineScripts('script1', 'script2', 'script3');
expect(result).toBe('script1\n\nscript2\n\nscript3');
});
it('prepareScript loads and interpolates', () => {
// create_project has {PROJECT_FILE_PATH} and {TEMPLATE_PROJECT_PATH} placeholders
const result = mgr.prepareScript('create_project', {
PROJECT_FILE_PATH: 'C:\\Projects\\test.project',
TEMPLATE_PROJECT_PATH: 'C:\\Templates\\Standard.project',
});
// Values should appear as-is (no escaping) since templates use r"..." raw strings
expect(result).toContain('C:\\Projects\\test.project');
expect(result).toContain('C:\\Templates\\Standard.project');
});
it('prepareScriptWithHelpers prepends helpers', () => {
const result = mgr.prepareScriptWithHelpers(
'open_project',
{ PROJECT_FILE_PATH: 'C:\\test.project' },
['ensure_project_open']
);
// ensure_project_open content should appear before open_project content
const ensureIdx = result.indexOf('def ensure_project_open');
const openIdx = result.indexOf('Project Opened');
expect(ensureIdx).toBeGreaterThan(-1);
expect(openIdx).toBeGreaterThan(-1);
expect(ensureIdx).toBeLessThan(openIdx);
});
it('Windows path with spaces passes through correctly', () => {
const result = mgr.interpolate('path = r"{PATH}"', {
PATH: 'C:\\Program Files\\CODESYS',
});
expect(result).toBe('path = r"C:\\Program Files\\CODESYS"');
});
});

18
tsconfig.json Normal file
View file

@ -0,0 +1,18 @@
{
"compilerOptions": {
"target": "ES2020",
"module": "commonjs",
"outDir": "dist",
"rootDir": "src",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"declaration": true,
"declarationMap": true,
"sourceMap": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist", "tests"]
}

8
vitest.config.ts Normal file
View file

@ -0,0 +1,8 @@
import { defineConfig } from 'vitest/config';
export default defineConfig({
test: {
testTimeout: 30_000,
hookTimeout: 15_000,
},
});