diff --git a/src/scripts/create_boot_application.py b/src/scripts/create_boot_application.py new file mode 100644 index 0000000..c88da5b --- /dev/null +++ b/src/scripts/create_boot_application.py @@ -0,0 +1,39 @@ +import sys, scriptengine as script_engine, os, traceback + +ONLINE_MODE = {ONLINE_MODE} +OUTPUT_PATH = r"{OUTPUT_PATH}" + +try: + print("DEBUG: create_boot_application script: online=%s, output='%s', Project='%s'" % ( + ONLINE_MODE, OUTPUT_PATH, PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + + if ONLINE_MODE: + # Creates the boot application directly ON the connected device. + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + online_app.create_boot_application() + print("Mode: online") + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Boot application created on device.") + else: + # Offline: writes .app next to the project (or given path). + # Requires generated code (compile first). + app = primary_project.active_application + if app is None: + raise RuntimeError("No active application in project.") + app_name = getattr(app, 'get_name', lambda: "Unknown")() + out = OUTPUT_PATH if OUTPUT_PATH else None + app.create_boot_application(out) + print("Mode: offline") + print("Application: %s" % app_name) + print("Output: %s" % (OUTPUT_PATH if OUTPUT_PATH else ".app next to project>")) + print("SCRIPT_SUCCESS: Offline boot application file created.") + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error creating boot application for project %s: %s\n%s" % (PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/force_variables.py b/src/scripts/force_variables.py new file mode 100644 index 0000000..8bab04a --- /dev/null +++ b/src/scripts/force_variables.py @@ -0,0 +1,46 @@ +import sys, scriptengine as script_engine, os, traceback + +# List of (expression, value) string tuples. +ASSIGNMENTS = {ASSIGNMENTS_PY} + +try: + print("DEBUG: force_variables script: %d assignment(s), Project='%s'" % (len(ASSIGNMENTS), PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + if not ASSIGNMENTS: + raise ValueError("Assignments list empty.") + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + # Prepare each value, then commit the batch as FORCE (value is pinned + # against task writes until unforced). + for expr, val in ASSIGNMENTS: + online_app.set_prepared_value(expr, val) + print("DEBUG: prepared %s = %s" % (expr, val)) + online_app.force_prepared_values() + print("DEBUG: force_prepared_values OK") + + forced_now = [] + try: + forced_now = list(online_app.get_forced_expressions() or []) + except Exception as e: + print("DEBUG: get_forced_expressions failed after force: %s" % e) + + print("### FORCED_START ###") + for expr, val in ASSIGNMENTS: + print("%s = %s" % (expr, val)) + print("### FORCED_END ###") + print("Total Forced On App: %d" % len(forced_now)) + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Forced %d variable(s)." % len(ASSIGNMENTS)) + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + failed = getattr(e, 'failed_expressions', None) + failed_note = ("\nFailed expressions: %s" % list(failed)) if failed else "" + error_message = "Error forcing variables in project %s: %s%s\n%s" % ( + PROJECT_FILE_PATH, e, failed_note, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/list_forced_variables.py b/src/scripts/list_forced_variables.py new file mode 100644 index 0000000..34947b1 --- /dev/null +++ b/src/scripts/list_forced_variables.py @@ -0,0 +1,34 @@ +import sys, scriptengine as script_engine, os, traceback + +try: + print("DEBUG: list_forced_variables script: Project='%s'" % PROJECT_FILE_PATH) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + forced = list(online_app.get_forced_expressions() or []) + prepared = [] + try: + prepared = list(online_app.get_prepared_expressions() or []) + except Exception as e: + print("DEBUG: get_prepared_expressions failed: %s" % e) + + print("### FORCED_START ###") + for expr in forced: + print("forced: %s" % expr) + for expr in prepared: + print("prepared: %s" % expr) + print("### FORCED_END ###") + print("Forced Count: %d" % len(forced)) + print("Prepared Count: %d" % len(prepared)) + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Listed forced/prepared expressions.") + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error listing forced variables in project %s: %s\n%s" % (PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/plc_file_delete.py b/src/scripts/plc_file_delete.py new file mode 100644 index 0000000..a3a4f43 --- /dev/null +++ b/src/scripts/plc_file_delete.py @@ -0,0 +1,37 @@ +import sys, scriptengine as script_engine, os, traceback + +PLC_PATH = r"{PLC_PATH}" +IS_DIRECTORY = {IS_DIRECTORY} +RECURSIVE = {RECURSIVE} + +try: + print("DEBUG: plc_file_delete script: path='%s', isDir=%s, recursive=%s, Project='%s'" % ( + PLC_PATH, IS_DIRECTORY, RECURSIVE, PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + if not PLC_PATH: + raise ValueError("PLC path empty.") + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + online_device = online_app.get_online_device() + if IS_DIRECTORY: + online_device.delete_directory(PLC_PATH, RECURSIVE) + print("DEBUG: delete_directory OK") + print("Deleted Directory: %s (recursive=%s)" % (PLC_PATH, RECURSIVE)) + else: + online_device.delete_file(PLC_PATH) + print("DEBUG: delete_file OK") + print("Deleted File: %s" % PLC_PATH) + + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: PLC delete executed.") + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error deleting '%s' on PLC for project %s: %s\n%s" % ( + PLC_PATH, PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/plc_file_list.py b/src/scripts/plc_file_list.py new file mode 100644 index 0000000..6516349 --- /dev/null +++ b/src/scripts/plc_file_list.py @@ -0,0 +1,38 @@ +import sys, scriptengine as script_engine, os, traceback + +PLC_DIRECTORY = r"{PLC_DIRECTORY}" + +try: + print("DEBUG: plc_file_list script: dir='%s', Project='%s'" % (PLC_DIRECTORY, PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + online_device = online_app.get_online_device() + entries = online_device.get_file_list_of_directory(PLC_DIRECTORY) + entries = list(entries or []) + + print("### FILES_START ###") + for info in entries: + try: + kind = "dir" if info.is_directory else "file" + # size can be an IronPython long -- format with %s, never json. + size = info.size if not info.is_directory else 0 + mtime = str(info.last_modification_time) + print("%s\t%s\t%s\t%s" % (kind, info.name, size, mtime)) + except Exception as e: + print("?\t%s\t-\t" % (getattr(info, 'name', '?'), e)) + print("### FILES_END ###") + print("Entry Count: %d" % len(entries)) + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Listed %d entries in '%s'." % (len(entries), PLC_DIRECTORY or '')) + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error listing PLC directory '%s' for project %s: %s\n%s" % ( + PLC_DIRECTORY, PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/plc_file_transfer.py b/src/scripts/plc_file_transfer.py new file mode 100644 index 0000000..3ca9f2a --- /dev/null +++ b/src/scripts/plc_file_transfer.py @@ -0,0 +1,46 @@ +import sys, scriptengine as script_engine, os, traceback + +DIRECTION = "{DIRECTION}" +LOCAL_PATH = r"{LOCAL_PATH}" +PLC_PATH = r"{PLC_PATH}" +FORCE_OVERWRITE = {FORCE_OVERWRITE} + +try: + print("DEBUG: plc_file_transfer script: direction='%s', local='%s', plc='%s', overwrite=%s, Project='%s'" % ( + DIRECTION, LOCAL_PATH, PLC_PATH, FORCE_OVERWRITE, PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + direction = DIRECTION.lower().strip() + if direction not in ('to_plc', 'from_plc'): + raise ValueError("Invalid direction '%s'. Must be 'to_plc' or 'from_plc'." % DIRECTION) + if not LOCAL_PATH or not PLC_PATH: + raise ValueError("Both localPath and plcPath are required.") + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + online_device = online_app.get_online_device() + if direction == 'to_plc': + if not os.path.isfile(LOCAL_PATH): + raise ValueError("Local file does not exist: %s" % LOCAL_PATH) + # CODESYS naming: download_file = PC -> PLC. + online_device.download_file(LOCAL_PATH, PLC_PATH, FORCE_OVERWRITE) + print("DEBUG: download_file (PC -> PLC) OK") + else: + # upload_file = PLC -> PC. + online_device.upload_file(PLC_PATH, LOCAL_PATH, FORCE_OVERWRITE) + print("DEBUG: upload_file (PLC -> PC) OK") + + print("Direction: %s" % direction) + print("Local: %s" % LOCAL_PATH) + print("PLC: %s" % PLC_PATH) + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: File transfer (%s) completed." % direction) + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error transferring file (%s) for project %s: %s\n%s" % ( + DIRECTION, PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/read_variables.py b/src/scripts/read_variables.py new file mode 100644 index 0000000..7e68e76 --- /dev/null +++ b/src/scripts/read_variables.py @@ -0,0 +1,45 @@ +import sys, scriptengine as script_engine, os, traceback + +EXPRESSIONS = {EXPRESSIONS_PY} + +try: + print("DEBUG: read_variables script: %d expression(s), Project='%s'" % (len(EXPRESSIONS), PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + if not EXPRESSIONS: + raise ValueError("Expressions list empty.") + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + values = None + if hasattr(online_app, 'read_values'): + try: + values = online_app.read_values(EXPRESSIONS) + print("DEBUG: read_values returned %d value(s)" % (len(values) if values is not None else 0)) + except Exception as e: + print("DEBUG: read_values failed, falling back to per-expression read_value: %s" % e) + values = None + + if values is None: + values = [] + for expr in EXPRESSIONS: + try: + values.append(online_app.read_value(expr)) + except Exception as e: + values.append("" % e) + + print("### VALUES_START ###") + for i in range(len(EXPRESSIONS)): + val = values[i] if i < len(values) else None + print("%s = %s" % (EXPRESSIONS[i], val)) + print("### VALUES_END ###") + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Read %d variable(s)." % len(EXPRESSIONS)) + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error reading variables in project %s: %s\n%s" % (PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/reset_application.py b/src/scripts/reset_application.py new file mode 100644 index 0000000..7dfb673 --- /dev/null +++ b/src/scripts/reset_application.py @@ -0,0 +1,44 @@ +import sys, scriptengine as script_engine, os, traceback + +RESET_LEVEL = "{RESET_LEVEL}" + +try: + print("DEBUG: reset_application script: Level='%s', Project='%s'" % (RESET_LEVEL, PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + level = RESET_LEVEL.lower().strip() + if level not in ('warm', 'cold', 'origin'): + raise ValueError("Invalid reset level '%s'. Must be 'warm', 'cold' or 'origin'." % RESET_LEVEL) + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + # ResetOption enum (SP21 ScriptOnline.pyi): Warm / Cold / Original. + reset_option_enum = getattr(script_engine, 'ResetOption', None) + if reset_option_enum is None: + raise TypeError("scriptengine.ResetOption is not available on this SP.") + opt_map = {'warm': 'Warm', 'cold': 'Cold', 'origin': 'Original'} + reset_option = getattr(reset_option_enum, opt_map[level]) + + print("DEBUG: Calling reset(%s)..." % opt_map[level]) + online_app.reset(reset_option) + print("DEBUG: Reset done.") + + state = "unknown" + try: + state = str(online_app.application_state) + except Exception: + pass + + print("Reset Level: %s" % level) + print("Application: %s" % app_name) + print("State After: %s" % state) + print("SCRIPT_SUCCESS: Application reset (%s) executed." % level) + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error resetting application (level '%s') in project %s: %s\n%s" % ( + RESET_LEVEL, PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/source_download.py b/src/scripts/source_download.py new file mode 100644 index 0000000..7d26027 --- /dev/null +++ b/src/scripts/source_download.py @@ -0,0 +1,36 @@ +import sys, scriptengine as script_engine, os, traceback + +COMPACT = {COMPACT} + +try: + print("DEBUG: source_download script: compact=%s, Project='%s'" % (COMPACT, PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + online_device = online_app.get_online_device() + done = False + if hasattr(online_device, 'download_source'): + try: + online_device.download_source(COMPACT) + print("DEBUG: online_device.download_source(bCompact=%s) OK" % COMPACT) + done = True + except Exception as e: + print("DEBUG: online_device.download_source failed: %s" % e) + if not done: + # Fallback: application-level source_download (no compact option). + online_app.source_download() + print("DEBUG: online_app.source_download() OK (fallback)") + + print("Compact: %s" % COMPACT) + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Source archive downloaded to device.") + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error downloading source to device for project %s: %s\n%s" % (PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/source_upload.py b/src/scripts/source_upload.py new file mode 100644 index 0000000..ea4cff9 --- /dev/null +++ b/src/scripts/source_upload.py @@ -0,0 +1,28 @@ +import sys, scriptengine as script_engine, os, traceback + +ARCHIVE_PATH = r"{ARCHIVE_PATH}" + +try: + print("DEBUG: source_upload script: archive='%s', Project='%s'" % (ARCHIVE_PATH, PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + if not ARCHIVE_PATH: + raise ValueError("Archive path empty.") + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + online_device = online_app.get_online_device() + online_device.upload_source(ARCHIVE_PATH) + print("DEBUG: upload_source OK") + + print("Archive: %s" % ARCHIVE_PATH) + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Source archive uploaded from device.") + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error uploading source from device for project %s: %s\n%s" % (PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/unforce_variables.py b/src/scripts/unforce_variables.py new file mode 100644 index 0000000..77a20ac --- /dev/null +++ b/src/scripts/unforce_variables.py @@ -0,0 +1,43 @@ +import sys, scriptengine as script_engine, os, traceback + +# Empty list = unforce ALL forced values on the application. +EXPRESSIONS = {EXPRESSIONS_PY} +RESTORE = {RESTORE} + +try: + print("DEBUG: unforce_variables script: %d expression(s) (empty=all), restore=%s, Project='%s'" % ( + len(EXPRESSIONS), RESTORE, PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + if not EXPRESSIONS: + online_app.unforce_all_values() + print("DEBUG: unforce_all_values OK") + print("Unforced: ALL") + else: + # set_unforce_value stages the unforce; force_prepared_values commits it. + for expr in EXPRESSIONS: + online_app.set_unforce_value(expr, RESTORE) + print("DEBUG: staged unforce for %s" % expr) + online_app.force_prepared_values() + print("DEBUG: force_prepared_values (commit unforce) OK") + print("Unforced: %d expression(s)" % len(EXPRESSIONS)) + + remaining = [] + try: + remaining = list(online_app.get_forced_expressions() or []) + except Exception as e: + print("DEBUG: get_forced_expressions failed after unforce: %s" % e) + print("Still Forced: %d" % len(remaining)) + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Unforce executed.") + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + error_message = "Error unforcing variables in project %s: %s\n%s" % (PROJECT_FILE_PATH, e, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/scripts/write_variables.py b/src/scripts/write_variables.py new file mode 100644 index 0000000..381bf44 --- /dev/null +++ b/src/scripts/write_variables.py @@ -0,0 +1,39 @@ +import sys, scriptengine as script_engine, os, traceback + +# List of (expression, value) string tuples. +ASSIGNMENTS = {ASSIGNMENTS_PY} + +try: + print("DEBUG: write_variables script: %d assignment(s), Project='%s'" % (len(ASSIGNMENTS), PROJECT_FILE_PATH)) + primary_project = ensure_project_open(PROJECT_FILE_PATH) + if not ASSIGNMENTS: + raise ValueError("Assignments list empty.") + + online_app, target_app = ensure_online_connection(primary_project) + app_name = getattr(target_app, 'get_name', lambda: "Unknown")() + ensure_logged_in(online_app) + + # Two-step prepare-then-commit (same pattern as write_variable, but one + # commit for the whole batch so all values land in the same cycle). + for expr, val in ASSIGNMENTS: + online_app.set_prepared_value(expr, val) + print("DEBUG: prepared %s = %s" % (expr, val)) + online_app.write_prepared_values() + print("DEBUG: write_prepared_values OK") + + print("### WRITTEN_START ###") + for expr, val in ASSIGNMENTS: + print("%s = %s" % (expr, val)) + print("### WRITTEN_END ###") + print("Application: %s" % app_name) + print("SCRIPT_SUCCESS: Wrote %d variable(s)." % len(ASSIGNMENTS)) + sys.exit(0) +except Exception as e: + detailed_error = traceback.format_exc() + failed = getattr(e, 'failed_expressions', None) + failed_note = ("\nFailed expressions: %s" % list(failed)) if failed else "" + error_message = "Error writing variables in project %s: %s%s\n%s" % ( + PROJECT_FILE_PATH, e, failed_note, detailed_error) + print(error_message) + print("SCRIPT_ERROR: %s" % error_message) + sys.exit(1) diff --git a/src/server.ts b/src/server.ts index d92e7e6..d2b34ff 100644 --- a/src/server.ts +++ b/src/server.ts @@ -574,6 +574,24 @@ function sanitizePouPath(pouPath: string): string { return pouPath.replace(/\\/g, '/').replace(/^\/+|\/+$/g, ''); } +/** + * Escape an arbitrary string into a double-quoted Python string literal, + * for interpolating user-supplied values (expressions, values) into + * script templates as list/tuple elements. + */ +function pyStringLiteral(s: string): string { + return '"' + s + .replace(/\\/g, '\\\\') + .replace(/"/g, '\\"') + .replace(/\r/g, '\\r') + .replace(/\n/g, '\\n') + '"'; +} + +/** Python boolean literal from a JS boolean. */ +function pyBool(b: boolean): string { + return b ? 'True' : 'False'; +} + /** Format an IpcResult into an MCP tool response */ function formatToolResponse( result: IpcResult, @@ -2018,6 +2036,336 @@ export async function startMcpServer(config: ServerConfig): Promise { } ); + // ─── Online Runtime Tools (SP21 coverage phase 1) ──────────────────── + // API: SP21 ScriptOnline.pyi (ScriptOnlineApplication / ScriptOnlineDevice), + // semantics: helpme-codesys.com/en/ScriptingEngine/ScriptOnline.html + + // Extract the text between marker lines; raw output if markers missing. + const extractMarkerText = (output: string, startMarker: string, endMarker: string): string => { + const startIdx = output.indexOf(startMarker); + const endIdx = output.indexOf(endMarker); + return (startIdx >= 0 && endIdx > startIdx) + ? output.substring(startIdx + startMarker.length, endIdx).trim() + : output.trim(); + }; + + const ONLINE_HELPERS = ['ensure_project_open', 'ensure_online_connection']; + + s.tool( + 'reset_application', + "Resets the online application. 'warm' keeps retain variables, 'cold' clears retains but keeps persistents, 'origin' (ResetOption.Original) erases all variables AND the application from the device — destructive, ask the user before using 'origin'. Clears all breakpoints. Must be connected first (connect_to_device).", + { + projectFilePath: z.string().describe("Path to the project file."), + level: z.enum(['warm', 'cold', 'origin']).describe("Reset level: warm (keep retains), cold (clear retains), origin (erase application from device)."), + }, + async (args: { projectFilePath: string; level: 'warm' | 'cold' | 'origin' }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const script = scriptManager.prepareScriptWithHelpers( + 'reset_application', + { PROJECT_FILE_PATH: escaped, RESET_LEVEL: args.level }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script, 120_000); + return formatToolResponse(result, `Application reset (${args.level}) executed.`); + } + ); + + s.tool( + 'read_variables', + "Reads the current values of MULTIPLE variables from the running PLC application in one call (online_app.read_values). Much cheaper than repeated read_variable calls. Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + expressions: z.array(z.string()).min(1).describe("Variable expressions, e.g. ['PLC_PRG.bRun', 'GVL.nCounter']."), + }, + async (args: { projectFilePath: string; expressions: string[] }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const script = scriptManager.prepareScriptWithHelpers( + 'read_variables', + { + PROJECT_FILE_PATH: escaped, + EXPRESSIONS_PY: '[' + args.expressions.map((e) => pyStringLiteral(e.trim())).join(', ') + ']', + }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script); + const success = result.success && result.output.includes('SCRIPT_SUCCESS'); + if (!success) { + return formatToolResponse(result, ''); + } + const text = extractMarkerText(result.output, '### VALUES_START ###', '### VALUES_END ###'); + return { content: [{ type: 'text' as const, text }], isError: false }; + } + ); + + s.tool( + 'write_variables', + "Writes MULTIPLE variables to the running PLC application in one batch (set_prepared_value xN + one write_prepared_values commit, so all values land in the same cycle). Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + assignments: z.array(z.object({ + expression: z.string().describe("Variable expression, e.g. 'PLC_PRG.bRun'."), + value: z.string().describe("Value to write, e.g. 'TRUE', '42', '3.14'."), + })).min(1).describe("Expression/value pairs to write as one batch."), + }, + async (args: { projectFilePath: string; assignments: Array<{ expression: string; value: string }> }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const assignmentsPy = '[' + args.assignments + .map((a) => `(${pyStringLiteral(a.expression.trim())}, ${pyStringLiteral(a.value)})`) + .join(', ') + ']'; + const script = scriptManager.prepareScriptWithHelpers( + 'write_variables', + { PROJECT_FILE_PATH: escaped, ASSIGNMENTS_PY: assignmentsPy }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script); + return formatToolResponse(result, `Wrote ${args.assignments.length} variable(s) in one batch.`); + } + ); + + s.tool( + 'force_variables', + "FORCES variables in the running PLC application (set_prepared_value xN + force_prepared_values): the values are pinned against task writes until unforced (unforce_variables). Forces survive until unforce or application reset. Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + assignments: z.array(z.object({ + expression: z.string().describe("Variable expression, e.g. 'PLC_PRG.bOverride'."), + value: z.string().describe("Value to force, e.g. 'TRUE', '42'."), + })).min(1).describe("Expression/value pairs to force."), + }, + async (args: { projectFilePath: string; assignments: Array<{ expression: string; value: string }> }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const assignmentsPy = '[' + args.assignments + .map((a) => `(${pyStringLiteral(a.expression.trim())}, ${pyStringLiteral(a.value)})`) + .join(', ') + ']'; + const script = scriptManager.prepareScriptWithHelpers( + 'force_variables', + { PROJECT_FILE_PATH: escaped, ASSIGNMENTS_PY: assignmentsPy }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script); + return formatToolResponse(result, `Forced ${args.assignments.length} variable(s).`); + } + ); + + s.tool( + 'unforce_variables', + "Removes forces from variables in the running PLC application. Omit 'expressions' to unforce ALL forced values (unforce_all_values). With 'expressions', stages set_unforce_value per expression and commits via force_prepared_values. Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + expressions: z.array(z.string()).optional().describe("Expressions to unforce. Omit to unforce ALL."), + restore: z.boolean().optional().describe("If true, restore the value from before forcing (only with explicit expressions). Default false."), + }, + async (args: { projectFilePath: string; expressions?: string[]; restore?: boolean }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const script = scriptManager.prepareScriptWithHelpers( + 'unforce_variables', + { + PROJECT_FILE_PATH: escaped, + EXPRESSIONS_PY: '[' + (args.expressions ?? []).map((e) => pyStringLiteral(e.trim())).join(', ') + ']', + RESTORE: pyBool(args.restore ?? false), + }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script); + return formatToolResponse( + result, + args.expressions?.length + ? `Unforced ${args.expressions.length} variable(s).` + : 'Unforced ALL forced variables.' + ); + } + ); + + s.tool( + 'list_forced_variables', + "Lists all currently FORCED expressions (and staged/prepared expressions) on the online application, including ones forced by other clients/editors. Read-only. Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + }, + async (args: { projectFilePath: string }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const script = scriptManager.prepareScriptWithHelpers( + 'list_forced_variables', + { PROJECT_FILE_PATH: escaped }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script); + const success = result.success && result.output.includes('SCRIPT_SUCCESS'); + if (!success) { + return formatToolResponse(result, ''); + } + const body = extractMarkerText(result.output, '### FORCED_START ###', '### FORCED_END ###'); + const counts = result.output.match(/Forced Count:\s*\d+|Prepared Count:\s*\d+/g)?.join(', ') ?? ''; + return { content: [{ type: 'text' as const, text: body ? `${body}\n(${counts})` : `No forced or prepared expressions. (${counts})` }], isError: false }; + } + ); + + s.tool( + 'create_boot_application', + "Creates a boot application. online=true: creates it directly ON the connected device (survives reboot). online=false (default): writes an offline .app boot file (outputPath, or '.app' next to the project) — requires the project to be compiled first (compile_project).", + { + projectFilePath: z.string().describe("Path to the project file."), + online: z.boolean().optional().describe("true = create on the connected device; false/omitted = write offline .app file."), + outputPath: z.string().optional().describe("Offline only: where to write the .app file. Relative paths resolve against the project directory. Omit for '.app' next to the project."), + }, + async (args: { projectFilePath: string; online?: boolean; outputPath?: string }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const outPath = args.outputPath ?? ''; + const uncErr = outPath ? uncPathError(outPath) : null; + if (uncErr) { + return { content: [{ type: 'text' as const, text: uncErr }], isError: true }; + } + const script = scriptManager.prepareScriptWithHelpers( + 'create_boot_application', + { + PROJECT_FILE_PATH: escaped, + ONLINE_MODE: pyBool(args.online ?? false), + OUTPUT_PATH: outPath, + }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script, 180_000); + return formatToolResponse( + result, + args.online + ? 'Boot application created on device.' + : `Offline boot application file created${outPath ? `: ${outPath}` : ' (default location next to project)'}.` + ); + } + ); + + s.tool( + 'source_download', + "Downloads the project SOURCE archive onto the connected PLC (online_device.download_source), so the source can later be recovered from the device. compact=true stores only the current device's PLC + applications. Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + compact: z.boolean().optional().describe("true = only the current device's PLC and applications; false/omitted = all PLCs and applications in the project."), + }, + async (args: { projectFilePath: string; compact?: boolean }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const script = scriptManager.prepareScriptWithHelpers( + 'source_download', + { PROJECT_FILE_PATH: escaped, COMPACT: pyBool(args.compact ?? false) }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script, 300_000); + return formatToolResponse(result, 'Source archive downloaded to device.'); + } + ); + + s.tool( + 'source_upload', + "Uploads the SOURCE archive stored on the connected PLC and saves it locally as a project archive (usually .prj). Must be connected first; the device must contain a source download (see source_download).", + { + projectFilePath: z.string().describe("Path to the project file."), + archivePath: z.string().describe("Local path to save the uploaded project archive to (e.g. 'C:/temp/uploaded.prj')."), + }, + async (args: { projectFilePath: string; archivePath: string }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const escArchive = resolvePath(args.archivePath, workspaceDir); + const uncErr = uncPathError(escArchive); + if (uncErr) { + return { content: [{ type: 'text' as const, text: uncErr }], isError: true }; + } + const script = scriptManager.prepareScriptWithHelpers( + 'source_upload', + { PROJECT_FILE_PATH: escaped, ARCHIVE_PATH: escArchive }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script, 300_000); + return formatToolResponse(result, `Source archive uploaded from device to: ${escArchive}`); + } + ); + + s.tool( + 'plc_file_list', + "Lists files and directories in a directory on the connected PLC's filesystem (get_file_list_of_directory). Returns kind/name/size/mtime rows. Read-only. Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + plcDirectory: z.string().optional().describe("Remote directory on the PLC (e.g. 'PlcLogic'). Omit/empty for the PLC's root file area."), + }, + async (args: { projectFilePath: string; plcDirectory?: string }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const script = scriptManager.prepareScriptWithHelpers( + 'plc_file_list', + { PROJECT_FILE_PATH: escaped, PLC_DIRECTORY: args.plcDirectory ?? '' }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script, 60_000); + const success = result.success && result.output.includes('SCRIPT_SUCCESS'); + if (!success) { + return formatToolResponse(result, ''); + } + const body = extractMarkerText(result.output, '### FILES_START ###', '### FILES_END ###'); + const header = `PLC directory '${args.plcDirectory || ''}' (kind\tname\tsize\tmodified):`; + return { content: [{ type: 'text' as const, text: body ? `${header}\n${body}` : `${header}\n` }], isError: false }; + } + ); + + s.tool( + 'plc_file_transfer', + "Transfers a single file between the local machine and the connected PLC's filesystem. direction 'to_plc' copies localPath onto the PLC (CODESYS download_file); 'from_plc' copies plcPath to the local machine (upload_file). Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + direction: z.enum(['to_plc', 'from_plc']).describe("'to_plc' = local file onto PLC; 'from_plc' = PLC file to local machine."), + localPath: z.string().describe("Local file path (source for to_plc, destination for from_plc)."), + plcPath: z.string().describe("Remote path on the PLC (destination for to_plc, source for from_plc)."), + forceOverwrite: z.boolean().optional().describe("Overwrite the destination if it already exists. Default false."), + }, + async (args: { projectFilePath: string; direction: 'to_plc' | 'from_plc'; localPath: string; plcPath: string; forceOverwrite?: boolean }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const escLocal = resolvePath(args.localPath, workspaceDir); + const uncErr = uncPathError(escLocal); + if (uncErr) { + return { content: [{ type: 'text' as const, text: uncErr }], isError: true }; + } + const script = scriptManager.prepareScriptWithHelpers( + 'plc_file_transfer', + { + PROJECT_FILE_PATH: escaped, + DIRECTION: args.direction, + LOCAL_PATH: escLocal, + PLC_PATH: args.plcPath, + FORCE_OVERWRITE: pyBool(args.forceOverwrite ?? false), + }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script, 300_000); + return formatToolResponse( + result, + args.direction === 'to_plc' + ? `File transferred to PLC: ${escLocal} -> ${args.plcPath}` + : `File transferred from PLC: ${args.plcPath} -> ${escLocal}` + ); + } + ); + + s.tool( + 'plc_file_delete', + "Deletes a file (or directory) on the connected PLC's filesystem. DESTRUCTIVE — confirm with the user before deleting anything you did not create. Must be connected first.", + { + projectFilePath: z.string().describe("Path to the project file."), + plcPath: z.string().describe("Remote path on the PLC to delete."), + isDirectory: z.boolean().optional().describe("true if plcPath is a directory. Default false (file)."), + recursive: z.boolean().optional().describe("Directories only: delete recursively. Default false."), + }, + async (args: { projectFilePath: string; plcPath: string; isDirectory?: boolean; recursive?: boolean }) => { + const escaped = resolvePath(args.projectFilePath, workspaceDir); + const script = scriptManager.prepareScriptWithHelpers( + 'plc_file_delete', + { + PROJECT_FILE_PATH: escaped, + PLC_PATH: args.plcPath, + IS_DIRECTORY: pyBool(args.isDirectory ?? false), + RECURSIVE: pyBool(args.recursive ?? false), + }, + ONLINE_HELPERS + ); + const result = await executor.executeScript(script, 60_000); + return formatToolResponse(result, `Deleted on PLC: ${args.plcPath}`); + } + ); + // Extract a JSON block between marker lines and pretty-print it; if no // markers found, return the raw output. Used by the device tools so the // agent actually sees the scan results / reachability candidates. diff --git a/tests/integration/online-tools-prep.test.ts b/tests/integration/online-tools-prep.test.ts new file mode 100644 index 0000000..784da55 --- /dev/null +++ b/tests/integration/online-tools-prep.test.ts @@ -0,0 +1,178 @@ +import { describe, it, expect } from 'vitest'; +import * as path from 'path'; +import { ScriptManager } from '../../src/script-manager'; + +/** + * Script-preparation tests for the SP21-coverage phase 1 online/runtime + * tools. Like e2e.test.ts these don't require CODESYS — they verify the + * template + helper + interpolation pipeline end-to-end. + */ +describe('E2E Script Preparation — online runtime tools (SP21 coverage phase 1)', () => { + const scriptsDir = path.join(__dirname, '..', '..', 'src', 'scripts'); + const mgr = new ScriptManager(scriptsDir); + const ONLINE_HELPERS = ['ensure_project_open', 'ensure_online_connection']; + + it('reset_application prepares with level and online helpers', () => { + const script = mgr.prepareScriptWithHelpers( + 'reset_application', + { PROJECT_FILE_PATH: 'C:\\test.project', RESET_LEVEL: 'warm' }, + ONLINE_HELPERS + ); + expect(script).toContain('def ensure_project_open'); + expect(script).toContain('def ensure_online_connection'); + expect(script).toContain('RESET_LEVEL = "warm"'); + expect(script).toContain('ResetOption'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('read_variables prepares with a Python list literal', () => { + const script = mgr.prepareScriptWithHelpers( + 'read_variables', + { PROJECT_FILE_PATH: 'C:\\test.project', EXPRESSIONS_PY: '["PLC_PRG.bRun", "GVL.nCounter"]' }, + ONLINE_HELPERS + ); + expect(script).toContain('EXPRESSIONS = ["PLC_PRG.bRun", "GVL.nCounter"]'); + expect(script).toContain('read_values'); + expect(script).toContain('### VALUES_START ###'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('write_variables prepares with assignment tuples', () => { + const script = mgr.prepareScriptWithHelpers( + 'write_variables', + { PROJECT_FILE_PATH: 'C:\\test.project', ASSIGNMENTS_PY: '[("PLC_PRG.bRun", "TRUE"), ("GVL.n", "42")]' }, + ONLINE_HELPERS + ); + expect(script).toContain('ASSIGNMENTS = [("PLC_PRG.bRun", "TRUE"), ("GVL.n", "42")]'); + expect(script).toContain('write_prepared_values'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('force_variables prepares and commits via force_prepared_values', () => { + const script = mgr.prepareScriptWithHelpers( + 'force_variables', + { PROJECT_FILE_PATH: 'C:\\test.project', ASSIGNMENTS_PY: '[("PLC_PRG.bOverride", "TRUE")]' }, + ONLINE_HELPERS + ); + expect(script).toContain('force_prepared_values'); + expect(script).toContain('get_forced_expressions'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('unforce_variables supports both all and selective unforce', () => { + const script = mgr.prepareScriptWithHelpers( + 'unforce_variables', + { PROJECT_FILE_PATH: 'C:\\test.project', EXPRESSIONS_PY: '[]', RESTORE: 'False' }, + ONLINE_HELPERS + ); + expect(script).toContain('EXPRESSIONS = []'); + expect(script).toContain('RESTORE = False'); + expect(script).toContain('unforce_all_values'); + expect(script).toContain('set_unforce_value'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('list_forced_variables prepares with markers', () => { + const script = mgr.prepareScriptWithHelpers( + 'list_forced_variables', + { PROJECT_FILE_PATH: 'C:\\test.project' }, + ONLINE_HELPERS + ); + expect(script).toContain('get_forced_expressions'); + expect(script).toContain('### FORCED_START ###'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('create_boot_application prepares both online and offline modes', () => { + const script = mgr.prepareScriptWithHelpers( + 'create_boot_application', + { PROJECT_FILE_PATH: 'C:\\test.project', ONLINE_MODE: 'False', OUTPUT_PATH: 'C:\\out\\app.app' }, + ONLINE_HELPERS + ); + expect(script).toContain('ONLINE_MODE = False'); + expect(script).toContain('OUTPUT_PATH = r"C:\\out\\app.app"'); + expect(script).toContain('active_application'); + expect(script).toContain('create_boot_application'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('source_download prepares with compact flag', () => { + const script = mgr.prepareScriptWithHelpers( + 'source_download', + { PROJECT_FILE_PATH: 'C:\\test.project', COMPACT: 'True' }, + ONLINE_HELPERS + ); + expect(script).toContain('COMPACT = True'); + expect(script).toContain('download_source'); + expect(script).toContain('source_download'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('source_upload prepares with archive path', () => { + const script = mgr.prepareScriptWithHelpers( + 'source_upload', + { PROJECT_FILE_PATH: 'C:\\test.project', ARCHIVE_PATH: 'C:\\temp\\up.prj' }, + ONLINE_HELPERS + ); + expect(script).toContain('ARCHIVE_PATH = r"C:\\temp\\up.prj"'); + expect(script).toContain('upload_source'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('plc_file_list prepares with directory and markers', () => { + const script = mgr.prepareScriptWithHelpers( + 'plc_file_list', + { PROJECT_FILE_PATH: 'C:\\test.project', PLC_DIRECTORY: 'PlcLogic' }, + ONLINE_HELPERS + ); + expect(script).toContain('PLC_DIRECTORY = r"PlcLogic"'); + expect(script).toContain('get_file_list_of_directory'); + expect(script).toContain('### FILES_START ###'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('plc_file_transfer prepares both directions', () => { + const script = mgr.prepareScriptWithHelpers( + 'plc_file_transfer', + { + PROJECT_FILE_PATH: 'C:\\test.project', + DIRECTION: 'to_plc', + LOCAL_PATH: 'C:\\local\\f.txt', + PLC_PATH: 'PlcLogic/f.txt', + FORCE_OVERWRITE: 'True', + }, + ONLINE_HELPERS + ); + expect(script).toContain('DIRECTION = "to_plc"'); + expect(script).toContain('FORCE_OVERWRITE = True'); + expect(script).toContain('download_file'); + expect(script).toContain('upload_file'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('plc_file_delete prepares file and directory variants', () => { + const script = mgr.prepareScriptWithHelpers( + 'plc_file_delete', + { PROJECT_FILE_PATH: 'C:\\test.project', PLC_PATH: 'PlcLogic/old.txt', IS_DIRECTORY: 'False', RECURSIVE: 'False' }, + ONLINE_HELPERS + ); + expect(script).toContain('PLC_PATH = r"PlcLogic/old.txt"'); + expect(script).toContain('delete_file'); + expect(script).toContain('delete_directory'); + expect(script).toContain('SCRIPT_SUCCESS'); + }); + + it('all phase-1 scripts are ASCII-only (IronPython 2.7 constraint)', () => { + const names = [ + 'reset_application', 'read_variables', 'write_variables', 'force_variables', + 'unforce_variables', 'list_forced_variables', 'create_boot_application', + 'source_download', 'source_upload', 'plc_file_list', 'plc_file_transfer', + 'plc_file_delete', + ]; + for (const name of names) { + const content = mgr.loadTemplate(name); + // eslint-disable-next-line no-control-regex + expect(/^[\x00-\x7F]*$/.test(content), `${name}.py must be ASCII-only`).toBe(true); + } + }); +});