557 lines
22 KiB
Python
557 lines
22 KiB
Python
# 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<project>[^\n]*?)[ \t]*\n"
|
||
r"[ \t]*Device[ \t]*:[ \t]*(?P<device>[^\n]*?)[ \t]*\n"
|
||
r"[ \t]*Path[ \t]*:[ \t]*(?P<path>[^\n]*?)[ \t]*\n"
|
||
r"[ \t]*Name[ \t]*:[ \t]*(?P<name>[^\n]*?)[ \t]*\n"
|
||
r"[ \t]*Type[ \t]*:[ \t]*(?P<type>[^\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)
|