feat(mirror_export): write the project tree out as a browseable .iecst mirror
Phase 1 of the "CODESYS project as a filesystem you can edit + diff +
ai-tooling against" idea. Today the only way to read code in a project
is via a one-shot get_all_pou_code dump (one giant JSON) or by clicking
through the IDE; neither is friendly to AI-assisted edits, code review,
git diffs or external tooling. mirror_export walks the live project
tree and emits one .iecst file per code-bearing object, preserving the
project's folder structure as nested directories on disk.
What the tool does:
- Walks every node from script_engine.projects.primary.get_children()
recursively (depth-first).
- Structural nodes (Device, Application, Folder, ...) become
directories under MIRROR_ROOT.
- Code-bearing nodes (Program, FB, Function, Method, Property, DUT,
GVL, Interface, ...) become <name>.iecst files in their parent
directory; if a code-bearing node has child code objects (e.g. an
FB with methods), those children land in a sibling subdirectory
with the parent's name.
- File header: `(* === CODESYS export -- KIND === *)` + project path
+ generated timestamp, so a future write-back tool can map each
file back to set_pou_code's pouPath.
- Body: declaration block, then `(* === IMPLEMENTATION === *)`
separator (when both are present), then implementation block.
Defaults the mirror root to `<projectDir>/MCP/mirror` so it lands next
to the existing library.md / pou-dump.md if the user has been
following the same folder convention.
Implementation details that would have bitten without testing:
- UTF-8 output via codecs.open. CODESYS POU text occasionally
contains non-ASCII (smart quotes from copy-paste, degree signs,
etc.). IronPython 2.7's builtin open() defaults to ASCII and
would raise; saw it on X33's ST_HetronicIn2 (smart quote) until
fixed.
- Filesystem-illegal characters in CODESYS object names (`/`, `\`,
`<>:"|?*`) replaced with `_`. CODESYS lets you name a folder
"Remote / Hetronic"; on Windows that splits as two folders with
naive os.path.join.
- Kind classifier strips leading `//` and `(* *)` comments and
`{attribute := '...'}` pragmas before matching the IEC keyword,
otherwise GVLs decorated with attribute pragmas got bucketed as
OTHER (X33 had 2 of these; first try misclassified them).
Verified against MRCodesysX33_0021 (X33): 91 files, 254 KB, 0 errors,
preserves the full tree (MainPLC/Plc Logic/Application/MRLib/...).
Phase 2 (write-back via a `sync_pou_from_file` tool that reads a
mirror file, splits decl/impl on the IMPLEMENTATION separator, and
calls set_pou_code) is out of scope for this commit.
This commit is contained in:
parent
9b766c8b6b
commit
7a6e7254a6
2 changed files with 232 additions and 0 deletions
202
src/scripts/mirror_export.py
Normal file
202
src/scripts/mirror_export.py
Normal file
|
|
@ -0,0 +1,202 @@
|
||||||
|
import sys, scriptengine as script_engine, os, traceback, time, codecs
|
||||||
|
|
||||||
|
# Mirrors the CODESYS project tree into a filesystem layout under MIRROR_ROOT
|
||||||
|
# so the project becomes browseable / diffable / AI-editable as plain text.
|
||||||
|
#
|
||||||
|
# - Structural nodes (Device, Application, Folder, ...) become directories.
|
||||||
|
# - Code-bearing nodes (Program, FB, Function, Method, Property, DUT, GVL,
|
||||||
|
# Interface, ...) become .iecst files in their parent directory.
|
||||||
|
# - If a code-bearing node has child code objects (e.g. an FB with methods)
|
||||||
|
# those children land in a sibling subdirectory with the parent's name.
|
||||||
|
# - Filesystem-illegal characters in CODESYS object names are replaced with
|
||||||
|
# '_'; the original CODESYS project path is recorded as a header comment
|
||||||
|
# in each file so a future write-back tool can map it back to set_pou_code's
|
||||||
|
# pouPath.
|
||||||
|
#
|
||||||
|
# Phase 1: read-only export. No write-back here.
|
||||||
|
|
||||||
|
MIRROR_ROOT = r"{MIRROR_ROOT}"
|
||||||
|
ILLEGAL = '<>:"|?*'
|
||||||
|
|
||||||
|
|
||||||
|
def sanitise(name):
|
||||||
|
s = (name or '').replace('/', '_').replace('\\', '_')
|
||||||
|
for c in ILLEGAL:
|
||||||
|
s = s.replace(c, '_')
|
||||||
|
s = s.strip().rstrip('.')
|
||||||
|
return s if s else '_unnamed_'
|
||||||
|
|
||||||
|
|
||||||
|
def _strip_leading_noise(decl):
|
||||||
|
"""Drop leading whitespace, // and (* *) comments, and {attribute := ''}
|
||||||
|
pragmas so the kind classifier matches the actual IEC keyword."""
|
||||||
|
s = decl
|
||||||
|
changed = True
|
||||||
|
while changed:
|
||||||
|
changed = False
|
||||||
|
s2 = s.lstrip()
|
||||||
|
if s2 != s:
|
||||||
|
s = s2
|
||||||
|
changed = True
|
||||||
|
if s.startswith('//'):
|
||||||
|
nl = s.find('\n')
|
||||||
|
s = s[nl + 1:] if nl >= 0 else ''
|
||||||
|
changed = True
|
||||||
|
continue
|
||||||
|
if s.startswith('(*'):
|
||||||
|
end = s.find('*)')
|
||||||
|
s = s[end + 2:] if end >= 0 else ''
|
||||||
|
changed = True
|
||||||
|
continue
|
||||||
|
if s.startswith('{'):
|
||||||
|
end = s.find('}')
|
||||||
|
s = s[end + 1:] if end >= 0 else ''
|
||||||
|
changed = True
|
||||||
|
continue
|
||||||
|
return s
|
||||||
|
|
||||||
|
|
||||||
|
def classify(decl):
|
||||||
|
if not decl:
|
||||||
|
return 'UNKNOWN'
|
||||||
|
head = _strip_leading_noise(decl).upper()
|
||||||
|
if head.startswith('TYPE'):
|
||||||
|
return 'DUT'
|
||||||
|
if head.startswith('VAR_GLOBAL'):
|
||||||
|
return 'GVL'
|
||||||
|
if head.startswith('PROGRAM'):
|
||||||
|
return 'PROGRAM'
|
||||||
|
if head.startswith('FUNCTION_BLOCK'):
|
||||||
|
return 'FB'
|
||||||
|
if head.startswith('FUNCTION'):
|
||||||
|
return 'FUNCTION'
|
||||||
|
if head.startswith('METHOD'):
|
||||||
|
return 'METHOD'
|
||||||
|
if head.startswith('PROPERTY'):
|
||||||
|
return 'PROPERTY'
|
||||||
|
if head.startswith('INTERFACE'):
|
||||||
|
return 'INTERFACE'
|
||||||
|
return 'OTHER'
|
||||||
|
|
||||||
|
|
||||||
|
def get_text(obj, attr):
|
||||||
|
if not hasattr(obj, attr):
|
||||||
|
return ''
|
||||||
|
try:
|
||||||
|
x = getattr(obj, attr)
|
||||||
|
if x and hasattr(x, 'text'):
|
||||||
|
return x.text or ''
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return ''
|
||||||
|
|
||||||
|
|
||||||
|
def write_one(parent_dir, name, decl, impl, project_path):
|
||||||
|
if not os.path.exists(parent_dir):
|
||||||
|
os.makedirs(parent_dir)
|
||||||
|
kind = classify(decl)
|
||||||
|
fname = sanitise(name) + '.iecst'
|
||||||
|
fpath = os.path.join(parent_dir, fname)
|
||||||
|
|
||||||
|
lines = []
|
||||||
|
lines.append(u'(* === CODESYS export -- %s === *)' % kind)
|
||||||
|
lines.append(u'(* Project path: %s *)' % project_path)
|
||||||
|
lines.append(u'(* Generated: %s *)' % time.strftime('%Y-%m-%d %H:%M:%S'))
|
||||||
|
lines.append(u'')
|
||||||
|
if decl:
|
||||||
|
lines.append(decl.rstrip())
|
||||||
|
lines.append(u'')
|
||||||
|
if impl:
|
||||||
|
if decl:
|
||||||
|
lines.append(u'(* ============ IMPLEMENTATION ============ *)')
|
||||||
|
lines.append(u'')
|
||||||
|
lines.append(impl.rstrip())
|
||||||
|
lines.append(u'')
|
||||||
|
|
||||||
|
# UTF-8 because CODESYS POU text occasionally contains non-ASCII (smart
|
||||||
|
# quotes, degree signs, etc.). IronPython 2.7's builtin open() defaults to
|
||||||
|
# ASCII and would raise.
|
||||||
|
f = codecs.open(fpath, 'w', encoding='utf-8')
|
||||||
|
try:
|
||||||
|
f.write(u'\n'.join(unicode(l) for l in lines))
|
||||||
|
finally:
|
||||||
|
f.close()
|
||||||
|
return fpath, kind, os.path.getsize(fpath)
|
||||||
|
|
||||||
|
|
||||||
|
def walk(node, parent_fs_dir, parent_proj_path, stats):
|
||||||
|
try:
|
||||||
|
gn = getattr(node, 'get_name', None)
|
||||||
|
name = gn() if gn else '?'
|
||||||
|
except Exception:
|
||||||
|
name = '?'
|
||||||
|
safe_name = sanitise(name)
|
||||||
|
proj_path = (parent_proj_path + '/' + name) if parent_proj_path else name
|
||||||
|
|
||||||
|
decl = get_text(node, 'textual_declaration')
|
||||||
|
impl = get_text(node, 'textual_implementation')
|
||||||
|
|
||||||
|
if decl or impl:
|
||||||
|
try:
|
||||||
|
fpath, kind, size = write_one(parent_fs_dir, name, decl, impl, proj_path)
|
||||||
|
stats['files'].append({'path': fpath, 'project_path': proj_path, 'kind': kind, 'bytes': size})
|
||||||
|
except Exception as e:
|
||||||
|
stats['errors'].append({'project_path': proj_path, 'error': str(e)})
|
||||||
|
|
||||||
|
new_dir = os.path.join(parent_fs_dir, safe_name)
|
||||||
|
try:
|
||||||
|
children = list(node.get_children(False))
|
||||||
|
except Exception:
|
||||||
|
children = []
|
||||||
|
if children:
|
||||||
|
if not os.path.exists(new_dir):
|
||||||
|
try:
|
||||||
|
os.makedirs(new_dir)
|
||||||
|
stats['dirs_created'] += 1
|
||||||
|
except Exception as e:
|
||||||
|
stats['errors'].append({'project_path': proj_path, 'error': 'mkdir: %s' % e})
|
||||||
|
return
|
||||||
|
for c in children:
|
||||||
|
walk(c, new_dir, proj_path, stats)
|
||||||
|
|
||||||
|
|
||||||
|
try:
|
||||||
|
print("DEBUG: mirror_export: Project='%s' MirrorRoot='%s'" % (PROJECT_FILE_PATH, MIRROR_ROOT))
|
||||||
|
primary_project = ensure_project_open(PROJECT_FILE_PATH)
|
||||||
|
|
||||||
|
if not MIRROR_ROOT.strip():
|
||||||
|
raise ValueError("MIRROR_ROOT is empty -- pass mirrorRoot to the tool or rely on the default '<projectDir>/MCP/mirror'.")
|
||||||
|
|
||||||
|
if not os.path.exists(MIRROR_ROOT):
|
||||||
|
os.makedirs(MIRROR_ROOT)
|
||||||
|
|
||||||
|
stats = {'files': [], 'dirs_created': 0, 'errors': []}
|
||||||
|
|
||||||
|
for child in primary_project.get_children(False):
|
||||||
|
walk(child, MIRROR_ROOT, '', stats)
|
||||||
|
|
||||||
|
by_kind = {}
|
||||||
|
total_bytes = 0
|
||||||
|
for entry in stats['files']:
|
||||||
|
by_kind[entry['kind']] = by_kind.get(entry['kind'], 0) + 1
|
||||||
|
total_bytes += entry['bytes']
|
||||||
|
|
||||||
|
print("--- Mirror summary ---")
|
||||||
|
print("Files written: %d" % len(stats['files']))
|
||||||
|
print("Directories made: %d" % stats['dirs_created'])
|
||||||
|
print("Total bytes: %d" % total_bytes)
|
||||||
|
print("By kind:")
|
||||||
|
for k in sorted(by_kind.keys()):
|
||||||
|
print(" %-10s %d" % (k, by_kind[k]))
|
||||||
|
if stats['errors']:
|
||||||
|
print("Errors: %d" % len(stats['errors']))
|
||||||
|
for er in stats['errors'][:10]:
|
||||||
|
print(" %s -> %s" % (er.get('project_path', '?'), er.get('error', '?')))
|
||||||
|
print("SCRIPT_SUCCESS: mirror exported to %s" % MIRROR_ROOT)
|
||||||
|
sys.exit(0)
|
||||||
|
except Exception as e:
|
||||||
|
msg = "Error in mirror_export for project '%s': %s\n%s" % (
|
||||||
|
PROJECT_FILE_PATH, e, traceback.format_exc())
|
||||||
|
print(msg)
|
||||||
|
print("SCRIPT_ERROR: %s" % msg)
|
||||||
|
sys.exit(1)
|
||||||
|
|
@ -1346,6 +1346,36 @@ export async function startMcpServer(config: ServerConfig): Promise<void> {
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
|
|
||||||
|
// ─── Filesystem mirror (Phase 1: read-only export) ────────────────────
|
||||||
|
|
||||||
|
s.tool(
|
||||||
|
'mirror_export',
|
||||||
|
"Walks the CODESYS project tree and writes one .iecst file per code-bearing object into a filesystem mirror, preserving the project tree as nested directories. Programs / Function Blocks / Functions / Methods / Properties / DUTs / GVLs / Interfaces all become text files; structural nodes (Devices, Applications, Folders) become directories. Each file carries a header comment with its original CODESYS project path so a future write-back tool can map it back to set_pou_code's pouPath. Read-only -- does NOT modify the CODESYS project. UTF-8 output. If mirrorRoot is omitted, defaults to '<projectDir>/MCP/mirror'.",
|
||||||
|
{
|
||||||
|
projectFilePath: z.string().describe("Path to the project file."),
|
||||||
|
mirrorRoot: z.string().optional().describe("Filesystem path where the mirror tree gets written. If omitted, defaults to '<projectDir>/MCP/mirror'. Created automatically if missing; existing files at the same paths are overwritten."),
|
||||||
|
},
|
||||||
|
async (args: { projectFilePath: string; mirrorRoot?: string }) => {
|
||||||
|
const escaped = resolvePath(args.projectFilePath, workspaceDir);
|
||||||
|
const mirrorRoot = args.mirrorRoot
|
||||||
|
? resolvePath(args.mirrorRoot, workspaceDir)
|
||||||
|
: path.join(path.dirname(escaped), 'MCP', 'mirror');
|
||||||
|
const script = scriptManager.prepareScriptWithHelpers(
|
||||||
|
'mirror_export',
|
||||||
|
{
|
||||||
|
PROJECT_FILE_PATH: escaped,
|
||||||
|
MIRROR_ROOT: mirrorRoot,
|
||||||
|
},
|
||||||
|
['ensure_project_open']
|
||||||
|
);
|
||||||
|
const result = await executor.executeScript(script);
|
||||||
|
return formatToolResponse(
|
||||||
|
result,
|
||||||
|
`mirror_export complete for ${args.projectFilePath} -> ${mirrorRoot}.`
|
||||||
|
);
|
||||||
|
}
|
||||||
|
);
|
||||||
|
|
||||||
// ─── Resources ───────────────────────────────────────────────────────
|
// ─── Resources ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
server.resource(
|
server.resource(
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue