Скрипт импорта

Скрипты/import_plc_src.py

@@ -0,0 +1,464 @@
+# encoding:utf-8
+# import_plc_src.py
+# Обратный скрипт к export_plc_src.py
+# Читает .st файлы из plc_src/ и создаёт/обновляет соответствующие
+# POU / METHOD / PROPERTY / ACTION / GVL / DUT в открытом проекте CODESYS.
+#
+# ВАЖНО про ACTION / END_ACTION / END_PROGRAM и т.п.:
+# CODESYS сам оборачивает textual_declaration / textual_implementation в
+# неявный блок в зависимости от типа объекта. Заголовок "ACTION Имя :" и
+# закрывающие END_* НЕ нужно (и часто нельзя) писать в текст через API --
+# они генерируются автоматически самим объектом. Поэтому при импорте эти
+# строки сначала вырезаются (см. strip_end_keyword / parse_block_text),
+# и в API передаётся только "голое" тело.
+#
+# Запуск: аналогично export_plc_src.py --
+#   Tools -> Scripting -> Execute Script File, либо --runscript=...
+#
+# ПЕРЕД РЕАЛЬНЫМ ИМПОРТОМ: скрипт по умолчанию работает в режиме DRY_RUN
+# (см. ниже) -- он только печатает, что бы сделал, ничего не создавая и
+# не перезаписывая. Проверь вывод, затем поставь DRY_RUN = False.
+#
+# ПРЕДУПРЕЖДЕНИЕ: сигнатуры create_pou / create_method / create_property /
+# create_action / create_global_var_list / create_dut могут отличаться
+# между версиями CODESYS. Значения ниже рассчитаны на типовой scripting
+# API (аналог CODESYS V3.5 SP17). Если создание объекта падает с ошибкой --
+# смотри Tools -> Scripting -> справку по API твоей версии и поправь
+# соответствующую get_or_create_* функцию.
+
+from __future__ import print_function
+import os
+import re
+import codecs
+
+# Должно совпадать с EXCLUDED_PATH_NAMES в export_plc_src.py --
+# именно эти узлы дерева были "прозрачно" пропущены при формировании Path.
+EXCLUDED_PATH_NAMES = set(["application", "plc logic"])
+
+END_KEYWORDS = {
+    "PROGRAM": "END_PROGRAM",
+    "FUNCTION_BLOCK": "END_FUNCTION_BLOCK",
+    "FUNCTION": "END_FUNCTION",
+    "INTERFACE": "END_INTERFACE",
+    "METHOD": "END_METHOD",
+    "PROPERTY": "END_PROPERTY",
+    "ACTION": "END_ACTION",
+}
+
+IMPL_MARKER = "(*----- IMPLEMENTATION -----*)"
+
+HEADER_RE = re.compile(
+    r"^\(\*\n"
+    r"=+\n"
+    r" Project\s*: (?P<project>[^\n]*)\n"
+    r" Device\s*: (?P<device>[^\n]*)\n"
+    r" Path\s*: (?P<path>[^\n]*)\n"
+    r" Name\s*: (?P<name>[^\n]*)\n"
+    r" Type\s*: (?P<type>[^\n]*)\n"
+    r"=+\n"
+    r"\*\)\n",
+    re.MULTILINE,
+)
+
+
+def normalize_newlines(text):
+    """CODESYS/IronPython на Windows при записи через codecs.open иногда
+    транслирует каждый \\n в \\r\\n даже там, где \\r уже стоит вручную --
+    получается \\r\\r\\n вместо \\r\\n. Лишние оказываются именно \\r, а не
+    \\n, поэтому надёжная нормализация -- просто выбросить все \\r: структура
+    строк (в т.ч. пустые строки) при этом сохраняется, независимо от того,
+    что реально лежит на диске (\\r\\n, \\r\\r\\n, голый \\n, смесь)."""
+    return text.replace("\r", "")
+
+POU_TYPES = ("PROGRAM", "FUNCTION_BLOCK", "FUNCTION", "INTERFACE")
+
+# ==== НАСТРОЙКА ====
+DRY_RUN = False  # поставь False, когда убедишься, что вывод корректный
+MANUAL_INPUT_ROOT = r""  # пусто -- искать plc_src рядом с .project
+# =====================================================
+
+
+# ---------- вспомогательные функции (частично повторяют export_plc_src.py) ----------
+
+def get_project_file_path(proj):
+    for attr in ("path", "get_path", "project_path", "full_path"):
+        try:
+            val = getattr(proj, attr)
+            val = val() if callable(val) else val
+            if val:
+                return val
+        except Exception:
+            pass
+    try:
+        s = str(proj)
+        m = re.search(r"stPath=(.*?)\)", s)
+        if m:
+            return m.group(1)
+    except Exception:
+        pass
+    return None
+
+
+def safe_name(obj):
+    try:
+        return obj.get_name(False)
+    except Exception:
+        return "unnamed"
+
+
+def is_device_node(obj):
+    try:
+        return bool(getattr(obj, "is_device", False))
+    except Exception:
+        return False
+
+
+def find_child(node, name):
+    for c in node.get_children(False):
+        if safe_name(c) == name:
+            return c
+    return None
+
+
+def find_device(proj, device_name):
+    for c in proj.get_children(False):
+        if is_device_node(c) and safe_name(c) == device_name:
+            return c
+    return None
+
+
+def find_entry_container(root):
+    """Спускается через служебные узлы (Application / PLC Logic и т.п.) --
+    те же имена, что фильтровались при построении Path на экспорте."""
+    node = root
+    while True:
+        matched = None
+        for c in node.get_children(False):
+            if safe_name(c).strip().lower() in EXCLUDED_PATH_NAMES:
+                matched = c
+                break
+        if matched is None:
+            return node
+        node = matched
+
+
+def get_or_create_folder(parent, name):
+    child = find_child(parent, name)
+    if child is not None:
+        return child
+    try:
+        parent.create_folder(name, "")
+    except TypeError:
+        parent.create_folder(name)
+    # create_folder может не возвращать созданный объект (возвращает None) --
+    # поэтому вместо доверия return-значению ищем папку заново по имени.
+    child = find_child(parent, name)
+    if child is None:
+        raise RuntimeError("create_folder('%s') не создал папку и не найден через find_child" % name)
+    return child
+
+
+def navigate_path(container, path_parts):
+    node = container
+    for part in path_parts:
+        node = get_or_create_folder(node, part)
+    return node
+
+
+# ---------- разбор .st файла на блоки ----------
+
+def strip_end_keyword(text, type_label):
+    """Убирает END_* в конце текста -- CODESYS добавляет его сам, в API
+    он не передаётся. Работает с уже нормализованным (\\n) текстом."""
+    kw = END_KEYWORDS.get(type_label)
+    if not kw:
+        return text
+    pattern = re.compile(r"\n\s*" + re.escape(kw) + r"\s*\n?\s*$")
+    new_text, n = pattern.subn("", text, count=1)
+    return new_text if n else text
+
+
+def split_blocks(raw_text):
+    """raw_text уже нормализован (только \\n). Один .st файл может содержать
+    несколько блоков: основной POU и дописанные METHOD/PROPERTY/ACTION.
+    Разбиваем по заголовкам-паспортам."""
+    matches = list(HEADER_RE.finditer(raw_text))
+    blocks = []
+    for i, m in enumerate(matches):
+        start = m.end()
+        end = matches[i + 1].start() if i + 1 < len(matches) else len(raw_text)
+        body = raw_text[start:end]
+        blocks.append((m.groupdict(), body))
+    return blocks
+
+
+def to_crlf(text):
+    """CODESYS ожидает обычный текст -- перед передачей в API возвращаем
+    стандартный \\r\\n (на входе текст уже нормализован в \\n)."""
+    if not text:
+        return text
+    return text.replace("\n", "\r\n")
+
+
+def parse_block_text(body, type_label):
+    """Возвращает (declaration_text_or_None, implementation_text_or_None)
+    -- зеркало build_pou_text / build_action_text из export_plc_src.py.
+    body уже нормализован (только \\n)."""
+    body = body.strip("\n")
+    if not body:
+        return None, None
+    body = strip_end_keyword(body, type_label)
+
+    if type_label == "ACTION":
+        # первая строка "ACTION Имя :" синтезирована на экспорте вручную --
+        # убираем её, для API нужна только implementation
+        parts = body.split("\n", 1)
+        impl = parts[1].strip("\n") if len(parts) > 1 else ""
+        return None, (to_crlf(impl) + "\r\n" if impl else "")
+
+    if IMPL_MARKER in body:
+        decl, impl = body.split(IMPL_MARKER, 1)
+        decl = decl.strip("\n")
+        impl = impl.strip("\n")
+        return (to_crlf(decl) + "\r\n" if decl else ""), (to_crlf(impl) + "\r\n" if impl else "")
+
+    # GVL / DUT -- только декларация, END_* для них не определён (и не нужен)
+    decl = body.strip("\n")
+    return (to_crlf(decl) + "\r\n" if decl else ""), None
+
+
+# ---------- запись в объекты проекта ----------
+
+def set_declaration(obj, text):
+    if not text:
+        return
+    try:
+        obj.textual_declaration.replace(text)
+    except Exception as e:
+        print("  ! ошибка записи declaration: %s" % e)
+
+
+def set_implementation(obj, text):
+    if not text:
+        return
+    try:
+        obj.textual_implementation.replace(text)
+    except Exception as e:
+        print("  ! ошибка записи implementation: %s" % e)
+
+
+def get_or_create_pou(parent, name, type_label):
+    existing = find_child(parent, name)
+    if existing is not None:
+        return existing, False
+    try:
+        if type_label == "INTERFACE":
+            obj = parent.create_interface(name)
+        else:
+            pou_kind = {
+                "PROGRAM": "Program",
+                "FUNCTION_BLOCK": "FunctionBlock",
+                "FUNCTION": "Function",
+            }[type_label]
+            try:
+                pou_type_enum = getattr(PouType, pou_kind)  # noqa: F821 (глобал CODESYS)
+            except Exception:
+                pou_type_enum = pou_kind
+            obj = parent.create_pou(name, pou_type_enum, ImplementationLanguages.st)  # noqa: F821
+        return obj, True
+    except Exception as e:
+        print("  ! не удалось создать %s '%s': %s" % (type_label, name, e))
+        return None, False
+
+
+def get_or_create_method(parent_pou, name):
+    existing = find_child(parent_pou, name)
+    if existing is not None:
+        return existing, False
+    try:
+        obj = parent_pou.create_method(name, "", "", ImplementationLanguages.st)  # noqa: F821
+        return obj, True
+    except Exception as e:
+        print("  ! не удалось создать METHOD '%s': %s" % (name, e))
+        return None, False
+
+
+def get_or_create_property(parent_pou, name):
+    existing = find_child(parent_pou, name)
+    if existing is not None:
+        return existing, False
+    try:
+        obj = parent_pou.create_property(name, "", ImplementationLanguages.st)  # noqa: F821
+        return obj, True
+    except Exception as e:
+        print("  ! не удалось создать PROPERTY '%s': %s" % (name, e))
+        return None, False
+
+
+def get_or_create_action(parent_pou, name):
+    existing = find_child(parent_pou, name)
+    if existing is not None:
+        return existing, False
+    try:
+        obj = parent_pou.create_action(name, ImplementationLanguages.st)  # noqa: F821
+        return obj, True
+    except Exception as e:
+        print("  ! не удалось создать ACTION '%s': %s" % (name, e))
+        return None, False
+
+
+def get_or_create_gvl(parent, name):
+    existing = find_child(parent, name)
+    if existing is not None:
+        return existing, False
+    for method_name in ("create_gvl", "create_global_variable_list", "create_global_var_list"):
+        method = getattr(parent, method_name, None)
+        if method is None:
+            continue
+        try:
+            obj = method(name)
+            return obj, True
+        except Exception as e:
+            print("  ! не удалось создать GVL '%s' через %s: %s" % (name, method_name, e))
+    print("  ! не удалось создать GVL '%s': подходящий метод API не найден" % name)
+    return None, False
+
+
+def get_or_create_dut(parent, name):
+    existing = find_child(parent, name)
+    if existing is not None:
+        return existing, False
+    try:
+        obj = parent.create_dut(name, DutType.Structure)  # noqa: F821 -- перезапишется текстом ниже
+        return obj, True
+    except Exception as e:
+        print("  ! не удалось создать DUT '%s': %s" % (name, e))
+        return None, False
+
+
+# ---------- обработка одного файла ----------
+
+def process_st_file(proj, filepath, dry_run):
+    with codecs.open(filepath, "r", encoding="utf-8") as f:
+        raw = f.read()
+    raw = normalize_newlines(raw)
+
+    blocks = split_blocks(raw)
+    if not blocks:
+        print("  (пропущено, заголовок не найден)")
+        return
+
+    parent_pou = None  # POU, к которому относятся идущие следом METHOD/PROPERTY/ACTION
+
+    for header, body in blocks:
+        device_name = header["device"].strip()
+        path_str = header["path"].strip()
+        name = header["name"].strip()
+        type_label = header["type"].strip()
+        path_parts = [] if path_str == "-" else path_str.split("/")
+
+        if device_name == "-":
+            root = proj
+        else:
+            root = find_device(proj, device_name)
+            if root is None:
+                print("  ! устройство '%s' не найдено, пропуск '%s'" % (device_name, name))
+                continue
+
+        container = find_entry_container(root)
+        target_parent = navigate_path(container, path_parts)
+        decl_text, impl_text = parse_block_text(body, type_label)
+
+        print("  %s: %s/%s" % (type_label, path_str, name))
+        if dry_run:
+            continue
+
+        if type_label in POU_TYPES:
+            obj, _ = get_or_create_pou(target_parent, name, type_label)
+            if obj is None:
+                parent_pou = None
+                continue
+            set_declaration(obj, decl_text)
+            set_implementation(obj, impl_text)
+            parent_pou = obj
+
+        elif type_label in ("METHOD", "PROPERTY"):
+            if parent_pou is None:
+                print("    ! нет родительского POU в этом файле, пропуск")
+                continue
+            if type_label == "METHOD":
+                obj, _ = get_or_create_method(parent_pou, name)
+            else:
+                obj, _ = get_or_create_property(parent_pou, name)
+            if obj is None:
+                continue
+            set_declaration(obj, decl_text)
+            set_implementation(obj, impl_text)
+
+        elif type_label == "ACTION":
+            if parent_pou is None:
+                print("    ! нет родительского POU в этом файле, пропуск")
+                continue
+            obj, _ = get_or_create_action(parent_pou, name)
+            if obj is None:
+                continue
+            set_implementation(obj, impl_text)
+
+        elif type_label == "GVL":
+            obj, _ = get_or_create_gvl(target_parent, name)
+            if obj is None:
+                continue
+            set_declaration(obj, decl_text)
+
+        elif type_label == "DUT":
+            obj, _ = get_or_create_dut(target_parent, name)
+            if obj is None:
+                continue
+            set_declaration(obj, decl_text)
+
+        else:
+            print("    ! неизвестный тип '%s', пропуск" % type_label)
+
+
+def find_st_files(root):
+    result = []
+    for dirpath, _dirnames, filenames in os.walk(root):
+        for fn in filenames:
+            if fn.lower().endswith(".st"):
+                result.append(os.path.join(dirpath, fn))
+    return sorted(result)
+
+
+# ==== ТОЧКА ВХОДА ====
+
+proj = projects.primary  # noqa: F821 -- глобал CODESYS scripting
+
+if proj is None:
+    print("ОШИБКА: нет открытого проекта. Открой .project перед запуском скрипта.")
+else:
+    project_file = get_project_file_path(proj)
+
+    if MANUAL_INPUT_ROOT:
+        INPUT_ROOT = MANUAL_INPUT_ROOT
+        print("Папка ввода задана вручную: %s" % INPUT_ROOT)
+    elif project_file:
+        INPUT_ROOT = os.path.join(os.path.dirname(project_file), "plc_src")
+        print("Проект найден: %s" % project_file)
+        print("Папка ввода определена автоматически: %s" % INPUT_ROOT)
+    else:
+        INPUT_ROOT = None
+        print("ОШИБКА: не удалось определить путь к .project. Задай MANUAL_INPUT_ROOT.")
+
+    if INPUT_ROOT and os.path.isdir(INPUT_ROOT):
+        print("--- Импорт из %s (DRY_RUN=%s) ---" % (INPUT_ROOT, DRY_RUN))
+        files = find_st_files(INPUT_ROOT)
+        print("Найдено файлов: %d" % len(files))
+        for fp in files:
+            print("Файл: %s" % fp)
+            process_st_file(proj, fp, DRY_RUN)
+        print("--- Импорт завершён ---")
+        if DRY_RUN:
+            print("Это был DRY-RUN -- ничего не записано в проект.")
+            print("Проверь вывод выше и поставь DRY_RUN = False для реального импорта.")
+    elif INPUT_ROOT:
+        print("ОШИБКА: папка не найдена: %s" % INPUT_ROOT)