Modbus and Bacnet websettings : Rename variables and functions to avoid name collisions.
Websettings for modbus and bacnet are now passed to runtime as python files loaded (execfile) at init of PLCObject with the same globals.
Because if this, same names used in previously different modules now colide.
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This file is part of Beremiz
#
# Copyright (C) 2011-2014: Laurent BESSARD, Edouard TISSERANT
# RTES Lab : CRKim, JBLee, youcu
# Higen Motor : Donggu Kang
#
# See COPYING file for copyrights details.
from __future__ import absolute_import
import os
import shutil
import csv
from builtins import str as text
from lxml import etree
import wx
from xmlclass import *
from ConfigTreeNode import XSDSchemaErrorMessage
from etherlab.EthercatSlave import ExtractHexDecValue, ExtractName
from etherlab.EthercatMaster import _EthercatCTN
from etherlab.ConfigEditor import LibraryEditor, ETHERCAT_VENDOR, ETHERCAT_GROUP, ETHERCAT_DEVICE
ScriptDirectory = os.path.split(os.path.realpath(__file__))[0]
# --------------------------------------------------
# Ethercat ConfNode
# --------------------------------------------------
EtherCATInfoParser = GenerateParserFromXSD(os.path.join(os.path.dirname(__file__), "EtherCATInfo.xsd"))
def EtherCATInfo_XPath(xpath):
return etree.XPath(xpath)
EtherCATBaseParser = GenerateParserFromXSD(os.path.join(os.path.dirname(__file__), "EtherCATBase.xsd"))
def EtherCATBase_XPath(xpath):
return etree.XPath(xpath)
def HexDecValue(context, *args):
return str(ExtractHexDecValue(args[0][0]))
def EntryName(context, *args):
return ExtractName(args[0], args[1][0] if len(args) > 1 else None)
ENTRY_INFOS_KEYS = [
("Index", lambda x: "#x%4.4X" % int(x), "#x0000"),
("SubIndex", str, "0"),
("Name", str, ""),
("Type", str, ""),
("BitSize", int, 0),
("Access", str, ""),
("PDOMapping", str, ""),
("PDO index", str, ""),
("PDO name", str, ""),
("PDO type", str, "")]
# Read DefaultValue from ESI file
# Add by jblee 151229
ENTRY_INFOS_KEYS_FOR_DV = [
("Index", lambda x: "#x%4.4X" % int(x), "#x0000"),
("SubIndex", str, "0"),
("Name", str, ""),
("Type", str, ""),
("BitSize", int, 0),
("Access", str, ""),
("PDOMapping", str, ""),
("DefaultValue", str, ""),
("Sub_entry_flag", str, "0")]
class EntryListFactory(object):
def __init__(self, entries):
self.Entries = entries
def AddEntry(self, context, *args):
index, subindex = map(lambda x: int(x[0]), args[:2])
if len(args) > 9:
new_entry_infos = {
key: translate(arg[0]) if len(arg) > 0 else default
for (key, translate, default), arg
in zip(ENTRY_INFOS_KEYS, args)}
else:
new_entry_infos = {
key: translate(arg[0]) if len(arg) > 0 else default
for (key, translate, default), arg
in zip(ENTRY_INFOS_KEYS_FOR_DV, args)}
if (index, subindex) != (0, 0):
entry_infos = self.Entries.get((index, subindex))
if entry_infos is not None:
for param in ["PDO index", "PDO name", "PDO type"]:
value = new_entry_infos.get(param)
if value is not None:
entry_infos[param] = value
else:
self.Entries[(index, subindex)] = new_entry_infos
entries_list_xslt = etree.parse(
os.path.join(ScriptDirectory, "entries_list.xslt"))
cls = EtherCATInfoParser.GetElementClass("DeviceType")
if cls:
cls.DataTypes = None
profile_numbers_xpath = EtherCATInfo_XPath("Profile/ProfileNo")
def GetProfileNumbers(self):
return [number.text for number in profile_numbers_xpath(self)]
setattr(cls, "GetProfileNumbers", GetProfileNumbers)
def getCoE(self):
mailbox = self.getMailbox()
if mailbox is not None:
return mailbox.getCoE()
return None
setattr(cls, "getCoE", getCoE)
# Modify by jblee
def ExtractDataTypes(self):
#self.DataTypes = {}
#self.DT = {}
DT = {}
objects = []
# get Profile Field
for profile in self.getProfile():
# get each (ProfileNo, Dictionary) Field as child
for child in profile.getchildren():
# child.text is not None -> ProfileNo, is None -> Dictionary
if child.text is None:
# get each (DataTypes, Objects) Field
dataTypes = child.getDataTypes()
objects = child.getObjects()
for dataType in dataTypes.getDataType():
#if dataType.getName() is not None:
# print dataType.getName(), dataType
DT[dataType.getName()] = dataType
return DT, objects
setattr(cls, "ExtractDataTypes", ExtractDataTypes)
def GetEntriesList(self, limits=None):
DataTypes, objects = self.ExtractDataTypes()
entries = {}
# get each Object Field
for object in objects:
# Object Field mendatory : Index, Name, Type, BitSize
# Frequently Use : Info, Flags
# Info Field -> DefaultData, SubItem
# Flags Field -> Access, Category, PdoMapping
object_index = object.getIndex().getcontent()
index = ExtractHexDecValue(object_index)
if limits is None or limits[0] <= index <= limits[1]:
object_type = object.getType()
object_name = ExtractName(object.getName())
object_size = object.getBitSize()
defaultData = ""
object_access = ""
object_PDOMapping_data = ""
object_type_infos = DataTypes.get(object_type, None)
subItem_infos = object_type_infos.getchildren()
countSubIndex = 0
if len(subItem_infos) > 2:
for subItem_info in subItem_infos:
if subItem_info.tag == "SubItem" :
subItemName = subItem_info.getName()
subIdx = subItem_info.getSubIdx()
if subIdx is not None:
object_subidx = ExtractHexDecValue(subIdx)
else:
object_subidx = ExtractHexDecValue(countSubIndex)
subType = subItem_info.getType()
subBitSize = subItem_info.getBitSize()
subFlags = subItem_info.getFlags()
subAccess = ""
subPDOMapping_data = ""
if subFlags is not None:
subAccess = subFlags.getAccess().getcontent()
subPDOMapping = subFlags.getPdoMapping()
if subPDOMapping is not None:
subPDOMapping_data = subFlags.getPdoMapping().upper()
entries[(index, object_subidx)] = {
"Index": object_index,
"SubIndex": subIdx,
"Name": "%s - %s" %
(object_name.decode("utf-8"),
subItemName.decode("utf-8")),
"Type": subType,
"BitSize": subBitSize,
"Access": subAccess,
"PDOMapping": subPDOMapping_data}
countSubIndex += 1
info = object.getInfo()
# subItemTest : check subItem
countSubIndex = 0
if info is not None:
subItems = info.getchildren()
if len(subItems) > 1:
for subItem in subItems:
defaultdata_subidx = ExtractHexDecValue(countSubIndex)
defaultData = subItem.getchildren()[1].findtext("DefaultData")
entry = entries.get((index, defaultdata_subidx), None)
if entry is not None:
entry["DefaultData"] = defaultData
countSubIndex += 1
else :
info = object.getInfo()
if info is not None:
subItems = info.getchildren()
if len(subItems) <= 1:
defaultData = subItems[0].text
object_flag = object.getFlags()
object_access = object_flag.getAccess().getcontent()
object_PDOMapping = object_flag.getPdoMapping()
if object_PDOMapping is not None:
object_PDOMapping_data = object_flag.getPdoMapping().upper()
entries[(index, 0)] = {
"Index": object_index,
"SubIndex": "0",
"Name": object_name,
"Type": object_type,
"BitSize": object_size,
"DefaultData" : defaultData,
"Access": object_access,
"PDOMapping": object_PDOMapping_data}
for TxPdo in self.getTxPdo():
ExtractPdoInfos(TxPdo, "Transmit", entries, limits)
for RxPdo in self.getRxPdo():
ExtractPdoInfos(RxPdo, "Receive", entries, limits)
return entries
setattr(cls, "GetEntriesList", GetEntriesList)
# def GetEntriesList(self, limits=None):
# entries = {}
# factory = EntryListFactory(entries)
# entries_list_xslt_tree = etree.XSLT(
# entries_list_xslt, extensions = {
# ("entries_list_ns", "AddEntry"): factory.AddEntry,
# ("entries_list_ns", "HexDecValue"): HexDecValue,
# ("entries_list_ns", "EntryName"): EntryName})
# entries_list_xslt_tree(self, **dict(zip(
# ["min_index", "max_index"],
# map(lambda x: etree.XSLT.strparam(str(x)),
# limits if limits is not None else [0x0000, 0xFFFF])
# )))
#
# return entries
# setattr(cls, "GetEntriesList", GetEntriesList)
def GetSyncManagers(self):
sync_managers = []
for sync_manager in self.getSm():
sync_manager_infos = {}
for name, value in [("Name", sync_manager.getcontent()),
("Start Address", sync_manager.getStartAddress()),
("Default Size", sync_manager.getDefaultSize()),
("Control Byte", sync_manager.getControlByte()),
("Enable", sync_manager.getEnable())]:
if value is None:
value = ""
sync_manager_infos[name] = value
sync_managers.append(sync_manager_infos)
return sync_managers
setattr(cls, "GetSyncManagers", GetSyncManagers)
def GroupItemCompare(x, y):
if x["type"] == y["type"]:
if x["type"] == ETHERCAT_GROUP:
return cmp(x["order"], y["order"])
else:
return cmp(x["name"], y["name"])
elif x["type"] == ETHERCAT_GROUP:
return -1
return 1
def SortGroupItems(group):
for item in group["children"]:
if item["type"] == ETHERCAT_GROUP:
SortGroupItems(item)
group["children"].sort(GroupItemCompare)
def ExtractPdoInfos(pdo, pdo_type, entries, limits=None):
pdo_index = pdo.getIndex().getcontent()
pdo_name = ExtractName(pdo.getName())
exclude = pdo.getExclude()
for pdo_entry in pdo.getEntry():
entry_index = pdo_entry.getIndex().getcontent()
entry_subindex = pdo_entry.getSubIndex()
index = ExtractHexDecValue(entry_index)
subindex = ExtractHexDecValue(entry_subindex)
object_size = pdo_entry.getBitLen()
if limits is None or limits[0] <= index <= limits[1]:
entry = entries.get((index, subindex), None)
if entry is not None:
entry["PDO index"] = pdo_index
entry["PDO name"] = pdo_name
entry["PDO type"] = pdo_type
else:
entry_type = pdo_entry.getDataType()
if entry_type is not None:
if pdo_type == "Transmit":
access = "ro"
pdomapping = "T"
else:
access = "wo"
pdomapping = "R"
entries[(index, subindex)] = {
"Index": entry_index,
"SubIndex": entry_subindex,
"Name": ExtractName(pdo_entry.getName()),
"Type": entry_type.getcontent(),
"BitSize": object_size,
"Access": access,
"PDOMapping": pdomapping}
class ModulesLibrary(object):
MODULES_EXTRA_PARAMS = [
(
"pdo_alignment",
{
"column_label": _("PDO alignment"),
"column_size": 150,
"default": 8,
"description": _("Minimal size in bits between 2 pdo entries")
}
),
(
"max_pdo_size",
{
"column_label": _("Max entries by PDO"),
"column_size": 150,
"default": 255,
"description": _("""Maximal number of entries mapped in a PDO
including empty entries used for PDO alignment""")
}
),
(
"add_pdo",
{
"column_label": _("Creating new PDO"),
"column_size": 150,
"default": 0,
"description": _("""Adding a PDO not defined in default configuration
for mapping needed location variables
(1 if possible)""")
}
)
]
def __init__(self, path, parent_library=None):
self.Path = path
if not os.path.exists(self.Path):
os.makedirs(self.Path)
self.ParentLibrary = parent_library
if parent_library is not None:
self.LoadModules()
else:
self.Library = None
self.LoadModulesExtraParams()
def GetPath(self):
return self.Path
def GetModulesExtraParamsFilePath(self):
return os.path.join(self.Path, "modules_extra_params.cfg")
groups_xpath = EtherCATInfo_XPath("Descriptions/Groups/Group")
devices_xpath = EtherCATInfo_XPath("Descriptions/Devices/Device")
module_xpath = EtherCATBase_XPath("Descriptions/Modules/Module")
def LoadModules(self):
self.Library = {}
# add by jblee for Modular Device Profile
self.MDPList = []
self.ModuleList = []
self.MDPEntryList = {}
dtDic = {}
self.idxIncrement = 0
self.slotIncrement = 0
# add by jblee for PDO Mapping
self.DataTypes = {}
self.ObjectDictionary = {}
files = os.listdir(self.Path)
for file in files:
filepath = os.path.join(self.Path, file)
if os.path.isfile(filepath) and os.path.splitext(filepath)[-1] == ".xml":
self.modules_infos = None
xmlfile = open(filepath, 'r')
try:
self.modules_infos, error = EtherCATInfoParser.LoadXMLString(xmlfile.read())
# if error is not None:
# self.GetCTRoot().logger.write_warning(
# XSDSchemaErrorMessage % (filepath + error))
except Exception as exc:
self.modules_infos, error = None, text(exc)
xmlfile.close()
if self.modules_infos is not None:
vendor = self.modules_infos.getVendor()
vendor_category = self.Library.setdefault(
ExtractHexDecValue(vendor.getId()),
{"name": ExtractName(vendor.getName(), _("Miscellaneous")),
"groups": {}})
for group in self.groups_xpath(self.modules_infos):
group_type = group.getType()
# add for XmlToEeprom Func by jblee.
self.LcId_data = group.getchildren()[1]
self.Image16x14_data = group.getchildren()[2]
vendor_category["groups"].setdefault(
group_type,
{
"name": ExtractName(group.getName(), group_type),
"parent": group.getParentGroup(),
"order": group.getSortOrder(),
"devices": [],
# add jblee for support Moduler Device Profile (MDP)
"modules": []})
})
for device in self.devices_xpath(self.modules_infos):
device_group = device.getGroupType()
if device_group not in vendor_category["groups"]:
raise ValueError("Not such group \"%s\"" % device_group)
vendor_category["groups"][device_group]["devices"].append(
(device.getType().getcontent(), device))
# ------------------ Test Section --------------------#
slots = device.getSlots()
if slots is not None:
for slot in slots.getSlot():
self.idxIncrement = slot.getSlotIndexIncrement()
self.slotIncrement = slot.getSlotPdoIncrement()
for child in slot.getchildren():
if child.tag == "ModuleClass":
child_class = child.getClass()
child_name = child.getName()
# -------------------- Test Section ----------------------------------#
LocalMDPList = []
for module in self.module_xpath(self.modules_infos):
module_type = module.getType().getModuleClass()
module_name = module.getName()
LocalMDPData = ExtractName(module_name) + " (" + module_type + ")"
self.ModuleList.append(module)
try :
module_pdos = module.getTxPdo()
module_pdos += module.getRxPdo()
for module_pdo in module_pdos:
device_name = ExtractName(module_name)
pdo_index = module_pdo.getIndex().getcontent()
pdo_name = ExtractName(module_pdo.getName())
pdo_entry = module_pdo.getEntry()
if module_pdo.tag == "TxPdo":
mapping_type = "T"
else :
mapping_type = "R"
LocalMDPEntry = []
for entry in pdo_entry:
entry_index = entry.getIndex().getcontent()
entry_subidx = entry.getSubIndex()
entry_name = ExtractName(entry.getName())
entry_bitsize = entry.getBitLen()
try :
entry_type = entry.getDataType().getcontent()
except :
entry_type = ""
LocalMDPEntry.append({
"Index": entry_index,
"SubIndex": entry_subidx,
"Name": "%s - %s" % (pdo_name, entry_name),
"Type": entry_type,
"BitSize": entry_bitsize,
"Access": "",
"PDOMapping": mapping_type})
self.MDPEntryList[device_name] = LocalMDPEntry
LocalMDPList.append([LocalMDPData, module, LocalMDPEntry])
except :
LocalMDPList.append([LocalMDPData, module, []])
if LocalMDPList:
vendor_category["groups"][device_group]["modules"].append(
(device.getType().getcontent(), LocalMDPList, self.idxIncrement, self.slotIncrement))
#self.MDPList.append([device.getType().getcontent(), LocalMDPList,
# self.idxIncrement, self.slotIncrement])
# --------------------------------------------------------------------- #
# else:
# self.GetCTRoot().logger.write_error(
# _("Couldn't load {a1} XML file:\n{a2}").format(a1=filepath, a2=error))
return self.Library
# add jblee
def GetMDPList(self):
return self.MDPList
# add jblee
def GetSelectModule(self, idx):
return self.ModuleList[idx]
# add jblee
def GetModuleEntryList(self):
return self.MDPEntryList
# add jblee
def GetModuleIncrement(self):
return (self.idxIncrement, self.slotIncrement)
# add jblee
#def GetEntriesList(self):
# return self.ObjectDictionary
def GetModulesLibrary(self, profile_filter=None):
if self.Library is None:
self.LoadModules()
library = []
for vendor_id, vendor in self.Library.iteritems():
groups = []
children_dict = {}
for group_type, group in vendor["groups"].iteritems():
group_infos = {"name": group["name"],
"order": group["order"],
"type": ETHERCAT_GROUP,
"infos": None,
"children": children_dict.setdefault(group_type, [])}
device_dict = {}
for device_type, device in group["devices"]:
if profile_filter is None or profile_filter in device.GetProfileNumbers():
product_code = device.getType().getProductCode()
revision_number = device.getType().getRevisionNo()
module_infos = {"device_type": device_type,
"vendor": vendor_id,
"product_code": product_code,
"revision_number": revision_number}
module_infos.update(self.GetModuleExtraParams(vendor_id, product_code, revision_number))
device_infos = {"name": ExtractName(device.getName()),
"type": ETHERCAT_DEVICE,
"infos": module_infos,
"children": []}
group_infos["children"].append(device_infos)
device_type_occurrences = device_dict.setdefault(device_type, [])
device_type_occurrences.append(device_infos)
for device_type_occurrences in device_dict.itervalues():
if len(device_type_occurrences) > 1:
for occurrence in device_type_occurrences:
occurrence["name"] += _(" (rev. %s)") % occurrence["infos"]["revision_number"]
if len(group_infos["children"]) > 0:
if group["parent"] is not None:
parent_children = children_dict.setdefault(group["parent"], [])
parent_children.append(group_infos)
else:
groups.append(group_infos)
if len(groups) > 0:
library.append({"name": vendor["name"],
"type": ETHERCAT_VENDOR,
"infos": None,
"children": groups})
library.sort(lambda x, y: cmp(x["name"], y["name"]))
return library
def GetVendors(self):
return [(vendor_id, vendor["name"]) for vendor_id, vendor in self.Library.items()]
def GetModuleInfos(self, module_infos):
vendor = ExtractHexDecValue(module_infos["vendor"])
vendor_infos = self.Library.get(vendor)
if vendor_infos is not None:
for _group_name, group_infos in vendor_infos["groups"].iteritems():
for device_type, device_infos in group_infos["devices"]:
product_code = ExtractHexDecValue(device_infos.getType().getProductCode())
revision_number = ExtractHexDecValue(device_infos.getType().getRevisionNo())
if product_code == ExtractHexDecValue(module_infos["product_code"]) and \
revision_number == ExtractHexDecValue(module_infos["revision_number"]):
self.cntdevice = device_infos
self.cntdeviceType = device_type
return device_infos, self.GetModuleExtraParams(vendor, product_code, revision_number)
return None, None
# add jblee for MDP
def GetMDPInfos(self, module_infos):
vendor = ExtractHexDecValue(module_infos["vendor"])
vendor_infos = self.Library.get(vendor)
if vendor_infos is not None:
for group_name, group_infos in vendor_infos["groups"].iteritems():
return group_infos["modules"]
#for device_type, module_list, idx_inc, slot_inc in group_infos["modules"]:
# return module_list, idx_inc, slot_inc
#return None, None, None
def ImportModuleLibrary(self, filepath):
if os.path.isfile(filepath):
shutil.copy(filepath, self.Path)
self.LoadModules()
return True
return False
def LoadModulesExtraParams(self):
self.ModulesExtraParams = {}
csvfile_path = self.GetModulesExtraParamsFilePath()
if os.path.exists(csvfile_path):
csvfile = open(csvfile_path, "rb")
sample = csvfile.read(1024)
csvfile.seek(0)
dialect = csv.Sniffer().sniff(sample)
has_header = csv.Sniffer().has_header(sample)
reader = csv.reader(csvfile, dialect)
for row in reader:
if has_header:
has_header = False
else:
params_values = {}
for (param, _param_infos), value in zip(
self.MODULES_EXTRA_PARAMS, row[3:]):
if value != "":
params_values[param] = int(value)
self.ModulesExtraParams[
tuple(map(int, row[:3]))] = params_values
csvfile.close()
def SaveModulesExtraParams(self):
csvfile = open(self.GetModulesExtraParamsFilePath(), "wb")
extra_params = [param for param, _params_infos in self.MODULES_EXTRA_PARAMS]
writer = csv.writer(csvfile, delimiter=';')
writer.writerow(['Vendor', 'product_code', 'revision_number'] + extra_params)
for (vendor, product_code, revision_number), module_extra_params in self.ModulesExtraParams.iteritems():
writer.writerow([vendor, product_code, revision_number] +
[module_extra_params.get(param, '')
for param in extra_params])
csvfile.close()
def SetModuleExtraParam(self, vendor, product_code, revision_number, param, value):
vendor = ExtractHexDecValue(vendor)
product_code = ExtractHexDecValue(product_code)
revision_number = ExtractHexDecValue(revision_number)
module_infos = (vendor, product_code, revision_number)
self.ModulesExtraParams.setdefault(module_infos, {})
self.ModulesExtraParams[module_infos][param] = value
self.SaveModulesExtraParams()
def GetModuleExtraParams(self, vendor, product_code, revision_number):
vendor = ExtractHexDecValue(vendor)
product_code = ExtractHexDecValue(product_code)
revision_number = ExtractHexDecValue(revision_number)
if self.ParentLibrary is not None:
extra_params = self.ParentLibrary.GetModuleExtraParams(vendor, product_code, revision_number)
else:
extra_params = {}
extra_params.update(self.ModulesExtraParams.get((vendor, product_code, revision_number), {}))
for param, param_infos in self.MODULES_EXTRA_PARAMS:
extra_params.setdefault(param, param_infos["default"])
return extra_params
USERDATA_DIR = wx.StandardPaths.Get().GetUserDataDir()
if wx.Platform != '__WXMSW__':
USERDATA_DIR += '_files'
ModulesDatabase = ModulesLibrary(
os.path.join(USERDATA_DIR, "ethercat_modules"))
class RootClass(object):
CTNChildrenTypes = [("EthercatNode", _EthercatCTN, "Ethercat Master")]
EditorType = LibraryEditor
def __init__(self):
self.ModulesLibrary = None
self.LoadModulesLibrary()
def GetIconName(self):
return "Ethercat"
def GetModulesLibraryPath(self, project_path=None):
if project_path is None:
project_path = self.CTNPath()
return os.path.join(project_path, "modules")
def OnCTNSave(self, from_project_path=None):
if from_project_path is not None:
shutil.copytree(self.GetModulesLibraryPath(from_project_path),
self.GetModulesLibraryPath())
return True
def CTNGenerate_C(self, buildpath, locations):
return [], "", False
def LoadModulesLibrary(self):
if self.ModulesLibrary is None:
self.ModulesLibrary = ModulesLibrary(self.GetModulesLibraryPath(), ModulesDatabase)
else:
self.ModulesLibrary.LoadModulesLibrary()
def GetModulesDatabaseInstance(self):
return ModulesDatabase
def GetModulesLibraryInstance(self):
return self.ModulesLibrary
def GetModulesLibrary(self, profile_filter=None):
return self.ModulesLibrary.GetModulesLibrary(profile_filter)
# add jblee
def GetMDPList(self):
return self.ModulesLibrary.GetMDPList()
# add jblee
def GetSelectModule(self, idx):
return self.ModulesLibrary.GetSelectModule(idx)
# add jblee
def GetModuleEntryList(self):
return self.ModulesLibrary.GetModuleEntryList()
# add jblee
def GetModuleIncrement(self):
return self.ModulesLibrary.GetModuleIncrement()
# add jblee
#def GetEntriesList(self, limits = None):
# return self.ModulesLibrary.GetEntriesList()
def GetVendors(self):
return self.ModulesLibrary.GetVendors()
def GetModuleInfos(self, module_infos):
return self.ModulesLibrary.GetModuleInfos(module_infos)
def GetMDPInfos(self, module_infos):
return self.ModulesLibrary.GetMDPInfos(module_infos)