nikon/Codesys-MCP-SP21-plus
0
0
Fork 0
Code Activity Actions
Codesys-MCP-SP21-plus/src/scripts/ensure_project_open.py
Karstein Kvistad 2607063306 fix: 4 broken-tool fixes (compile json long, connect_to_device LoginMode, 
create_folder SP21+ fallback, ensure_project_open cross-project switch)
+ bench results

compile_project.py + get_compile_messages.py:
- IronPython 2.7's json.dumps cannot serialize System.Int64-backed `long`
  values, which is what CODESYS's compile-message objects expose as
  line_number / position. Added _coerce_int + _coerce_str helpers and a
  shared _build_message_entry function. Three duplicated message-building
  blocks collapsed into single helper calls.
- Defensive `try: json.dumps(...) except TypeError: json.dumps(default=str)`
  so a stray field that slips past the helpers doesn't kill the emit.

connect_to_device.py:
- SP21+ may expose the login enum as LoginMode rather than
  OnlineChangeOption. Extended the candidate sweep to probe both
  script_engine.LoginMode and script_engine.OnlineChangeOption AND
  online_app.LoginMode/OnlineChangeOption (some builds attach it to the
  app object). Added "OnlineChange" + "Login" + "Download" to the
  preferred-priority list. Added a 3-arg call shape variant for SPs
  that take (mode, secondary-mode, force-bool).

create_folder.py:
- parent_object.create_folder() is no longer exposed on every parent type
  in SP21+. Added two fallback factories tried in order:
    1. parent.create_object(typeUuid='85d1215e-6520-4983-9a55-2d39d1f24cb4', name=...)
    2. parent.add(script_engine.types.IecFolder, name=...)
  with detailed warnings when each path fails. Final TypeError now lists
  every factory tried so a future SP rotation surfaces clearly.

ensure_project_open.py:
- Uncommented the close-prior-project branch (was a TODO since the
  initial fork). Cross-project switches in a persistent CODESYS now do
  save() -> close() -> 500ms pump -> open(target). Without this,
  projects.open against a different already-primary project fails
  intermittently on file lock contention or pops a "project in use"
  modal that freezes the IDE thread.
- save() is best-effort: if it raises (transient lock, save-as required)
  we still proceed with close + open rather than getting stuck in a
  half-switched state forever.

tests/bench-results.json:
- Captured timings from a clean run on MCPTest2 (PLCWinNT, 5 lib refs,
  ~12 POUs). 9 tools x 2 modes x iterations.
  Headers (mean ms): persistent vs headless --
    open_project              7700  vs   40041   (~5x; first call cold)
    mirror_export             1547  vs   23723   (~15x)
    list_project_libraries    1565  vs   23322   (~15x)
    get_all_pou_code          1607  vs   23376   (~15x)
    save_project              2095  vs   23321   (~11x)
    create_pou (FB)           1540  vs   23903   (~16x)
    delete_object             1544  vs   27420   (~18x)
    bump_project_version      1540  vs   30678   (~20x)
    bump_project_version #2   1556  vs   37769   (~24x)
- set_pou_code FAILED in both modes -- bench harness param-shape issue
  (multi-line code passed verbatim to triple-quoted-string interpolation
  doesn't survive the round-trip). Tool itself works fine through the
  MCP tool call path; bench needs to escape newlines / use the same
  prepareScriptWithHelpers shape the server uses. Filed for follow-up.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-26 16:53:25 +02:00

192 lines
12 KiB
Python
Raw Blame History

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