# 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 = True. # # ПРЕДУПРЕЖДЕНИЕ: сигнатуры 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", } # Терпимо к вариациям пробелов/кол-ва дефисов, которые может внести # автоформатирование редактора, напр. "(* ----- IMPLEMENTATION ----- *)" # вместо канонического "(*----- IMPLEMENTATION -----*)". IMPL_MARKER_RE = re.compile(r"\(\*\s*-{3,}\s*IMPLEMENTATION\s*-{3,}\s*\*\)") HEADER_RE = re.compile( r"^\(\*[ \t]*\n" r"=+[ \t]*\n" r"[ \t]*Project[ \t]*:[ \t]*(?P[^\n]*?)[ \t]*\n" r"[ \t]*Device[ \t]*:[ \t]*(?P[^\n]*?)[ \t]*\n" r"[ \t]*Path[ \t]*:[ \t]*(?P[^\n]*?)[ \t]*\n" r"[ \t]*Name[ \t]*:[ \t]*(?P[^\n]*?)[ \t]*\n" r"[ \t]*Type[ \t]*:[ \t]*(?P[^\n]*?)[ \t]*\n" r"=+[ \t]*\n" r"[ \t]*\*\)[ \t]*\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 # поставь True, дле тествового запуска без записи в проект, 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 "") m = IMPL_MARKER_RE.search(body) if m: decl, impl = body[:m.start()], body[m.end():] 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 extract_return_type(declaration, type_label, object_name): """Извлекает return type из FUNCTION/METHOD/PROPERTY declaration.""" if not declaration: return None text = normalize_newlines(declaration) # Удаляем только ведущие прагмы для анализа первой строки. while True: new_text = re.sub(r"^\s*\{[^}]*\}\s*", "", text, count=1) if new_text == text: break text = new_text for line in text.split("\n"): line = line.strip() if not line: continue # Для анализа нам не нужны inline block-comments. line = re.sub(r"\(\*.*?\*\)", "", line).strip() m = re.match( r"^" + re.escape(type_label) + r"\s+\S+\s*:\s*(.+?)\s*$", line, re.IGNORECASE ) if m: return m.group(1).strip() # Если это нужный заголовок, но return type отсутствует. if line.upper().startswith(type_label + " "): return None return 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, declaration_text): """Создаёт PROGRAM/FB/FUNCTION/INTERFACE через специализированный API.""" existing = find_child(parent, name) if existing is not None: print(" = %s '%s' уже существует" % (type_label, name)) return existing, False try: if type_label == "PROGRAM": create_program = getattr(parent, "create_program", None) if create_program is not None: return create_program(name, ImplementationLanguages.st), True return parent.create_pou(name, PouType.Program, ImplementationLanguages.st), True if type_label == "FUNCTION_BLOCK": create_fb = getattr(parent, "create_function_block", None) if create_fb is not None: return create_fb(name, ImplementationLanguages.st), True return parent.create_pou(name, PouType.FunctionBlock, ImplementationLanguages.st), True if type_label == "FUNCTION": return_type = extract_return_type(declaration_text, "FUNCTION", name) if not return_type: raise RuntimeError("не удалось определить return type FUNCTION '%s'" % name) print(" -> FUNCTION '%s' : %s" % (name, return_type)) create_function = getattr(parent, "create_function", None) if create_function is not None: return create_function(name, return_type, ImplementationLanguages.st), True # Fallback для старых версий API. try: return parent.create_pou(name, PouType.Function, return_type, ImplementationLanguages.st), True except TypeError: return parent.create_pou(name, PouType.Function, ImplementationLanguages.st, return_type), True if type_label == "INTERFACE": return parent.create_interface(name), True raise RuntimeError("неизвестный тип POU '%s'" % type_label) except Exception as e: print(" ! НЕ УДАЛОСЬ создать %s '%s': %s" % (type_label, name, e)) return None, False def get_or_create_method(parent_pou, name, declaration_text): existing = find_child(parent_pou, name) if existing is not None: print(" = METHOD '%s' уже существует" % name) return existing, False return_type = extract_return_type(declaration_text, "METHOD", name) try: return parent_pou.create_method( name, return_type, ImplementationLanguages.st ), True except TypeError: try: return parent_pou.create_method(name, return_type), True except Exception as e: print(" ! НЕ УДАЛОСЬ создать METHOD '%s': %s" % (name, e)) return None, False except Exception as e: print(" ! НЕ УДАЛОСЬ создать METHOD '%s': %s" % (name, e)) return None, False def get_or_create_property(parent_pou, name, declaration_text): existing = find_child(parent_pou, name) if existing is not None: print(" = PROPERTY '%s' уже существует" % name) return existing, False return_type = extract_return_type(declaration_text, "PROPERTY", name) if not return_type: print(" ! НЕ УДАЛОСЬ определить return type PROPERTY '%s'" % name) return None, False try: return parent_pou.create_property( name, return_type, ImplementationLanguages.st ), True except TypeError: try: return parent_pou.create_property(name, return_type), True except Exception as e: print(" ! НЕ УДАЛОСЬ создать PROPERTY '%s': %s" % (name, e)) return None, False 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: print(" = ACTION '%s' уже существует" % name) return existing, False try: return parent_pou.create_action(name, ImplementationLanguages.st), True except TypeError: try: return parent_pou.create_action(name), True except Exception as e: print(" ! НЕ УДАЛОСЬ создать ACTION '%s': %s" % (name, e)) return None, False 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, decl_text) 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, decl_text) else: obj, _ = get_or_create_property(parent_pou, name, decl_text) 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)