Codesys-MCP/python_scripting/windows_codesys_agent/scripts/write_pou.py

#!/usr/bin/env python
"""Write/update textual objects (POU, FB, GVL, NVL, struct, program) in a CODESYS project."""

from __future__ import print_function

import os

from _common import emit, load_payload


def _discover_project_file(path_value):
    if os.path.isfile(path_value):
        lower_path = path_value.lower()
        if lower_path.endswith(".project") or lower_path.endswith(".projectarchive"):
            return path_value
        raise RuntimeError("project_path must be .project or .projectarchive: {0}".format(path_value))
    if os.path.isdir(path_value):
        candidates = []
        for entry in os.listdir(path_value):
            full_path = os.path.join(path_value, entry)
            if os.path.isfile(full_path):
                lower_name = entry.lower()
                if lower_name.endswith(".project") or lower_name.endswith(".projectarchive"):
                    candidates.append(full_path)
        if len(candidates) == 1:
            return candidates[0]
        if not candidates:
            raise RuntimeError("No .project/.projectarchive in: {0}".format(path_value))
        raise RuntimeError("Multiple project files: {0}".format(", ".join(candidates)))
    raise RuntimeError("Path does not exist: {0}".format(path_value))


def _safe_str(val):
    try:
        return "{0}".format(val)
    except Exception:
        return "?"


def _name_of(obj):
    try:
        return obj.get_name(False)
    except Exception:
        try:
            return obj.get_name()
        except Exception:
            return _safe_str(obj)


def _children_of(obj):
    try:
        return list(obj.get_children(False))
    except Exception:
        try:
            return list(obj.get_children())
        except Exception:
            return []


def _find_by_path(root, path_str):
    """Walk slash-separated path from project root to find the target object."""
    parts = [p.strip() for p in path_str.split("/") if p.strip()]
    current = root
    for i, part in enumerate(parts):
        found = None
        for child in _children_of(current):
            if _name_of(child) == part:
                found = child
                break
        if found is None:
            raise RuntimeError("Object not found at '{0}' (segment {1} of '{2}')".format(
                part, i, path_str))
        current = found
    return current


def _find_by_name(root, name):
    """Use project.find() with recursive search."""
    results = root.find(name, True)
    if not results:
        raise RuntimeError("Object '{0}' not found in project.".format(name))
    if len(results) > 1:
        paths = []
        for r in results:
            try:
                paths.append(_name_of(r))
            except Exception:
                paths.append("?")
        raise RuntimeError("Multiple objects named '{0}': {1}. Use object_path instead.".format(
            name, ", ".join(paths)))
    return results[0]


def _set_text(doc, new_text):
    """Replace full text of a ScriptTextDocument."""
    try:
        doc.replace(new_text=new_text)
    except TypeError:
        length = doc.length
        if length > 0:
            doc.remove(offset=0, length=length)
        if new_text:
            doc.append(new_text)


def main():
    payload = load_payload()
    project_path = payload.get("project_path")
    if not project_path:
        raise ValueError("project_path is required.")

    updates = payload.get("updates", [])
    if not updates:
        single = {}
        for key in ("object_path", "object_name", "declaration", "implementation"):
            if key in payload:
                single[key] = payload[key]
        if single:
            updates = [single]

    if not updates:
        raise ValueError("Provide 'updates' list or single object_path/object_name + declaration/implementation.")

    dry_run = bool(payload.get("dry_run", False))

    pp = os.path.abspath(os.path.expanduser(project_path))
    pp = _discover_project_file(pp)

    if "projects" not in globals():
        raise RuntimeError("Run through CODESYS ScriptEngine (--runscript).")

    if projects.primary:
        projects.primary.close()

    proj = projects.open(pp, primary=True)

    results = []
    warnings = []

    for upd in updates:
        obj_path = upd.get("object_path", "")
        obj_name = upd.get("object_name", "")
        decl_text = upd.get("declaration")
        impl_text = upd.get("implementation")

        if not obj_path and not obj_name:
            warnings.append("update skipped: no object_path or object_name")
            continue

        try:
            if obj_path:
                obj = _find_by_path(proj, obj_path)
            else:
                obj = _find_by_name(proj, obj_name)

            name = _name_of(obj)
            entry = {
                "name": name,
                "object_path": obj_path or obj_name,
                "has_declaration": getattr(obj, "has_textual_declaration", False),
                "has_implementation": getattr(obj, "has_textual_implementation", False),
                "declaration_updated": False,
                "implementation_updated": False,
            }

            if decl_text is not None:
                if not getattr(obj, "has_textual_declaration", False):
                    warnings.append("{0}: object has no textual declaration".format(name))
                elif dry_run:
                    entry["declaration_updated"] = "would_update"
                else:
                    _set_text(obj.textual_declaration, decl_text)
                    entry["declaration_updated"] = True

            if impl_text is not None:
                if not getattr(obj, "has_textual_implementation", False):
                    warnings.append("{0}: object has no textual implementation".format(name))
                elif dry_run:
                    entry["implementation_updated"] = "would_update"
                else:
                    _set_text(obj.textual_implementation, impl_text)
                    entry["implementation_updated"] = True

            results.append(entry)
        except Exception as exc:
            warnings.append("update_failed({0}): {1}".format(obj_path or obj_name, exc))

    if not dry_run:
        try:
            proj.save()
        except Exception as exc:
            warnings.append("project_save_failed: {0}".format(exc))

    proj.close()

    emit({
        "ok": True,
        "action": "write_pou",
        "project_path": pp,
        "dry_run": dry_run,
        "results": results,
        "result_count": len(results),
        "warnings": warnings,
    })


if __name__ == "__main__":
    main()