From 787f875581228e1252923a4275c0b5a724e811d8 Mon Sep 17 00:00:00 2001 From: Iaroshuk Dmitrii Date: Mon, 3 Aug 2026 11:46:24 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A1=D0=BA=D1=80=D0=B8=D0=BF=D1=82=20=D0=B8?= =?UTF-8?q?=D0=BC=D0=BF=D0=BE=D1=80=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Скрипты/import_plc_src.py | 464 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 464 insertions(+) create mode 100644 Скрипты/import_plc_src.py diff --git a/Скрипты/import_plc_src.py b/Скрипты/import_plc_src.py new file mode 100644 index 0000000..24aad02 --- /dev/null +++ b/Скрипты/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[^\n]*)\n" + r" Device\s*: (?P[^\n]*)\n" + r" Path\s*: (?P[^\n]*)\n" + r" Name\s*: (?P[^\n]*)\n" + r" Type\s*: (?P[^\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)