304 lines
9.6 KiB
Python
304 lines
9.6 KiB
Python
#!/usr/bin/env python
|
|
"""Read device I/O channels, parameters, and variable mappings from a CODESYS project."""
|
|
|
|
from __future__ import print_function
|
|
|
|
import os
|
|
|
|
from _common import emit, load_payload
|
|
|
|
|
|
def _discover_project_file(path_value):
|
|
if os.path.isfile(path_value):
|
|
lower_path = path_value.lower()
|
|
if lower_path.endswith(".project") or lower_path.endswith(".projectarchive"):
|
|
return path_value
|
|
raise RuntimeError(
|
|
"project_path must point to a .project or .projectarchive file. Got: {0}".format(path_value)
|
|
)
|
|
if os.path.isdir(path_value):
|
|
candidates = []
|
|
for entry in os.listdir(path_value):
|
|
full_path = os.path.join(path_value, entry)
|
|
if os.path.isfile(full_path):
|
|
lower_name = entry.lower()
|
|
if lower_name.endswith(".project") or lower_name.endswith(".projectarchive"):
|
|
candidates.append(full_path)
|
|
if len(candidates) == 1:
|
|
return candidates[0]
|
|
if not candidates:
|
|
raise RuntimeError("No .project/.projectarchive in directory: {0}".format(path_value))
|
|
raise RuntimeError("Multiple project files found: {0}".format(", ".join(candidates)))
|
|
raise RuntimeError("Project path does not exist: {0}".format(path_value))
|
|
|
|
|
|
def _safe_str(val):
|
|
try:
|
|
return "{0}".format(val)
|
|
except Exception:
|
|
return "?"
|
|
|
|
|
|
def _name_of(obj):
|
|
try:
|
|
return obj.get_name(False)
|
|
except Exception:
|
|
try:
|
|
return obj.get_name()
|
|
except Exception:
|
|
return _safe_str(obj)
|
|
|
|
|
|
def _children_of(obj):
|
|
try:
|
|
return list(obj.get_children(False))
|
|
except Exception:
|
|
try:
|
|
return list(obj.get_children())
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _get_device_id(obj):
|
|
try:
|
|
did = obj.get_device_identification()
|
|
return {
|
|
"type": int(did.type),
|
|
"id": _safe_str(did.id),
|
|
"version": _safe_str(did.version),
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _channel_type_str(ct):
|
|
mapping = {0: "none", 1: "input", 2: "output", 3: "output_readonly"}
|
|
try:
|
|
return mapping.get(int(ct), _safe_str(ct))
|
|
except Exception:
|
|
return _safe_str(ct)
|
|
|
|
|
|
def _read_io_mapping(param):
|
|
"""Extract IO mapping info from a device parameter."""
|
|
try:
|
|
iom = param.io_mapping
|
|
if iom is None:
|
|
return None
|
|
result = {}
|
|
try:
|
|
result["variable"] = _safe_str(iom.variable) if iom.variable else ""
|
|
except Exception:
|
|
result["variable"] = ""
|
|
try:
|
|
result["default_variable"] = _safe_str(iom.default_variable) if iom.default_variable else ""
|
|
except Exception:
|
|
result["default_variable"] = ""
|
|
try:
|
|
result["creates_variable"] = bool(iom.mapping_creates_variable)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
result["maps_to_existing"] = bool(iom.maps_to_existing_variable)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
result["auto_iec_address"] = bool(iom.automatic_iec_address)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
addr = iom.manual_iec_address
|
|
result["iec_address"] = _safe_str(addr) if addr else ""
|
|
except Exception:
|
|
result["iec_address"] = ""
|
|
return result
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _read_params(param_set, channels_only):
|
|
"""Read device parameters, optionally filtering to I/O channels only."""
|
|
params = []
|
|
try:
|
|
param_list = list(param_set)
|
|
except Exception:
|
|
return params
|
|
|
|
for param in param_list:
|
|
try:
|
|
ct = param.channel_type
|
|
ct_str = _channel_type_str(ct)
|
|
if channels_only and ct_str == "none":
|
|
continue
|
|
|
|
entry = {
|
|
"name": _safe_str(getattr(param, "visible_name", "") or getattr(param, "name", "")),
|
|
"id": int(param.id) if hasattr(param, "id") else -1,
|
|
"channel_type": ct_str,
|
|
"bit_size": int(param.bit_size) if hasattr(param, "bit_size") else 0,
|
|
}
|
|
try:
|
|
entry["iec_type"] = _safe_str(param.iec_type) if param.iec_type else ""
|
|
except Exception:
|
|
entry["iec_type"] = ""
|
|
try:
|
|
entry["type_string"] = _safe_str(param.type_string)
|
|
except Exception:
|
|
entry["type_string"] = ""
|
|
try:
|
|
entry["section"] = _safe_str(param.section) if param.section else ""
|
|
except Exception:
|
|
entry["section"] = ""
|
|
|
|
mapping = _read_io_mapping(param)
|
|
if mapping is not None:
|
|
entry["io_mapping"] = mapping
|
|
|
|
if getattr(param, "has_sub_elements", False):
|
|
sub_entries = []
|
|
try:
|
|
for sub in list(param):
|
|
sub_entry = {
|
|
"name": _safe_str(getattr(sub, "visible_name", "") or getattr(sub, "identifier", "")),
|
|
"bit_size": int(sub.bit_size) if hasattr(sub, "bit_size") else 0,
|
|
}
|
|
try:
|
|
sub_entry["iec_type"] = _safe_str(sub.iec_type) if hasattr(sub, "iec_type") and sub.iec_type else ""
|
|
except Exception:
|
|
pass
|
|
sub_mapping = _read_io_mapping(sub)
|
|
if sub_mapping is not None:
|
|
sub_entry["io_mapping"] = sub_mapping
|
|
sub_entries.append(sub_entry)
|
|
except Exception:
|
|
pass
|
|
if sub_entries:
|
|
entry["sub_elements"] = sub_entries
|
|
|
|
params.append(entry)
|
|
except Exception:
|
|
continue
|
|
return params
|
|
|
|
|
|
def _walk_devices(node, path_parts, results, warnings, channels_only, max_depth, depth):
|
|
if depth > max_depth:
|
|
return
|
|
|
|
name = _name_of(node)
|
|
is_dev = getattr(node, "is_device", False)
|
|
|
|
if is_dev:
|
|
device_info = {
|
|
"name": name,
|
|
"path": "/".join(path_parts + [name]),
|
|
}
|
|
|
|
dev_id = _get_device_id(node)
|
|
if dev_id:
|
|
device_info["device_id"] = dev_id
|
|
|
|
try:
|
|
params = node.device_parameters
|
|
device_info["parameters"] = _read_params(params, channels_only)
|
|
device_info["parameter_count"] = len(device_info["parameters"])
|
|
except Exception as exc:
|
|
device_info["parameters"] = []
|
|
device_info["parameter_count"] = 0
|
|
warnings.append("params_failed({0}): {1}".format(name, exc))
|
|
|
|
connectors_info = []
|
|
try:
|
|
conn_set = node.connectors
|
|
for conn in list(conn_set):
|
|
ci = {
|
|
"connector_id": int(conn.connector_id),
|
|
"interface": _safe_str(conn.interface),
|
|
"role": "parent" if int(conn.connector_role) == 0 else "child",
|
|
}
|
|
try:
|
|
hp = conn.host_parameters
|
|
ci["host_parameters"] = _read_params(hp, channels_only)
|
|
except Exception:
|
|
pass
|
|
connectors_info.append(ci)
|
|
except Exception as exc:
|
|
warnings.append("connectors_failed({0}): {1}".format(name, exc))
|
|
|
|
if connectors_info:
|
|
device_info["connectors"] = connectors_info
|
|
|
|
results.append(device_info)
|
|
|
|
for child in _children_of(node):
|
|
try:
|
|
_walk_devices(child, path_parts + [name], results, warnings, channels_only, max_depth, depth + 1)
|
|
except Exception as exc:
|
|
warnings.append("walk_child_failed({0}): {1}".format(name, exc))
|
|
|
|
|
|
def main():
|
|
payload = load_payload()
|
|
project_path = payload.get("project_path")
|
|
if not project_path:
|
|
raise ValueError("project_path is required.")
|
|
|
|
channels_only = bool(payload.get("channels_only", True))
|
|
max_depth = int(payload.get("max_depth", 10))
|
|
export_csv_path = payload.get("export_csv_path", "")
|
|
|
|
pp = os.path.abspath(os.path.expanduser(project_path))
|
|
pp = _discover_project_file(pp)
|
|
|
|
if "projects" not in globals():
|
|
raise RuntimeError("Run through CODESYS ScriptEngine (--runscript).")
|
|
|
|
if projects.primary:
|
|
projects.primary.close()
|
|
|
|
if pp.lower().endswith(".projectarchive"):
|
|
project_dir = os.path.dirname(pp)
|
|
proj = projects.open_archive(pp, project_dir, overwrite=False)
|
|
else:
|
|
proj = projects.open(pp, primary=True, allow_readonly=True)
|
|
|
|
devices = []
|
|
warnings = []
|
|
|
|
try:
|
|
top_nodes = _children_of(proj)
|
|
except Exception as exc:
|
|
top_nodes = []
|
|
warnings.append("top_level_enumeration_failed: {0}".format(exc))
|
|
|
|
for top in top_nodes:
|
|
try:
|
|
_walk_devices(top, [], devices, warnings, channels_only, max_depth, 0)
|
|
except Exception as exc:
|
|
warnings.append("walk_failed: {0}".format(exc))
|
|
|
|
if export_csv_path:
|
|
try:
|
|
for top in top_nodes:
|
|
if getattr(top, "is_device", False):
|
|
top.export_io_mappings_as_csv(export_csv_path)
|
|
break
|
|
except Exception as exc:
|
|
warnings.append("csv_export_failed: {0}".format(exc))
|
|
|
|
proj.close()
|
|
|
|
emit({
|
|
"ok": True,
|
|
"action": "read_device_io",
|
|
"project_path": pp,
|
|
"devices": devices,
|
|
"device_count": len(devices),
|
|
"channels_only": channels_only,
|
|
"warnings": warnings,
|
|
})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|