319 lines
9.3 KiB
Python
319 lines
9.3 KiB
Python
#!/usr/bin/env python
|
|
"""Add, remove, or update devices in a CODESYS project via ScriptEngine."""
|
|
|
|
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 be .project or .projectarchive: {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: {0}".format(path_value))
|
|
raise RuntimeError("Multiple project files: {0}".format(", ".join(candidates)))
|
|
raise RuntimeError("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 _find_by_path(root, path_str):
|
|
parts = [p.strip() for p in path_str.split("/") if p.strip()]
|
|
current = root
|
|
for i, part in enumerate(parts):
|
|
found = None
|
|
for child in _children_of(current):
|
|
if _name_of(child) == part:
|
|
found = child
|
|
break
|
|
if found is None:
|
|
raise RuntimeError("Not found: '{0}' at segment {1} of '{2}'".format(part, i, path_str))
|
|
current = found
|
|
return current
|
|
|
|
|
|
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 _action_add(proj, op, warnings):
|
|
parent_path = op.get("parent_path", "")
|
|
name = op.get("name", "")
|
|
dev_type = op.get("device_type")
|
|
dev_id = op.get("device_id", "")
|
|
dev_version = op.get("device_version", "")
|
|
module_id = op.get("module_id")
|
|
|
|
if not parent_path or not name or dev_type is None or not dev_id or not dev_version:
|
|
raise ValueError("add requires: parent_path, name, device_type, device_id, device_version")
|
|
|
|
parent = _find_by_path(proj, parent_path)
|
|
args = [name, int(dev_type), dev_id, dev_version]
|
|
if module_id is not None:
|
|
args.append(module_id)
|
|
|
|
method = op.get("method", "add")
|
|
if method == "plug":
|
|
parent.plug(*args)
|
|
elif method == "insert":
|
|
index = int(op.get("index", -1))
|
|
parent.insert(name, index, int(dev_type), dev_id, dev_version)
|
|
else:
|
|
parent.add(*args)
|
|
|
|
return {
|
|
"action": "add",
|
|
"parent_path": parent_path,
|
|
"name": name,
|
|
"device_type": dev_type,
|
|
"device_id": dev_id,
|
|
"device_version": dev_version,
|
|
"status": "added",
|
|
}
|
|
|
|
|
|
def _action_remove(proj, op, warnings):
|
|
device_path = op.get("device_path", "")
|
|
if not device_path:
|
|
raise ValueError("remove requires: device_path")
|
|
obj = _find_by_path(proj, device_path)
|
|
name = _name_of(obj)
|
|
dev_id = _get_device_id(obj)
|
|
obj.remove()
|
|
return {
|
|
"action": "remove",
|
|
"device_path": device_path,
|
|
"name": name,
|
|
"device_id": dev_id,
|
|
"status": "removed",
|
|
}
|
|
|
|
|
|
def _action_update(proj, op, warnings):
|
|
device_path = op.get("device_path", "")
|
|
dev_type = op.get("device_type")
|
|
dev_id = op.get("device_id", "")
|
|
dev_version = op.get("device_version", "")
|
|
module_id = op.get("module_id")
|
|
|
|
if not device_path or dev_type is None or not dev_id or not dev_version:
|
|
raise ValueError("update requires: device_path, device_type, device_id, device_version")
|
|
|
|
obj = _find_by_path(proj, device_path)
|
|
old_id = _get_device_id(obj)
|
|
|
|
args = [int(dev_type), dev_id, dev_version]
|
|
if module_id is not None:
|
|
args.append(module_id)
|
|
obj.update(*args)
|
|
|
|
return {
|
|
"action": "update",
|
|
"device_path": device_path,
|
|
"old_device_id": old_id,
|
|
"new_device_id": {"type": dev_type, "id": dev_id, "version": dev_version},
|
|
"status": "updated",
|
|
}
|
|
|
|
|
|
def _action_rename(proj, op, warnings):
|
|
device_path = op.get("device_path", "")
|
|
new_name = op.get("new_name", "")
|
|
if not device_path or not new_name:
|
|
raise ValueError("rename requires: device_path, new_name")
|
|
obj = _find_by_path(proj, device_path)
|
|
old_name = _name_of(obj)
|
|
obj.rename(new_name)
|
|
return {
|
|
"action": "rename",
|
|
"device_path": device_path,
|
|
"old_name": old_name,
|
|
"new_name": new_name,
|
|
"status": "renamed",
|
|
}
|
|
|
|
|
|
def _action_enable(proj, op, warnings):
|
|
device_path = op.get("device_path", "")
|
|
enabled = op.get("enabled", True)
|
|
if not device_path:
|
|
raise ValueError("enable/disable requires: device_path")
|
|
obj = _find_by_path(proj, device_path)
|
|
if enabled:
|
|
obj.enable()
|
|
else:
|
|
obj.disable()
|
|
return {
|
|
"action": "enable" if enabled else "disable",
|
|
"device_path": device_path,
|
|
"status": "enabled" if enabled else "disabled",
|
|
}
|
|
|
|
|
|
def _action_set_param(proj, op, warnings):
|
|
"""Set a device parameter value by id or name."""
|
|
device_path = op.get("device_path", "")
|
|
param_id = op.get("param_id")
|
|
param_name = op.get("param_name", "")
|
|
value = op.get("value")
|
|
|
|
if not device_path:
|
|
raise ValueError("set_param requires: device_path")
|
|
if value is None:
|
|
raise ValueError("set_param requires: value")
|
|
|
|
obj = _find_by_path(proj, device_path)
|
|
params = obj.device_parameters
|
|
target = None
|
|
|
|
for p in list(params):
|
|
if param_id is not None and int(p.id) == int(param_id):
|
|
target = p
|
|
break
|
|
if param_name and (_safe_str(getattr(p, "visible_name", "")) == param_name or _safe_str(getattr(p, "name", "")) == param_name):
|
|
target = p
|
|
break
|
|
|
|
if target is None:
|
|
raise RuntimeError("Parameter not found: id={0} name={1} on {2}".format(param_id, param_name, device_path))
|
|
|
|
old_value = _safe_str(target.value) if hasattr(target, "value") else "?"
|
|
target.value = _safe_str(value)
|
|
|
|
return {
|
|
"action": "set_param",
|
|
"device_path": device_path,
|
|
"param_id": int(target.id),
|
|
"param_name": _safe_str(getattr(target, "visible_name", "")),
|
|
"old_value": old_value,
|
|
"new_value": _safe_str(value),
|
|
"status": "set",
|
|
}
|
|
|
|
|
|
ACTIONS = {
|
|
"add": _action_add,
|
|
"remove": _action_remove,
|
|
"update": _action_update,
|
|
"rename": _action_rename,
|
|
"enable": _action_enable,
|
|
"disable": _action_enable,
|
|
"set_param": _action_set_param,
|
|
}
|
|
|
|
|
|
def main():
|
|
payload = load_payload()
|
|
project_path = payload.get("project_path")
|
|
if not project_path:
|
|
raise ValueError("project_path is required.")
|
|
|
|
operations = payload.get("operations", [])
|
|
if not operations:
|
|
single_action = payload.get("action_type", "")
|
|
if single_action:
|
|
operations = [payload]
|
|
else:
|
|
raise ValueError("Provide 'operations' list or a single operation with 'action_type'.")
|
|
|
|
dry_run = bool(payload.get("dry_run", False))
|
|
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()
|
|
|
|
proj = projects.open(pp, primary=True)
|
|
results = []
|
|
warnings = []
|
|
|
|
for op in operations:
|
|
action_type = op.get("action_type", "")
|
|
handler = ACTIONS.get(action_type)
|
|
if not handler:
|
|
warnings.append("unknown action_type: {0}".format(action_type))
|
|
continue
|
|
|
|
try:
|
|
if dry_run:
|
|
results.append({
|
|
"action": action_type,
|
|
"dry_run": True,
|
|
"status": "would_execute",
|
|
"details": op,
|
|
})
|
|
else:
|
|
result = handler(proj, op, warnings)
|
|
results.append(result)
|
|
except Exception as exc:
|
|
warnings.append("{0}_failed: {1}".format(action_type, exc))
|
|
|
|
if not dry_run:
|
|
try:
|
|
proj.save()
|
|
except Exception as exc:
|
|
warnings.append("project_save_failed: {0}".format(exc))
|
|
|
|
proj.close()
|
|
|
|
emit({
|
|
"ok": True,
|
|
"action": "manage_device",
|
|
"project_path": pp,
|
|
"dry_run": dry_run,
|
|
"results": results,
|
|
"result_count": len(results),
|
|
"warnings": warnings,
|
|
})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|