feat: add 12 online/runtime tools (SP21 ScriptOnline coverage, phase 1)
reset_application (warm/cold/origin), read_variables, write_variables, force_variables, unforce_variables, list_forced_variables, create_boot_application (online/offline), source_download, source_upload, plc_file_list, plc_file_transfer, plc_file_delete. API per SP21 ScriptOnline.pyi (ScriptOnlineApplication/ScriptOnlineDevice); semantics cross-checked against helpme-codesys.com/en/ScriptingEngine/ScriptOnline.html. Plan: docs/superpowers/plans/2026-06-12-sp21-api-coverage.md (phase 1).
This commit is contained in:
Karstein Phobic Nyvold Kvistad
parent
a990cf23ee
commit
3a012a7149
14 changed files with 1001 additions and 0 deletions
39
src/scripts/create_boot_application.py
Normal file
39
src/scripts/create_boot_application.py
Normal file
|
|
@ -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 <output>.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 "<default: <application>.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)
|
||||
46
src/scripts/force_variables.py
Normal file
46
src/scripts/force_variables.py
Normal file
|
|
@ -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)
|
||||
34
src/scripts/list_forced_variables.py
Normal file
34
src/scripts/list_forced_variables.py
Normal file
|
|
@ -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)
|
||||
37
src/scripts/plc_file_delete.py
Normal file
37
src/scripts/plc_file_delete.py
Normal file
|
|
@ -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)
|
||||
38
src/scripts/plc_file_list.py
Normal file
38
src/scripts/plc_file_list.py
Normal file
|
|
@ -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<info error: %s>" % (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 '<root>'))
|
||||
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)
|
||||
46
src/scripts/plc_file_transfer.py
Normal file
46
src/scripts/plc_file_transfer.py
Normal file
|
|
@ -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)
|
||||
45
src/scripts/read_variables.py
Normal file
45
src/scripts/read_variables.py
Normal file
|
|
@ -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("<read failed: %s>" % 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)
|
||||
44
src/scripts/reset_application.py
Normal file
44
src/scripts/reset_application.py
Normal file
|
|
@ -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)
|
||||
36
src/scripts/source_download.py
Normal file
36
src/scripts/source_download.py
Normal file
|
|
@ -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)
|
||||
28
src/scripts/source_upload.py
Normal file
28
src/scripts/source_upload.py
Normal file
|
|
@ -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)
|
||||
43
src/scripts/unforce_variables.py
Normal file
43
src/scripts/unforce_variables.py
Normal file
|
|
@ -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)
|
||||
39
src/scripts/write_variables.py
Normal file
39
src/scripts/write_variables.py
Normal file
|
|
@ -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)
|
||||
348
src/server.ts
348
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<void> {
|
|||
}
|
||||
);
|
||||
|
||||
// ─── 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 '<application>.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 '<application>.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 || '<root>'}' (kind\tname\tsize\tmodified):`;
|
||||
return { content: [{ type: 'text' as const, text: body ? `${header}\n${body}` : `${header}\n<empty>` }], 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.
|
||||
|
|
|
|||
178
tests/integration/online-tools-prep.test.ts
Normal file
178
tests/integration/online-tools-prep.test.ts
Normal file
|
|
@ -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);
|
||||
}
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue