Adding support for uploading default value for unused entries in output pdos from slaves before starting master
import os, shutil
import cPickle
from xml.dom import minidom
import wx
from xmlclass import *
from plugger import PlugTemplate
from PLCControler import UndoBuffer, LOCATION_PLUGIN, LOCATION_MODULE, LOCATION_GROUP, LOCATION_VAR_INPUT, LOCATION_VAR_OUTPUT, LOCATION_VAR_MEMORY
from ConfigEditor import ConfigEditor, DS402NodeEditor, ETHERCAT_VENDOR, ETHERCAT_GROUP, ETHERCAT_DEVICE
#--------------------------------------------------
# Ethercat DS402 Node
#--------------------------------------------------
NODE_VARIABLES = [
("ControlWord", 0x6040, 0x00, "UINT", "Q"),
("TargetPosition", 0x607a, 0x00, "DINT", "Q"),
("StatusWord", 0x6041, 0x00, "UINT", "I"),
("ModesOfOperationDisplay", 0x06061, 0x00, "SINT", "I"),
("ActualPosition", 0x6064, 0x00, "DINT", "I"),
("ErrorCode", 0x603f, 0x00, "UINT", "I"),
]
class _EthercatDS402SlavePlug:
XSD = """<?xml version="1.0" encoding="ISO-8859-1" ?>
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<xsd:element name="EtherlabDS402Slave">
<xsd:complexType>
<xsd:attribute name="Node_Type" type="xsd:string" use="optional"/>
</xsd:complexType>
</xsd:element>
</xsd:schema>
"""
EditorType = DS402NodeEditor
def ExtractHexDecValue(self, value):
return ExtractHexDecValue(value)
def GetSizeOfType(self, type):
return TYPECONVERSION.get(self.GetPlugRoot().GetBaseType(type), None)
def _GetChildBySomething(self, something, toks):
return self
def GetParamsAttributes(self, path = None):
infos = PlugTemplate.GetParamsAttributes(self, path = None)
for element in infos:
if element["name"] == "EtherlabDS402Slave":
for child in element["children"]:
if child["name"] == "Node_Type":
child["type"] = [module[0] for module in self.PlugParent.GetModulesByProfile(402)]
return infos
def GetAllChannels(self):
AllChannels = PlugTemplate.GetAllChannels(self)
for slave_pos in self.PlugParent.GetSlaves():
if slave_pos[0] not in AllChannels:
AllChannels.append(slave_pos[0])
AllChannels.sort()
return AllChannels
def GetCurrentLocation(self):
"""
@return: Tupple containing plugin IEC location of current plugin : %I0.0.4.5 => (0,0,4,5)
"""
return self.PlugParent.GetCurrentLocation() + self.GetSlavePos()
def GetSlavePos(self):
return self.BaseParams.getIEC_Channel(), 0
def GetSlaveTypeInfos(self):
slave_type = self.EtherlabDS402Slave.getNode_Type()
for module_type, vendor_id, product_code, revision_number in self.PlugParent.GetModulesByProfile(402):
if module_type == slave_type:
return {"device_type": module_type,
"vendor": GenerateHexDecValue(vendor_id),
"product_code": GenerateHexDecValue(product_code, 16),
"revision_number": GenerateHexDecValue(revision_number, 16)}
return None
def GetSlaveInfos(self):
slave_typeinfos = self.GetSlaveTypeInfos()
if slave_typeinfos is not None:
device = self.PlugParent.GetModuleInfos(slave_typeinfos)
if device is not None:
infos = slave_typeinfos.copy()
entries = device.GetEntriesList()
entries_list = entries.items()
entries_list.sort()
entries = []
current_index = None
current_entry = None
for (index, subindex), entry in entries_list:
entry["children"] = []
if index != current_index:
current_index = index
current_entry = entry
entries.append(entry)
elif current_entry is not None:
current_entry["children"].append(entry)
else:
entries.append(entry)
infos.update({"physics": device.getPhysics(),
"sync_managers": device.GetSyncManagers(),
"entries": entries})
return infos
return None
def GetVariableLocationTree(self):
slave_typeinfos = self.GetSlaveTypeInfos()
vars = []
if slave_typeinfos is not None:
vars = self.PlugParent.GetDeviceLocationTree(self.GetCurrentLocation(), slave_typeinfos)
return {"name": self.BaseParams.getName(),
"type": LOCATION_PLUGIN,
"location": self.GetFullIEC_Channel(),
"children": vars}
PluginMethods = [
{"bitmap" : os.path.join("images", "EditCfile"),
"name" : _("Edit Node"),
"tooltip" : _("Edit Node"),
"method" : "_OpenView"},
]
def PlugGenerate_C(self, buildpath, locations):
"""
Generate C code
@param current_location: Tupple containing plugin IEC location : %I0.0.4.5 => (0,0,4,5)
@param locations: List of complete variables locations \
[{"IEC_TYPE" : the IEC type (i.e. "INT", "STRING", ...)
"NAME" : name of the variable (generally "__IW0_1_2" style)
"DIR" : direction "Q","I" or "M"
"SIZE" : size "X", "B", "W", "D", "L"
"LOC" : tuple of interger for IEC location (0,1,2,...)
}, ...]
@return: [(C_file_name, CFLAGS),...] , LDFLAGS_TO_APPEND
"""
current_location = self.GetCurrentLocation()
location_str = "_".join(map(lambda x:str(x), current_location))
slave_pos = current_location[-2:]
slave_typeinfos = self.GetSlaveTypeInfos()
device = None
if slave_typeinfos is not None:
device = self.PlugParent.GetModuleInfos(slave_typeinfos)
if device is None:
raise (ValueError,
_("No information found for DS402 node \"%s\" at location %s!") % (
slave_typeinfos["device_type"], ".".join(current_location)))
self.PlugParent.FileGenerator.DeclareSlave(slave_pos, slave_typeinfos)
plc_ds402node_filepath = os.path.join(os.path.split(__file__)[0], "plc_ds402node.c")
plc_ds402node_file = open(plc_ds402node_filepath, 'r')
plc_ds402node_code = plc_ds402node_file.read()
plc_ds402node_file.close()
from plugins.motion import Headers
str_completion = {
"location": location_str,
"MCL_headers": Headers,
"extern_located_variables_declaration": [],
"entry_variables": [],
"init_entry_variables": [],
}
for variable in NODE_VARIABLES:
var_infos = dict(zip(["name", "index", "subindex", "var_type", "dir"], variable))
var_infos["location"] = location_str
var_infos["var_size"] = self.PlugParent.GetSizeOfType(var_infos["var_type"])
var_infos["var_name"] = "__%(dir)s%(var_size)s%(location)s_%(index)d_%(subindex)d" % var_infos
str_completion["extern_located_variables_declaration"].append(
"IEC_%(var_type)s *%(var_name)s;" % var_infos)
str_completion["entry_variables"].append(
" IEC_%(var_type)s *%(name)s;" % var_infos)
str_completion["init_entry_variables"].append(
" __DS402Node_%(location)s.%(name)s = %(var_name)s;" % var_infos)
self.PlugParent.FileGenerator.DeclareVariable(
slave_pos, var_infos["index"], var_infos["subindex"],
var_infos["var_type"], var_infos["dir"], var_infos["var_name"])
for element in ["extern_located_variables_declaration",
"entry_variables",
"init_entry_variables"]:
str_completion[element] = "\n".join(str_completion[element])
Gen_DS402Nodefile_path = os.path.join(buildpath, "ds402node_%s.c"%location_str)
ds402nodefile = open(Gen_DS402Nodefile_path, 'w')
ds402nodefile.write(plc_ds402node_code % str_completion)
ds402nodefile.close()
return [(Gen_DS402Nodefile_path, '"-I%s"'%os.path.abspath(self.GetPlugRoot().GetIECLibPath()))],"",True
TYPECONVERSION = {"BOOL" : "X", "SINT" : "B", "INT" : "W", "DINT" : "D", "LINT" : "L",
"USINT" : "B", "UINT" : "W", "UDINT" : "D", "ULINT" : "L",
"BYTE" : "B", "WORD" : "W", "DWORD" : "D", "LWORD" : "L"}
DATATYPECONVERSION = {"BOOL" : "BIT", "SINT" : "S8", "INT" : "S16", "DINT" : "S32", "LINT" : "S64",
"USINT" : "U8", "UINT" : "U16", "UDINT" : "U32", "ULINT" : "U64",
"BYTE" : "U8", "WORD" : "U16", "DWORD" : "U32", "LWORD" : "U64"}
VARCLASSCONVERSION = {"T": LOCATION_VAR_INPUT, "R": LOCATION_VAR_OUTPUT, "RT": LOCATION_VAR_MEMORY}
#--------------------------------------------------
# Ethercat MASTER
#--------------------------------------------------
EtherCATConfigClasses = GenerateClassesFromXSD(os.path.join(os.path.dirname(__file__), "EtherCATConfig.xsd"))
def ExtractHexDecValue(value):
try:
return int(value)
except:
pass
try:
return int(value.replace("#", "0"), 16)
except:
raise ValueError, "Invalid value for HexDecValue \"%s\"" % value
def GenerateHexDecValue(value, base=10):
if base == 10:
return str(value)
elif base == 16:
return "#x%.8x" % value
else:
raise ValueError, "Not supported base"
cls = EtherCATConfigClasses.get("Config_Slave", None)
if cls:
def getType(self):
slave_info = self.getInfo()
return {"device_type": slave_info.getName(),
"vendor": GenerateHexDecValue(slave_info.getVendorId()),
"product_code": GenerateHexDecValue(slave_info.getProductCode(), 16),
"revision_number": GenerateHexDecValue(slave_info.getRevisionNo(), 16)}
setattr(cls, "getType", getType)
def setType(self, type_infos):
slave_info = self.getInfo()
slave_info.setName(type_infos["device_type"])
slave_info.setVendorId(ExtractHexDecValue(type_infos["vendor"]))
slave_info.setProductCode(ExtractHexDecValue(type_infos["product_code"]))
slave_info.setRevisionNo(ExtractHexDecValue(type_infos["revision_number"]))
setattr(cls, "setType", setType)
cls = EtherCATConfigClasses.get("Slave_Info", None)
if cls:
def getSlavePosition(self):
return self.getPhysAddr(), self.getAutoIncAddr()
setattr(cls, "getSlavePosition", getSlavePosition)
def setSlavePosition(self, alias, pos):
self.setPhysAddr(alias)
self.setAutoIncAddr(pos)
setattr(cls, "setSlavePosition", setSlavePosition)
class _EthercatPlug:
XSD = """<?xml version="1.0" encoding="ISO-8859-1" ?>
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<xsd:element name="EtherlabNode">
<xsd:complexType>
<xsd:attribute name="MasterNumber" type="xsd:integer" use="optional" default="0"/>
<xsd:attribute name="ConfigurePDOs" type="xsd:boolean" use="optional" default="true"/>
</xsd:complexType>
</xsd:element>
</xsd:schema>
"""
EditorType = ConfigEditor
PlugChildsTypes = [("EthercatDS402Slave", _EthercatDS402SlavePlug, "Ethercat DS402 Slave")]
def __init__(self):
filepath = self.ConfigFileName()
self.Config = EtherCATConfigClasses["EtherCATConfig"]()
if os.path.isfile(filepath):
xmlfile = open(filepath, 'r')
tree = minidom.parse(xmlfile)
xmlfile.close()
for child in tree.childNodes:
if child.nodeType == tree.ELEMENT_NODE and child.nodeName == "EtherCATConfig":
self.Config.loadXMLTree(child, ["xmlns:xsi", "xsi:noNamespaceSchemaLocation"])
self.CreateConfigBuffer(True)
else:
self.CreateConfigBuffer(False)
self.OnPlugSave()
def ExtractHexDecValue(self, value):
return ExtractHexDecValue(value)
def GetSizeOfType(self, type):
return TYPECONVERSION.get(self.GetPlugRoot().GetBaseType(type), None)
def ConfigFileName(self):
return os.path.join(self.PlugPath(), "config.xml")
def GetSlaves(self):
slaves = []
for slave in self.Config.getConfig().getSlave():
slaves.append(slave.getInfo().getSlavePosition())
slaves.sort()
return slaves
def GetSlave(self, slave_pos):
for slave in self.Config.getConfig().getSlave():
slave_info = slave.getInfo()
if slave_info.getSlavePosition() == slave_pos:
return slave
return None
def AddSlave(self):
slaves = self.GetSlaves()
for PlugInstance in self.IterChilds():
slaves.append(PlugInstance.GetSlavePos())
slaves.sort()
if len(slaves) > 0:
new_pos = (slaves[-1][0] + 1, 0)
else:
new_pos = (0, 0)
slave = EtherCATConfigClasses["Config_Slave"]()
slave_infos = slave.getInfo()
slave_infos.setName("undefined")
slave_infos.setSlavePosition(new_pos[0], new_pos[1])
self.Config.getConfig().appendSlave(slave)
self.BufferConfig()
return new_pos
def RemoveSlave(self, slave_pos):
config = self.Config.getConfig()
for idx, slave in enumerate(config.getSlave()):
slave_infos = slave.getInfo()
if slave_infos.getSlavePosition() == slave_pos:
config.removeSlave(idx)
self.BufferConfig()
return True
return False
def SetSlavePos(self, slave_pos, alias=None, position=None):
for PlugInstance in self.IterChilds():
if PlugInstance.BaseParams.getIEC_Channel() == alias:
return _("Slave with alias \"%d\" already exists!" % alias)
slave = self.GetSlave(slave_pos)
if slave is not None:
slave_info = slave.getInfo()
new_pos = slave_pos
if alias is not None:
new_pos = (alias, new_pos[1])
if position is not None:
new_pos = (new_pos[0], position)
if self.GetSlave(new_pos) is not None:
return _("Slave with position \"%d:%d\" already exists!" % new_pos)
slave_info.setSlavePosition(*new_pos)
self.BufferConfig()
return None
def GetSlaveType(self, slave_pos):
slave = self.GetSlave(slave_pos)
if slave is not None:
return slave.getType()
return None
def SetSlaveType(self, slave_pos, type_infos):
slave = self.GetSlave(slave_pos)
if slave is not None:
slave.setType(type_infos)
self.BufferConfig()
return None
def GetSlaveInfos(self, slave_pos):
slave = self.GetSlave(slave_pos)
if slave is not None:
type_infos = slave.getType()
device = self.GetModuleInfos(type_infos)
if device is not None:
infos = type_infos.copy()
entries = device.GetEntriesList()
entries_list = entries.items()
entries_list.sort()
entries = []
current_index = None
current_entry = None
for (index, subindex), entry in entries_list:
entry["children"] = []
if index != current_index:
current_index = index
current_entry = entry
entries.append(entry)
elif current_entry is not None:
current_entry["children"].append(entry)
else:
entries.append(entry)
infos.update({"physics": device.getPhysics(),
"sync_managers": device.GetSyncManagers(),
"entries": entries})
return infos
return None
def GetModuleInfos(self, type_infos):
return self.PlugParent.GetModuleInfos(type_infos)
def GetModulesByProfile(self, profile_type):
return self.PlugParent.GetModulesByProfile(profile_type)
def GetSlaveTypesLibrary(self):
return self.PlugParent.GetModulesLibrary()
def GetDeviceLocationTree(self, current_location, type_infos):
vars = []
device = self.GetModuleInfos(type_infos)
if device is not None:
sync_managers = []
for sync_manager in device.getSm():
sync_manager_control_byte = ExtractHexDecValue(sync_manager.getControlByte())
sync_manager_direction = sync_manager_control_byte & 0x0c
if sync_manager_direction:
sync_managers.append(LOCATION_VAR_OUTPUT)
else:
sync_managers.append(LOCATION_VAR_INPUT)
entries = device.GetEntriesList().items()
entries.sort()
for (index, subindex), entry in entries:
var_size = self.GetSizeOfType(entry["Type"])
if var_size is not None:
var_class = VARCLASSCONVERSION.get(entry["PDOMapping"], None)
if var_class is not None:
if var_class == LOCATION_VAR_INPUT:
var_dir = "%I"
else:
var_dir = "%Q"
vars.append({"name": "0x%4.4x-0x%2.2x: %s" % (index, subindex, entry["Name"]),
"type": var_class,
"size": var_size,
"IEC_type": entry["Type"],
"var_name": "%s_%4.4x_%2.2x" % ("_".join(type_infos["device_type"].split()), index, subindex),
"location": "%s%s%s"%(var_dir, var_size, ".".join(map(str, current_location +
(index, subindex)))),
"description": "",
"children": []})
return vars
def GetVariableLocationTree(self):
'''See PlugTemplate.GetVariableLocationTree() for a description.'''
current_location = self.GetCurrentLocation()
groups = []
for slave_pos in self.GetSlaves():
slave = self.GetSlave(slave_pos)
if slave is not None:
type_infos = slave.getType()
vars = self.GetDeviceLocationTree(current_location + slave_pos, type_infos)
if len(vars) > 0:
groups.append({"name": "%s (%d,%d)" % ((type_infos["device_type"],) + slave_pos),
"type": LOCATION_GROUP,
"location": ".".join(map(str, current_location + slave_pos)) + ".x",
"children": vars})
return {"name": self.BaseParams.getName(),
"type": LOCATION_PLUGIN,
"location": self.GetFullIEC_Channel(),
"children": groups}
PluginMethods = [
{"bitmap" : os.path.join("images", "EditCfile"),
"name" : _("Edit Config"),
"tooltip" : _("Edit Config"),
"method" : "_OpenView"},
]
def PlugTestModified(self):
return self.ChangesToSave or not self.ConfigIsSaved()
def OnPlugSave(self):
filepath = self.ConfigFileName()
text = "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?>\n"
extras = {"xmlns:xsi":"http://www.w3.org/2001/XMLSchema-instance",
"xsi:noNamespaceSchemaLocation" : "EtherCATInfo.xsd"}
text += self.Config.generateXMLText("EtherCATConfig", 0, extras)
xmlfile = open(filepath,"w")
xmlfile.write(text.encode("utf-8"))
xmlfile.close()
self.ConfigBuffer.CurrentSaved()
return True
def _Generate_C(self, buildpath, locations):
current_location = self.GetCurrentLocation()
# define a unique name for the generated C file
location_str = "_".join(map(lambda x:str(x), current_location))
Gen_Ethercatfile_path = os.path.join(buildpath, "ethercat_%s.c"%location_str)
self.FileGenerator = _EthercatCFileGenerator(self)
LocationCFilesAndCFLAGS, LDFLAGS, extra_files = PlugTemplate._Generate_C(self, buildpath, locations)
self.FileGenerator.GenerateCFile(Gen_Ethercatfile_path, location_str, self.EtherlabNode)
LocationCFilesAndCFLAGS.append(
(current_location,
[(Gen_Ethercatfile_path, '"-I%s"'%os.path.abspath(self.GetPlugRoot().GetIECLibPath()))],
True))
LDFLAGS.append("-lethercat -lrtdm")
return LocationCFilesAndCFLAGS, LDFLAGS, extra_files
def PlugGenerate_C(self, buildpath, locations):
"""
Generate C code
@param current_location: Tupple containing plugin IEC location : %I0.0.4.5 => (0,0,4,5)
@param locations: List of complete variables locations \
[{"IEC_TYPE" : the IEC type (i.e. "INT", "STRING", ...)
"NAME" : name of the variable (generally "__IW0_1_2" style)
"DIR" : direction "Q","I" or "M"
"SIZE" : size "X", "B", "W", "D", "L"
"LOC" : tuple of interger for IEC location (0,1,2,...)
}, ...]
@return: [(C_file_name, CFLAGS),...] , LDFLAGS_TO_APPEND
"""
current_location = self.GetCurrentLocation()
slaves = self.GetSlaves()
for slave_pos in slaves:
slave = self.GetSlave(slave_pos)
if slave is not None:
self.FileGenerator.DeclareSlave(slave_pos, slave.getType())
for location in locations:
loc = location["LOC"][len(current_location):]
slave_pos = loc[:2]
if slave_pos in slaves:
self.FileGenerator.DeclareVariable(
slave_pos, loc[2], loc[3], location["IEC_TYPE"], location["DIR"], location["NAME"])
return [],"",False
#-------------------------------------------------------------------------------
# Current Buffering Management Functions
#-------------------------------------------------------------------------------
"""
Return a copy of the config
"""
def Copy(self, model):
return cPickle.loads(cPickle.dumps(model))
def CreateConfigBuffer(self, saved):
self.ConfigBuffer = UndoBuffer(cPickle.dumps(self.Config), saved)
def BufferConfig(self):
self.ConfigBuffer.Buffering(cPickle.dumps(self.Config))
def ConfigIsSaved(self):
if self.ConfigBuffer is not None:
return self.ConfigBuffer.IsCurrentSaved()
else:
return True
def LoadPrevious(self):
self.Config = cPickle.loads(self.ConfigBuffer.Previous())
def LoadNext(self):
self.Config = cPickle.loads(self.ConfigBuffer.Next())
def GetBufferState(self):
first = self.ConfigBuffer.IsFirst()
last = self.ConfigBuffer.IsLast()
return not first, not last
SLAVE_PDOS_CONFIGURATION_DECLARATION = """
/* Slave %(slave)d, "%(device_type)s"
* Vendor ID: 0x%(vendor).8x
* Product code: 0x%(product_code).8x
* Revision number: 0x%(revision_number).8x
*/
ec_pdo_entry_info_t slave_%(slave)d_pdo_entries[] = {
%(pdos_entries_infos)s
};
ec_pdo_info_t slave_%(slave)d_pdos[] = {
%(pdos_infos)s
};
ec_sync_info_t slave_%(slave)d_syncs[] = {
%(pdos_sync_infos)s
{0xff}
};
"""
SLAVE_CONFIGURATION_TEMPLATE = """
if (!(slave%(slave)d = ecrt_master_slave_config(master, %(alias)d, %(position)d, 0x%(vendor).8x, 0x%(product_code).8x))) {
fprintf(stderr, "Failed to get slave %(device_type)s configuration at alias %(alias)d and position %(position)d.\\n");
return -1;
}
if (ecrt_slave_config_pdos(slave%(slave)d, EC_END, slave_%(slave)d_syncs)) {
fprintf(stderr, "Failed to configure PDOs for slave %(device_type)s at alias %(alias)d and position %(position)d.\\n");
return -1;
}
"""
SLAVE_INITIALIZATION_TEMPLATE = """
{
uint8_t value[] = {%(data)s};
if (ecrt_master_sdo_download(master, %(slave)d, 0x%(index).4x, 0x%(subindex).2x, (uint8_t *)value, %(data_size)d, &abort_code)) {
fprintf(stderr, "Failed to initialize slave %(device_type)s at alias %(alias)d and position %(position)d.\\nError: %%d\\n", abort_code);
return -1;
}
}
"""
SLAVE_OUTPUT_PDO_DEFAULT_VALUE = """
{
uint8_t value[%(data_size)d];
if (ecrt_master_sdo_upload(master, %(slave)d, 0x%(index).4x, 0x%(subindex).2x, (uint8_t *)value, %(data_size)d, &result_size, &abort_code)) {
fprintf(stderr, "Failed to get default value for output PDO in slave %(device_type)s at alias %(alias)d and position %(position)d.\\nError: %%d\\n", abort_code);
return -1;
}
for (i = 0; i < result_size; i++) {
%(real_var)s = (%(real_var)s << 8) + value[i];
}
}
"""
def ConfigureVariable(entry_infos, str_completion):
entry_infos["data_type"] = DATATYPECONVERSION.get(entry_infos["var_type"], None)
if entry_infos["data_type"] is None:
raise ValueError, _("Type of location \"%s\" not yet supported!") % entry_infos["var_name"]
if entry_infos.has_key("real_var"):
str_completion["located_variables_declaration"].append(
"IEC_%(var_type)s %(real_var)s;" % entry_infos)
else:
entry_infos["real_var"] = "beremiz" + entry_infos["var_name"]
str_completion["located_variables_declaration"].extend(
["IEC_%(var_type)s %(real_var)s;" % entry_infos,
"IEC_%(var_type)s *%(var_name)s = &%(real_var)s;" % entry_infos])
str_completion["used_pdo_entry_offset_variables_declaration"].append(
"unsigned int slave%(slave)d_%(index).4x_%(subindex).2x;" % entry_infos)
if entry_infos["data_type"] == "BIT":
str_completion["used_pdo_entry_offset_variables_declaration"].append(
"unsigned int slave%(slave)d_%(index).4x_%(subindex).2x_bit;" % entry_infos)
str_completion["used_pdo_entry_configuration"].append(
(" {%(alias)d, %(position)d, 0x%(vendor).8x, 0x%(product_code).8x, " +
"0x%(index).4x, %(subindex)d, &slave%(slave)d_%(index).4x_%(subindex).2x, " +
"&slave%(slave)d_%(index).4x_%(subindex).2x_bit},") % entry_infos)
if entry_infos["dir"] == "I":
str_completion["retrieve_variables"].append(
(" %(real_var)s = EC_READ_BIT(domain1_pd + slave%(slave)d_%(index).4x_%(subindex).2x, " +
"slave%(slave)d_%(index).4x_%(subindex).2x_bit);") % entry_infos)
elif entry_infos["dir"] == "Q":
str_completion["publish_variables"].append(
(" EC_WRITE_BIT(domain1_pd + slave%(slave)d_%(index).4x_%(subindex).2x, " +
"slave%(slave)d_%(index).4x_%(subindex).2x_bit, %(real_var)s);") % entry_infos)
else:
str_completion["used_pdo_entry_configuration"].append(
(" {%(alias)d, %(position)d, 0x%(vendor).8x, 0x%(product_code).8x, 0x%(index).4x, " +
"%(subindex)d, &slave%(slave)d_%(index).4x_%(subindex).2x},") % entry_infos)
if entry_infos["dir"] == "I":
str_completion["retrieve_variables"].append(
(" %(real_var)s = EC_READ_%(data_type)s(domain1_pd + " +
"slave%(slave)d_%(index).4x_%(subindex).2x);") % entry_infos)
elif entry_infos["dir"] == "Q":
str_completion["publish_variables"].append(
(" EC_WRITE_%(data_type)s(domain1_pd + slave%(slave)d_%(index).4x_%(subindex).2x, " +
"%(real_var)s);") % entry_infos)
class _EthercatCFileGenerator:
def __init__(self, controler):
self.Controler = controler
self.Slaves = []
self.UsedVariables = {}
def __del__(self):
self.Controler = None
def DeclareSlave(self, slave_identifier, slave):
self.Slaves.append((slave_identifier, slave))
self.Slaves.sort()
return self.Slaves.index((slave_identifier, slave))
def DeclareVariable(self, slave_identifier, index, subindex, iec_type, dir, name):
slave_variables = self.UsedVariables.setdefault(slave_identifier, {})
entry_infos = slave_variables.get((index, subindex), None)
if entry_infos is None:
slave_variables[(index, subindex)] = {
"infos": (iec_type, dir, name),
"mapped": False}
elif entry_infos["infos"] != (iec_type, dir, name):
raise ValueError, _("Definition conflict for location \"%s\"") % name
def GenerateCFile(self, filepath, location_str, etherlab_node_infos):
plc_etherlab_filepath = os.path.join(os.path.split(__file__)[0], "plc_etherlab.c")
plc_etherlab_file = open(plc_etherlab_filepath, 'r')
plc_etherlab_code = plc_etherlab_file.read()
plc_etherlab_file.close()
str_completion = {
"location": location_str,
"configure_pdos": int(etherlab_node_infos.getConfigurePDOs()),
"master_number": etherlab_node_infos.getMasterNumber(),
"located_variables_declaration": [],
"used_pdo_entry_offset_variables_declaration": [],
"used_pdo_entry_configuration": [],
"pdos_configuration_declaration": "",
"slaves_declaration": "",
"slaves_configuration": "",
"slaves_output_pdos_default_values_extraction": "",
"slaves_initialization": "",
"retrieve_variables": [],
"publish_variables": [],
}
for slave_entries in self.UsedVariables.itervalues():
for entry_infos in slave_entries.itervalues():
entry_infos["mapped"] = False
for slave_idx, (slave_pos, type_infos) in enumerate(self.Slaves):
device = self.Controler.GetModuleInfos(type_infos)
if device is not None:
slave_variables = self.UsedVariables.get(slave_pos, {})
device_entries = device.GetEntriesList()
if len(device.getTxPdo() + device.getRxPdo()) > 0 or len(slave_variables) > 0:
for element in ["vendor", "product_code", "revision_number"]:
type_infos[element] = ExtractHexDecValue(type_infos[element])
type_infos.update(dict(zip(["slave", "alias", "position"], (slave_idx,) + slave_pos)))
str_completion["slaves_declaration"] += "static ec_slave_config_t *slave%(slave)d = NULL;\n" % type_infos
str_completion["slaves_configuration"] += SLAVE_CONFIGURATION_TEMPLATE % type_infos
for initCmd in device.getInitCmd():
index = ExtractHexDecValue(initCmd.getIndex())
subindex = ExtractHexDecValue(initCmd.getSubIndex())
entry = device_entries.get((index, subindex), None)
if entry is not None:
data_size = entry["BitSize"] / 8
data = ("%%.%dx" % (data_size * 2)) % initCmd.getData().getcontent()
data_str = ",".join(["0x%s" % data[i:i+2] for i in xrange(0, data_size * 2, 2)])
init_cmd_infos = {
"index": index,
"subindex": subindex,
"data": data_str,
"data_size": data_size
}
init_cmd_infos.update(type_infos)
str_completion["slaves_initialization"] += SLAVE_INITIALIZATION_TEMPLATE % init_cmd_infos
pdos_infos = {
"pdos_entries_infos": [],
"pdos_infos": [],
"pdos_sync_infos": [],
}
pdos_infos.update(type_infos)
sync_managers = []
for sync_manager_idx, sync_manager in enumerate(device.getSm()):
sync_manager_infos = {
"index": sync_manager_idx,
"name": sync_manager.getcontent(),
"slave": slave_idx,
"pdos": [],
"pdos_number": 0,
}
sync_manager_control_byte = ExtractHexDecValue(sync_manager.getControlByte())
sync_manager_direction = sync_manager_control_byte & 0x0c
sync_manager_watchdog = sync_manager_control_byte & 0x40
if sync_manager_direction:
sync_manager_infos["sync_manager_type"] = "EC_DIR_OUTPUT"
else:
sync_manager_infos["sync_manager_type"] = "EC_DIR_INPUT"
if sync_manager_watchdog:
sync_manager_infos["watchdog"] = "EC_WD_ENABLE"
else:
sync_manager_infos["watchdog"] = "EC_WD_DISABLE"
sync_managers.append(sync_manager_infos)
pdos_index = []
excluded_pdos = []
for only_mandatory in [True, False]:
for pdo, pdo_type in ([(pdo, "Inputs") for pdo in device.getTxPdo()] +
[(pdo, "Outputs") for pdo in device.getRxPdo()]):
entries = pdo.getEntry()
pdo_needed = pdo.getMandatory()
if pdo_needed is None:
pdo_needed = False
if only_mandatory != pdo_needed:
continue
pdo_index = ExtractHexDecValue(pdo.getIndex().getcontent())
pdos_index.append(pdo_index)
if pdo_index in excluded_pdos:
continue
entries_infos = []
for entry in entries:
index = ExtractHexDecValue(entry.getIndex().getcontent())
subindex = ExtractHexDecValue(entry.getSubIndex())
entry_infos = {
"index": index,
"subindex": subindex,
"name": ExtractName(entry.getName()),
"bitlen": entry.getBitLen(),
}
entry_infos.update(type_infos)
entries_infos.append(" {0x%(index).4x, 0x%(subindex).2x, %(bitlen)d}, /* %(name)s */" % entry_infos)
entry_declaration = slave_variables.get((index, subindex), None)
if entry_declaration is not None and not entry_declaration["mapped"]:
pdo_needed = True
entry_infos.update(dict(zip(["var_type", "dir", "var_name"], entry_declaration["infos"])))
entry_declaration["mapped"] = True
if entry_infos["var_type"] != entry.getDataType().getcontent():
raise ValueError, _("Wrong type for location \"%s\"!") % entry_infos["var_name"]
if (entry_infos["dir"] == "I" and pdo_type != "Inputs" or
entry_infos["dir"] == "Q" and pdo_type != "Outputs"):
raise ValueError, _("Wrong direction for location \"%s\"!") % entry_infos["var_name"]
ConfigureVariable(entry_infos, str_completion)
elif pdo_type == "Outputs" and entry.getDataType() is not None:
entry_infos["dir"] = "Q"
entry_infos["data_size"] = max(1, entry_infos["bitlen"] / 8)
entry_infos["var_type"] = entry.getDataType().getcontent()
entry_infos["real_var"] = "slave%(slave)d_%(index).4x_%(subindex).2x_default" % entry_infos
ConfigureVariable(entry_infos, str_completion)
str_completion["slaves_output_pdos_default_values_extraction"] += \
SLAVE_OUTPUT_PDO_DEFAULT_VALUE % entry_infos
if pdo_needed:
for excluded in pdo.getExclude():
excluded_index = ExtractHexDecValue(excluded.getcontent())
if excluded_index not in excluded_pdos:
excluded_pdos.append(excluded_index)
sm = pdo.getSm()
if sm is None:
for sm_idx, sync_manager in enumerate(sync_managers):
if sync_manager["name"] == pdo_type:
sm = sm_idx
if sm is None:
raise ValueError, _("No sync manager available for %s pdo!") % pdo_type
sync_managers[sm]["pdos_number"] += 1
sync_managers[sm]["pdos"].append(
{"slave": slave_idx,
"index": pdo_index,
"name": ExtractName(pdo.getName()),
"type": pdo_type,
"entries": entries_infos,
"entries_number": len(entries_infos),
"fixed": pdo.getFixed() == True})
dynamic_pdos = {}
dynamic_pdos_number = 0
for category, min_index, max_index in [("Inputs", 0x1600, 0x1800),
("Outputs", 0x1a00, 0x1C00)]:
for sync_manager in sync_managers:
if sync_manager["name"] == category:
category_infos = dynamic_pdos.setdefault(category, {})
category_infos["sync_manager"] = sync_manager
category_infos["pdos"] = [pdo for pdo in category_infos["sync_manager"]["pdos"]
if not pdo["fixed"] and pdo["type"] == category]
category_infos["current_index"] = min_index
category_infos["max_index"] = max_index
break
for (index, subindex), entry_declaration in slave_variables.iteritems():
if not entry_declaration["mapped"]:
entry = device_entries.get((index, subindex), None)
if entry is None:
raise ValueError, _("Unknown entry index 0x%4.4x, subindex 0x%2.2x for device %s") % \
(index, subindex, type_infos["device_type"])
entry_infos = {
"index": index,
"subindex": subindex,
"name": entry["Name"],
"bitlen": entry["BitSize"],
}
entry_infos.update(type_infos)
entry_infos.update(dict(zip(["var_type", "dir", "var_name", "real_var"], entry_declaration["infos"])))
entry_declaration["mapped"] = True
if entry_infos["var_type"] != entry["Type"]:
raise ValueError, _("Wrong type for location \"%s\"!") % entry_infos["var_name"]
if entry_infos["dir"] == "I" and entry["PDOMapping"] in ["T", "RT"]:
pdo_type = "Inputs"
elif entry_infos["dir"] == "Q" and entry["PDOMapping"] in ["R", "RT"]:
pdo_type = "Outputs"
else:
raise ValueError, _("Wrong direction for location \"%s\"!") % entry_infos["var_name"]
if not dynamic_pdos.has_key(pdo_type):
raise ValueError, _("No Sync manager defined for %s!") % pdo_type
ConfigureVariable(entry_infos, str_completion)
if len(dynamic_pdos[pdo_type]["pdos"]) > 0:
pdo = dynamic_pdos[pdo_type]["pdos"][0]
else:
while dynamic_pdos[pdo_type]["current_index"] in pdos_index:
dynamic_pdos[pdo_type]["current_index"] += 1
if dynamic_pdos[pdo_type]["current_index"] >= dynamic_pdos[pdo_type]["max_index"]:
raise ValueError, _("No more free PDO index available for %s!") % pdo_type
pdos_index.append(dynamic_pdos[pdo_type]["current_index"])
dynamic_pdos_number += 1
pdo = {"slave": slave_idx,
"index": dynamic_pdos[pdo_type]["current_index"],
"name": "Dynamic PDO %d" % dynamic_pdos_number,
"type": pdo_type,
"entries": [],
"entries_number": 0,
"fixed": False}
dynamic_pdos[pdo_type]["sync_manager"]["pdos_number"] += 1
dynamic_pdos[pdo_type]["sync_manager"]["pdos"].append(pdo)
dynamic_pdos[pdo_type]["pdos"].append(pdo)
pdo["entries"].append(" {0x%(index).4x, 0x%(subindex).2x, %(bitlen)d}, /* %(name)s */" % entry_infos)
pdo["entries_number"] += 1
if pdo["entries_number"] == 255:
dynamic_pdos[pdo_type]["pdos"].pop(0)
pdo_offset = 0
entry_offset = 0
for sync_manager_infos in sync_managers:
for pdo_infos in sync_manager_infos["pdos"]:
pdo_infos["offset"] = entry_offset
pdo_entries = pdo_infos["entries"]
pdos_infos["pdos_infos"].append(
(" {0x%(index).4x, %(entries_number)d, " +
"slave_%(slave)d_pdo_entries + %(offset)d}, /* %(name)s */") % pdo_infos)
entry_offset += len(pdo_entries)
pdos_infos["pdos_entries_infos"].extend(pdo_entries)
sync_manager_infos["offset"] = pdo_offset
pdos_infos["pdos_sync_infos"].append(
(" {%(index)d, %(sync_manager_type)s, %(pdos_number)d, " +
"slave_%(slave)d_pdos + %(offset)d, %(watchdog)s},") % sync_manager_infos)
pdo_offset += sync_manager_infos["pdos_number"]
for element in ["pdos_entries_infos", "pdos_infos", "pdos_sync_infos"]:
pdos_infos[element] = "\n".join(pdos_infos[element])
str_completion["pdos_configuration_declaration"] += SLAVE_PDOS_CONFIGURATION_DECLARATION % pdos_infos
for element in ["used_pdo_entry_offset_variables_declaration",
"used_pdo_entry_configuration",
"located_variables_declaration",
"retrieve_variables",
"publish_variables"]:
str_completion[element] = "\n".join(str_completion[element])
etherlabfile = open(filepath, 'w')
etherlabfile.write(plc_etherlab_code % str_completion)
etherlabfile.close()
#--------------------------------------------------
# Ethercat Plugin
#--------------------------------------------------
EtherCATInfoClasses = GenerateClassesFromXSD(os.path.join(os.path.dirname(__file__), "EtherCATInfo.xsd"))
cls = EtherCATInfoClasses["EtherCATInfo.xsd"].get("DeviceType", None)
if cls:
cls.DataTypes = None
def GetProfileNumbers(self):
profiles = []
for profile in self.getProfile():
profile_content = profile.getcontent()
if profile_content is None:
continue
for content_element in profile_content["value"]:
if content_element["name"] == "ProfileNo":
profiles.append(content_element["value"])
return profiles
setattr(cls, "GetProfileNumbers", GetProfileNumbers)
def GetProfileDictionaries(self):
dictionaries = []
for profile in self.getProfile():
profile_content = profile.getcontent()
if profile_content is None:
continue
for content_element in profile_content["value"]:
if content_element["name"] == "Dictionary":
dictionaries.append(content_element["value"])
elif content_element["name"] == "DictionaryFile":
raise ValueError, "DictionaryFile for defining Device Profile is not yet supported!"
return dictionaries
setattr(cls, "GetProfileDictionaries", GetProfileDictionaries)
def ExtractDataTypes(self):
self.DataTypes = {}
for dictionary in self.GetProfileDictionaries():
datatypes = dictionary.getDataTypes()
if datatypes is not None:
for datatype in datatypes.getDataType():
content = datatype.getcontent()
if content is not None and content["name"] == "SubItem":
self.DataTypes[datatype.getName()] = datatype
setattr(cls, "ExtractDataTypes", ExtractDataTypes)
def getInitCmd(self):
mailbox = self.getMailbox()
if mailbox is None:
return []
coe = mailbox.getCoE()
if coe is None:
return []
return coe.getInitCmd()
setattr(cls, "getInitCmd", getInitCmd)
def GetEntriesList(self):
if self.DataTypes is None:
self.ExtractDataTypes()
entries = {}
for dictionary in self.GetProfileDictionaries():
for object in dictionary.getObjects().getObject():
entry_index = object.getIndex().getcontent()
index = ExtractHexDecValue(entry_index)
entry_type = object.getType()
entry_name = ExtractName(object.getName())
entry_type_infos = self.DataTypes.get(entry_type, None)
if entry_type_infos is not None:
content = entry_type_infos.getcontent()
for subitem in content["value"]:
entry_subidx = subitem.getSubIdx()
if entry_subidx is None:
entry_subidx = "0"
subidx = ExtractHexDecValue(entry_subidx)
subitem_access = ""
subitem_pdomapping = ""
subitem_flags = subitem.getFlags()
if subitem_flags is not None:
access = subitem_flags.getAccess()
if access is not None:
subitem_access = access.getcontent()
pdomapping = subitem_flags.getPdoMapping()
if pdomapping is not None:
subitem_pdomapping = pdomapping.upper()
entries[(index, subidx)] = {
"Index": entry_index,
"SubIndex": entry_subidx,
"Name": "%s - %s" %
(entry_name.decode("utf-8"),
ExtractName(subitem.getDisplayName(),
subitem.getName()).decode("utf-8")),
"Type": subitem.getType(),
"BitSize": subitem.getBitSize(),
"Access": subitem_access,
"PDOMapping": subitem_pdomapping,
"PDO index": "",
"PDO name": "",
"PDO type": ""}
else:
entry_access = ""
entry_pdomapping = ""
entry_flags = object.getFlags()
if entry_flags is not None:
access = entry_flags.getAccess()
if access is not None:
entry_access = access.getcontent()
pdomapping = entry_flags.getPdoMapping()
if pdomapping is not None:
entry_pdomapping = pdomapping.upper()
entries[(index, 0)] = {
"Index": entry_index,
"SubIndex": "0",
"Name": entry_name,
"Type": entry_type,
"BitSize": object.getBitSize(),
"Access": entry_access,
"PDOMapping": entry_pdomapping,
"PDO index": "",
"PDO name": "",
"PDO type": ""}
for TxPdo in self.getTxPdo():
ExtractPdoInfos(TxPdo, "Transmit", entries)
for RxPdo in self.getRxPdo():
ExtractPdoInfos(RxPdo, "Receive", entries)
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 ExtractName(names, default=None):
if len(names) == 1:
return names[0].getcontent()
else:
for name in names:
if name.getLcId() == 1033:
return name.getcontent()
return default
def ExtractPdoInfos(pdo, pdo_type, entries):
pdo_index = pdo.getIndex().getcontent()
pdo_name = ExtractName(pdo.getName())
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)
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(),
"Access": access,
"PDOMapping": pdomapping,
"PDO index": pdo_index,
"PDO name": pdo_name,
"PDO type": pdo_type}
class RootClass:
PlugChildsTypes = [("EthercatNode",_EthercatPlug,"Ethercat Master")]
def __init__(self):
self.LoadModulesLibrary()
def GetModulesLibraryPath(self):
library_path = os.path.join(self.PlugPath(), "modules")
if not os.path.exists(library_path):
os.mkdir(library_path)
return library_path
def _ImportModuleLibrary(self):
dialog = wx.FileDialog(self.GetPlugRoot().AppFrame, _("Choose an XML file"), os.getcwd(), "", _("XML files (*.xml)|*.xml|All files|*.*"), wx.OPEN)
if dialog.ShowModal() == wx.ID_OK:
filepath = dialog.GetPath()
if os.path.isfile(filepath):
shutil.copy(filepath, self.GetModulesLibraryPath())
self.LoadModulesLibrary()
else:
self.GetPlugRoot().logger.write_error(_("No such XML file: %s\n") % filepath)
dialog.Destroy()
PluginMethods = [
{"bitmap" : os.path.join("images", "ImportDEF"),
"name" : _("Import module library"),
"tooltip" : _("Import module library"),
"method" : "_ImportModuleLibrary"},
]
def PlugGenerate_C(self, buildpath, locations):
return [],"",False
def LoadModulesLibrary(self):
self.ModulesLibrary = {}
library_path = self.GetModulesLibraryPath()
files = os.listdir(library_path)
for file in files:
filepath = os.path.join(library_path, file)
if os.path.isfile(filepath) and os.path.splitext(filepath)[-1] == ".xml":
xmlfile = open(filepath, 'r')
xml_tree = minidom.parse(xmlfile)
xmlfile.close()
modules_infos = None
for child in xml_tree.childNodes:
if child.nodeType == xml_tree.ELEMENT_NODE and child.nodeName == "EtherCATInfo":
modules_infos = EtherCATInfoClasses["EtherCATInfo.xsd"]["EtherCATInfo"]()
modules_infos.loadXMLTree(child, ["xmlns:xsi", "xsi:noNamespaceSchemaLocation"])
if modules_infos is not None:
vendor = modules_infos.getVendor()
vendor_category = self.ModulesLibrary.setdefault(ExtractHexDecValue(vendor.getId()),
{"name": ExtractName(vendor.getName(), _("Miscellaneous")),
"groups": {}})
for group in modules_infos.getDescriptions().getGroups().getGroup():
group_type = group.getType()
vendor_category["groups"].setdefault(group_type, {"name": ExtractName(group.getName(), group_type),
"parent": group.getParentGroup(),
"order": group.getSortOrder(),
"devices": []})
for device in modules_infos.getDescriptions().getDevices().getDevice():
device_group = device.getGroupType()
if not vendor_category["groups"].has_key(device_group):
raise ValueError, "Not such group \"%\"" % device_group
vendor_category["groups"][device_group]["devices"].append((device.getType().getcontent(), device))
def GetModulesLibrary(self):
library = []
children_dict = {}
for vendor_id, vendor in self.ModulesLibrary.iteritems():
groups = []
library.append({"name": vendor["name"],
"type": ETHERCAT_VENDOR,
"children": groups})
for group_type, group in vendor["groups"].iteritems():
group_infos = {"name": group["name"],
"order": group["order"],
"type": ETHERCAT_GROUP,
"children": children_dict.setdefault(group_type, [])}
if group["parent"] is not None:
parent_children = children_dict.setdefault(group["parent"], [])
parent_children.append(group_infos)
else:
groups.append(group_infos)
device_dict = {}
for device_type, device in group["devices"]:
device_infos = {"name": ExtractName(device.getName()),
"type": ETHERCAT_DEVICE,
"infos": {"device_type": device_type,
"vendor": vendor_id,
"product_code": device.getType().getProductCode(),
"revision_number": device.getType().getRevisionNo()}}
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"]
library.sort(lambda x, y: cmp(x["name"], y["name"]))
return library
def GetModuleInfos(self, type_infos):
vendor = self.ModulesLibrary.get(ExtractHexDecValue(type_infos["vendor"]), None)
if vendor is not None:
for group_name, group in vendor["groups"].iteritems():
for device_type, device in group["devices"]:
product_code = ExtractHexDecValue(device.getType().getProductCode())
revision_number = ExtractHexDecValue(device.getType().getRevisionNo())
if (device_type == type_infos["device_type"] and
product_code == ExtractHexDecValue(type_infos["product_code"]) and
revision_number == ExtractHexDecValue(type_infos["revision_number"])):
return device
return None
def GetModulesByProfile(self, profile_type):
modules = []
for vendor_id, vendor in self.ModulesLibrary.iteritems():
for group_type, group in vendor["groups"].iteritems():
for device_type, device in group["devices"]:
if profile_type in device.GetProfileNumbers():
product_code = ExtractHexDecValue(device.getType().getProductCode())
revision_number = ExtractHexDecValue(device.getType().getRevisionNo())
modules.append((device_type, vendor_id, product_code, revision_number))
return modules