410 lines
16 KiB
Python
410 lines
16 KiB
Python
# encoding:utf-8
|
||
# export_plc_src.py
|
||
# Экспорт POU / GVL / DUT из открытого проекта CODESYS в текстовые файлы
|
||
# plc_src/
|
||
# ├── PLC1/
|
||
# │ ├── POU/...
|
||
# │ ├── GVL/...
|
||
# │ └── DUT/...
|
||
# ├── PLC2/
|
||
# │ ├── POU/...
|
||
# │ ├── GVL/...
|
||
# │ └── DUT/...
|
||
# └── _common/
|
||
# │ ├── POU/...
|
||
# │ ├── GVL/...
|
||
# │ └── DUT/..
|
||
#
|
||
# Методы (METHOD) и действия (ACTION) не создают отдельных файлов --
|
||
# они дописываются в конец файла того POU/FB, в котором были созданы
|
||
#
|
||
# Запуск:
|
||
# 1) Внутри CODESYS: Tools -> Scripting -> Execute Script File -> выбрать этот файл
|
||
# 2) Из командной строки (без открытия GUI):
|
||
# "C:\Program Files (x86)\CODESYS 3.5.17.30\CODESYS\Common\CODESYS.exe" ^
|
||
# --Profile="CODESYS V3.5 SP17 Patch 3" --runscript="export_plc_src.py" ^
|
||
# --project="D:\path\to\ТвойПроект.project"
|
||
#
|
||
# По умолчанию папка plc_src создаётся автоматически рядом с файлом .project.
|
||
# Если авто-определение не сработает, впиши путь вручную в MANUAL_OUTPUT_ROOT ниже.
|
||
|
||
from __future__ import print_function
|
||
import os
|
||
import codecs
|
||
import re
|
||
import datetime
|
||
|
||
PRAGMA_RE = re.compile(r"^\s*\{[^}]*\}\s*")
|
||
|
||
# Единая метка времени на весь запуск экспорта (одинаковая во всех файлах).
|
||
EXPORT_TIMESTAMP = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# Имена служебных узлов дерева, которые не нужно показывать в поле Path
|
||
# (сравнение без учёта регистра). Сами узлы всё равно обходятся -- пропускаются
|
||
# только их имена при формировании пути.
|
||
EXCLUDED_PATH_NAMES = set(["application", "plc logic"])
|
||
|
||
# Закрывающие ключевые слова, которые CODESYS не включает ни в textual_declaration,
|
||
# ни в textual_implementation -- их нужно дописывать вручную, чтобы файл был
|
||
# синтаксически полным ST-блоком.
|
||
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",
|
||
}
|
||
|
||
|
||
def strip_leading_pragmas(text):
|
||
"""Убирает ведущие блоки вида {attribute 'qualified_only'} перед проверкой ключевого слова."""
|
||
while True:
|
||
new_text = PRAGMA_RE.sub("", text, 1)
|
||
if new_text == text:
|
||
return text
|
||
text = new_text
|
||
|
||
|
||
def get_project_file_path(proj):
|
||
"""Пытается получить полный путь к .project файлу разными способами,
|
||
так как имя атрибута может отличаться между версиями CODESYS."""
|
||
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 get_declaration_text(obj):
|
||
try:
|
||
if hasattr(obj, "textual_declaration") and obj.textual_declaration is not None:
|
||
return obj.textual_declaration.text
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
def has_implementation(obj):
|
||
try:
|
||
return bool(getattr(obj, "has_textual_implementation", False))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def get_implementation_text(obj):
|
||
try:
|
||
return obj.textual_implementation.text
|
||
except Exception as e:
|
||
return "(* ошибка чтения implementation: %s *)\r\n" % e
|
||
|
||
|
||
def classify(obj, declaration_text):
|
||
"""Определяет тип объекта.
|
||
Если есть декларация -- смотрим по ключевому слову (POU/GVL/DUT/METHOD/ACTION).
|
||
Если декларации нет, но есть implementation -- это ACTION (у него нет декларации в API)."""
|
||
if declaration_text:
|
||
stripped = strip_leading_pragmas(declaration_text.strip()).strip().upper()
|
||
if stripped.startswith(("PROGRAM", "FUNCTION_BLOCK", "FUNCTION", "INTERFACE")):
|
||
return "pou"
|
||
if stripped.startswith(("METHOD", "PROPERTY")):
|
||
return "method"
|
||
if stripped.startswith("VAR_GLOBAL"):
|
||
return "gvl"
|
||
if stripped.startswith("TYPE"):
|
||
return "dut"
|
||
return None
|
||
|
||
if has_implementation(obj):
|
||
return "action"
|
||
|
||
return None
|
||
|
||
|
||
def get_type_label(kind, declaration_text):
|
||
"""Возвращает читаемое название типа объекта, например 'FUNCTION_BLOCK', 'METHOD', 'ACTION'."""
|
||
if kind == "action":
|
||
return "ACTION"
|
||
if declaration_text:
|
||
stripped = strip_leading_pragmas(declaration_text.strip()).strip().upper()
|
||
for kw in ("FUNCTION_BLOCK", "PROGRAM", "FUNCTION", "INTERFACE", "METHOD", "PROPERTY"):
|
||
if stripped.startswith(kw):
|
||
return kw
|
||
if stripped.startswith("VAR_GLOBAL"):
|
||
return "GVL"
|
||
if stripped.startswith("TYPE"):
|
||
return "DUT"
|
||
return kind.upper() if kind else "UNKNOWN"
|
||
|
||
|
||
def close_block(text, type_label):
|
||
"""Дописывает END_PROGRAM / END_FUNCTION_BLOCK / END_METHOD / END_ACTION и т.п.,
|
||
если это применимо к данному типу объекта."""
|
||
kw = END_KEYWORDS.get(type_label)
|
||
if not kw:
|
||
return text
|
||
if not text.endswith("\n"):
|
||
text += "\r\n"
|
||
text += kw + "\r\n"
|
||
return text
|
||
|
||
|
||
def build_pou_text(obj, declaration_text, type_label):
|
||
"""Декларация + implementation (если есть) + закрывающее END_*."""
|
||
text = declaration_text or ""
|
||
if has_implementation(obj):
|
||
text += "\r\n(*----- IMPLEMENTATION -----*)\r\n"
|
||
text += get_implementation_text(obj)
|
||
text = close_block(text, type_label)
|
||
return text
|
||
|
||
|
||
def build_action_text(obj, name):
|
||
"""У Action нет declaration в API -- собираем заголовок 'ACTION Имя :' вручную,
|
||
затем тело и закрывающее END_ACTION."""
|
||
text = "ACTION %s :\r\n" % name
|
||
text += get_implementation_text(obj)
|
||
text = close_block(text, "ACTION")
|
||
return text
|
||
|
||
|
||
def build_header(project_name, device_name, path_parts, obj_name, type_label):
|
||
"""Заголовок-паспорт файла:
|
||
|
||
(*
|
||
===============================================================================
|
||
Project : MyProject
|
||
Device : PLC
|
||
Path : Application/Logic/Motion
|
||
Name : FB_Axis
|
||
Type : FUNCTION_BLOCK
|
||
Exported : 2026-07-30 14:32:18
|
||
===============================================================================
|
||
*)
|
||
"""
|
||
rel_path = "/".join(path_parts) if path_parts else "-"
|
||
sep = "=" * 79
|
||
lines = [
|
||
"(*",
|
||
sep,
|
||
" Project : %s" % project_name,
|
||
" Device : %s" % device_name,
|
||
" Path : %s" % rel_path,
|
||
" Name : %s" % obj_name,
|
||
" Type : %s" % type_label,
|
||
" Exported : %s" % EXPORT_TIMESTAMP,
|
||
sep,
|
||
"*)",
|
||
"",
|
||
"",
|
||
]
|
||
return "\r\n".join(lines)
|
||
|
||
|
||
def write_new_file(folder, path_parts, obj_name, text, project_name, device_name, type_label):
|
||
"""path_parts -- путь БЕЗ имени самого объекта (папки родителей).
|
||
Создаёт вложенные папки и записывает файл с заголовком-паспортом."""
|
||
sub_parts = [sanitize_folder_name(p) for p in path_parts]
|
||
sub_folder = os.path.join(folder, *sub_parts) if sub_parts else folder
|
||
if not os.path.exists(sub_folder):
|
||
os.makedirs(sub_folder)
|
||
file_name = sanitize_folder_name(obj_name)
|
||
path = os.path.join(sub_folder, file_name + ".st")
|
||
with codecs.open(path, "w", encoding="utf-8") as f:
|
||
f.write(build_header(project_name, device_name, path_parts, obj_name, type_label))
|
||
f.write(text)
|
||
print(" -> %s" % path)
|
||
return path
|
||
|
||
|
||
def append_to_file(path, path_parts, obj_name, text, project_name, device_name, type_label):
|
||
"""path_parts здесь включает имя родительского POU (т.к. метод/action лежит "внутри" него)."""
|
||
if not path or not os.path.exists(path):
|
||
return False
|
||
with codecs.open(path, "a", encoding="utf-8") as f:
|
||
f.write("\r\n")
|
||
f.write(build_header(project_name, device_name, path_parts, obj_name, type_label))
|
||
f.write(text)
|
||
print(" -> дописано в %s (%s: %s)" % (path, type_label, obj_name))
|
||
return True
|
||
|
||
|
||
def sanitize_folder_name(name):
|
||
"""Убирает символы, недопустимые в имени папки Windows."""
|
||
invalid = '\\/:*?"<>|'
|
||
for ch in invalid:
|
||
name = name.replace(ch, "_")
|
||
name = name.strip()
|
||
return name if name else "device"
|
||
|
||
|
||
def make_dirs(base):
|
||
d = {
|
||
"pou": os.path.join(base, "POU"),
|
||
"gvl": os.path.join(base, "GVL"),
|
||
"dut": os.path.join(base, "DUT"),
|
||
}
|
||
for p in d.values():
|
||
if not os.path.exists(p):
|
||
os.makedirs(p)
|
||
return d
|
||
|
||
|
||
def is_device_node(obj):
|
||
try:
|
||
return bool(getattr(obj, "is_device", False))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def walk(node, dirs, exported, project_name, device_name, current_path=None, parent_pou_path=None):
|
||
"""current_path -- список имён родителей от корня устройства (без учёта device_name самого),
|
||
накапливается по мере рекурсивного спуска по реальному дереву проекта -- поэтому Path
|
||
строится из фактической структуры, а не из строки, которую отдаёт CODESYS API."""
|
||
if current_path is None:
|
||
current_path = []
|
||
|
||
for child in node.get_children(False):
|
||
name = safe_name(child)
|
||
decl = get_declaration_text(child)
|
||
kind = classify(child, decl)
|
||
type_label = get_type_label(kind, decl)
|
||
|
||
next_parent_path = parent_pou_path
|
||
if name.strip().lower() in EXCLUDED_PATH_NAMES:
|
||
child_path = current_path # не добавляем служебное имя в путь
|
||
else:
|
||
child_path = current_path + [name] # путь для рекурсии в детей ЭТОГО child'а
|
||
|
||
if kind == "pou":
|
||
print("POU: %s/%s" % ("/".join(current_path), name))
|
||
text = build_pou_text(child, decl, type_label)
|
||
path = write_new_file(dirs["pou"], current_path, name, text,
|
||
project_name, device_name, type_label)
|
||
exported["pou"] += 1
|
||
next_parent_path = path
|
||
|
||
elif kind == "method":
|
||
print("METHOD: %s/%s" % ("/".join(current_path), name))
|
||
text = build_pou_text(child, decl, type_label)
|
||
if parent_pou_path and append_to_file(parent_pou_path, current_path, name, text,
|
||
project_name, device_name, type_label):
|
||
exported["method"] += 1
|
||
else:
|
||
write_new_file(dirs["pou"], current_path, name, text,
|
||
project_name, device_name, type_label)
|
||
exported["method"] += 1
|
||
|
||
elif kind == "action":
|
||
print("ACTION: %s/%s" % ("/".join(current_path), name))
|
||
text = build_action_text(child, name)
|
||
if parent_pou_path and append_to_file(parent_pou_path, current_path, name, text,
|
||
project_name, device_name, type_label):
|
||
exported["action"] += 1
|
||
else:
|
||
write_new_file(dirs["pou"], current_path, name, text,
|
||
project_name, device_name, type_label)
|
||
exported["action"] += 1
|
||
|
||
elif kind == "gvl":
|
||
print("GVL: %s/%s" % ("/".join(current_path), name))
|
||
write_new_file(dirs["gvl"], current_path, name, decl or "",
|
||
project_name, device_name, type_label)
|
||
exported["gvl"] += 1
|
||
|
||
elif kind == "dut":
|
||
print("DUT: %s/%s" % ("/".join(current_path), name))
|
||
write_new_file(dirs["dut"], current_path, name, decl or "",
|
||
project_name, device_name, type_label)
|
||
exported["dut"] += 1
|
||
|
||
else:
|
||
exported["skipped"] += 1
|
||
|
||
# рекурсивно обходим детей (методы/actions FB, вложенные папки и т.д.),
|
||
# передавая накопленный путь и (если только что создан POU) новый parent_pou_path
|
||
walk(child, dirs, exported, project_name, device_name, child_path, next_parent_path)
|
||
|
||
|
||
# ==== НАСТРОЙКА ====
|
||
# Оставить пустой строкой "" чтобы папка plc_src создавалась автоматически
|
||
# рядом с файлом .project. Указать путь вручную только если авто-определение
|
||
# не сработает (см. вывод скрипта при запуске).
|
||
MANUAL_OUTPUT_ROOT = r""
|
||
# =====================================================
|
||
|
||
proj = projects.primary
|
||
|
||
if proj is None:
|
||
print("ОШИБКА: нет открытого проекта. Открой .project перед запуском скрипта.")
|
||
else:
|
||
project_file = get_project_file_path(proj)
|
||
|
||
if project_file:
|
||
project_name = os.path.splitext(os.path.basename(project_file))[0]
|
||
else:
|
||
project_name = "UnknownProject"
|
||
|
||
if MANUAL_OUTPUT_ROOT:
|
||
OUTPUT_ROOT = MANUAL_OUTPUT_ROOT
|
||
print("Папка вывода задана вручную: %s" % OUTPUT_ROOT)
|
||
else:
|
||
if project_file:
|
||
project_dir = os.path.dirname(project_file)
|
||
OUTPUT_ROOT = os.path.join(project_dir, "plc_src")
|
||
print("Проект найден: %s" % project_file)
|
||
print("Папка вывода определена автоматически: %s" % OUTPUT_ROOT)
|
||
else:
|
||
OUTPUT_ROOT = None
|
||
print("ОШИБКА: не удалось определить путь к .project автоматически.")
|
||
print("Задай путь вручную в переменной MANUAL_OUTPUT_ROOT и запусти снова.")
|
||
|
||
if OUTPUT_ROOT:
|
||
exported = {"pou": 0, "gvl": 0, "dut": 0, "method": 0, "action": 0, "skipped": 0}
|
||
|
||
print("--- Экспорт проекта '%s' в %s ---" % (project_name, OUTPUT_ROOT))
|
||
|
||
top_children = proj.get_children(False)
|
||
device_nodes = [c for c in top_children if is_device_node(c)]
|
||
|
||
if device_nodes:
|
||
for device in device_nodes:
|
||
device_name = safe_name(device)
|
||
dev_folder = sanitize_folder_name(device_name)
|
||
print("=== Устройство: %s ===" % device_name)
|
||
dev_dirs = make_dirs(os.path.join(OUTPUT_ROOT, dev_folder))
|
||
walk(device, dev_dirs, exported, project_name, device_name)
|
||
|
||
other_dirs = make_dirs(os.path.join(OUTPUT_ROOT, "_common"))
|
||
for child in top_children:
|
||
if not is_device_node(child):
|
||
walk(child, other_dirs, exported, project_name, "-")
|
||
else:
|
||
print("Устройства верхнего уровня не найдены, экспорт без разбивки по папкам.")
|
||
dirs = make_dirs(OUTPUT_ROOT)
|
||
walk(proj, dirs, exported, project_name, "-")
|
||
|
||
print("--- Готово: POU=%d, METHOD=%d, ACTION=%d, GVL=%d, DUT=%d, пропущено=%d ---" % (
|
||
exported["pou"], exported["method"], exported["action"],
|
||
exported["gvl"], exported["dut"], exported["skipped"]
|
||
))
|