edouard@2165: #!/usr/bin/env python
edouard@2165: # -*- coding: utf-8 -*-
edouard@2165:
edouard@2165: # This file is part of Beremiz
edouard@2165: #
edouard@2165: # Copyright (C) 2011-2014: Laurent BESSARD, Edouard TISSERANT
edouard@2165: # RTES Lab : CRKim, JBLee, youcu
edouard@2165: # Higen Motor : Donggu Kang
edouard@2165: #
edouard@2165: # See COPYING file for copyrights details.
edouard@2165:
andrej@2405: from __future__ import absolute_import
andrej@2357: import os
andrej@2357: import shutil
andrej@2390: import csv
andrej@2434: from builtins import str as text
andrej@2434:
Laurent@2157: from lxml import etree
laurent@2022: import wx
laurent@2022:
laurent@2022: from xmlclass import *
Laurent@2111:
andrej@2396: from ConfigTreeNode import XSDSchemaErrorMessage
Laurent@2111:
andrej@2405: from etherlab.EthercatSlave import ExtractHexDecValue, ExtractName
andrej@2405: from etherlab.EthercatMaster import _EthercatCTN
andrej@2405: from etherlab.ConfigEditor import LibraryEditor, ETHERCAT_VENDOR, ETHERCAT_GROUP, ETHERCAT_DEVICE
laurent@2022:
Laurent@2157: ScriptDirectory = os.path.split(os.path.realpath(__file__))[0]
Laurent@2157:
andrej@2356: # --------------------------------------------------
Edouard@2048: # Ethercat ConfNode
andrej@2356: # --------------------------------------------------
laurent@2022:
andrej@2355: EtherCATInfoParser = GenerateParserFromXSD(os.path.join(os.path.dirname(__file__), "EtherCATInfo.xsd"))
andrej@2376:
andrej@2376:
andrej@2376: def EtherCATInfo_XPath(xpath):
andrej@2376: return etree.XPath(xpath)
Laurent@2157:
andrej@2360:
Laurent@2162: def HexDecValue(context, *args):
Laurent@2162: return str(ExtractHexDecValue(args[0][0]))
Laurent@2162:
andrej@2360:
Laurent@2162: def EntryName(context, *args):
andrej@2381: return ExtractName(args[0], args[1][0] if len(args) > 1 else None)
Laurent@2162:
andrej@2370:
Laurent@2162: ENTRY_INFOS_KEYS = [
Laurent@2162: ("Index", lambda x: "#x%4.4X" % int(x), "#x0000"),
Laurent@2162: ("SubIndex", str, "0"),
Laurent@2162: ("Name", str, ""),
Laurent@2162: ("Type", str, ""),
Laurent@2162: ("BitSize", int, 0),
Laurent@2162: ("Access", str, ""),
Laurent@2162: ("PDOMapping", str, ""),
Laurent@2162: ("PDO index", str, ""),
Laurent@2162: ("PDO name", str, ""),
Laurent@2162: ("PDO type", str, "")]
Laurent@2162:
andrej@2360:
andrej@2397: class EntryListFactory(object):
Laurent@2157:
Laurent@2157: def __init__(self, entries):
Laurent@2157: self.Entries = entries
andrej@2355:
Laurent@2162: def AddEntry(self, context, *args):
Laurent@2162: index, subindex = map(lambda x: int(x[0]), args[:2])
Laurent@2157: new_entry_infos = {
Laurent@2162: key: translate(arg[0]) if len(arg) > 0 else default
Laurent@2162: for (key, translate, default), arg
Laurent@2162: in zip(ENTRY_INFOS_KEYS, args)}
andrej@2355:
Laurent@2157: if (index, subindex) != (0, 0):
Laurent@2157: entry_infos = self.Entries.get((index, subindex))
Laurent@2157: if entry_infos is not None:
Laurent@2162: for param in ["PDO index", "PDO name", "PDO type"]:
Laurent@2162: value = new_entry_infos.get(param)
Laurent@2162: if value is not None:
Laurent@2162: entry_infos[param] = value
Laurent@2157: else:
Laurent@2157: self.Entries[(index, subindex)] = new_entry_infos
Laurent@2157:
andrej@2370:
Laurent@2157: entries_list_xslt = etree.parse(
Laurent@2157: os.path.join(ScriptDirectory, "entries_list.xslt"))
Laurent@2157:
Laurent@2157: cls = EtherCATInfoParser.GetElementClass("DeviceType")
Laurent@2074: if cls:
andrej@2355:
Laurent@2157: profile_numbers_xpath = EtherCATInfo_XPath("Profile/ProfileNo")
andrej@2387:
laurent@2032: def GetProfileNumbers(self):
Laurent@2157: return [number.text for number in profile_numbers_xpath(self)]
laurent@2032: setattr(cls, "GetProfileNumbers", GetProfileNumbers)
andrej@2355:
Laurent@2079: def getCoE(self):
laurent@2032: mailbox = self.getMailbox()
Laurent@2079: if mailbox is not None:
Laurent@2079: return mailbox.getCoE()
Laurent@2079: return None
Laurent@2079: setattr(cls, "getCoE", getCoE)
laurent@2032:
Laurent@2098: def GetEntriesList(self, limits=None):
laurent@2029: entries = {}
andrej@2355:
Laurent@2162: factory = EntryListFactory(entries)
andrej@2355:
Laurent@2157: entries_list_xslt_tree = etree.XSLT(
andrej@2366: entries_list_xslt, extensions={
Laurent@2162: ("entries_list_ns", "AddEntry"): factory.AddEntry,
Laurent@2157: ("entries_list_ns", "HexDecValue"): HexDecValue,
Laurent@2157: ("entries_list_ns", "EntryName"): EntryName})
Laurent@2157: entries_list_xslt_tree(self, **dict(zip(
andrej@2355: ["min_index", "max_index"],
Laurent@2157: map(lambda x: etree.XSLT.strparam(str(x)),
Laurent@2157: limits if limits is not None else [0x0000, 0xFFFF])
Laurent@2157: )))
andrej@2355:
laurent@2029: return entries
laurent@2029: setattr(cls, "GetEntriesList", GetEntriesList)
Laurent@2157:
laurent@2029: def GetSyncManagers(self):
laurent@2029: sync_managers = []
laurent@2029: for sync_manager in self.getSm():
laurent@2029: sync_manager_infos = {}
laurent@2029: for name, value in [("Name", sync_manager.getcontent()),
laurent@2029: ("Start Address", sync_manager.getStartAddress()),
laurent@2029: ("Default Size", sync_manager.getDefaultSize()),
laurent@2029: ("Control Byte", sync_manager.getControlByte()),
laurent@2029: ("Enable", sync_manager.getEnable())]:
laurent@2029: if value is None:
andrej@2365: value = ""
laurent@2029: sync_manager_infos[name] = value
laurent@2029: sync_managers.append(sync_manager_infos)
laurent@2029: return sync_managers
laurent@2029: setattr(cls, "GetSyncManagers", GetSyncManagers)
laurent@2029:
andrej@2360:
laurent@2022: def GroupItemCompare(x, y):
laurent@2022: if x["type"] == y["type"]:
laurent@2022: if x["type"] == ETHERCAT_GROUP:
laurent@2031: return cmp(x["order"], y["order"])
laurent@2022: else:
laurent@2031: return cmp(x["name"], y["name"])
laurent@2022: elif x["type"] == ETHERCAT_GROUP:
laurent@2022: return -1
laurent@2022: return 1
laurent@2022:
andrej@2360:
laurent@2022: def SortGroupItems(group):
laurent@2022: for item in group["children"]:
laurent@2022: if item["type"] == ETHERCAT_GROUP:
laurent@2022: SortGroupItems(item)
laurent@2022: group["children"].sort(GroupItemCompare)
laurent@2022:
andrej@2360:
andrej@2397: class ModulesLibrary(object):
Laurent@2137:
Laurent@2137: MODULES_EXTRA_PARAMS = [
andrej@2384: (
andrej@2384: "pdo_alignment",
andrej@2384: {
andrej@2384: "column_label": _("PDO alignment"),
andrej@2384: "column_size": 150,
andrej@2384: "default": 8,
andrej@2384: "description": _("Minimal size in bits between 2 pdo entries")
andrej@2384: }
andrej@2384: ),
andrej@2384: (
andrej@2384: "max_pdo_size",
andrej@2384: {
andrej@2384: "column_label": _("Max entries by PDO"),
andrej@2384: "column_size": 150,
andrej@2384: "default": 255,
andrej@2384: "description": _("""Maximal number of entries mapped in a PDO
andrej@2384: including empty entries used for PDO alignment""")
andrej@2384: }
andrej@2384: ),
andrej@2384: (
andrej@2384: "add_pdo",
andrej@2384: {
andrej@2384: "column_label": _("Creating new PDO"),
andrej@2384: "column_size": 150,
andrej@2384: "default": 0,
andrej@2384: "description": _("""Adding a PDO not defined in default configuration
Laurent@2138: for mapping needed location variables
andrej@2384: (1 if possible)""")
andrej@2384: }
andrej@2384: )
Laurent@2137: ]
andrej@2355:
Laurent@2097: def __init__(self, path, parent_library=None):
Laurent@2097: self.Path = path
Laurent@2097: if not os.path.exists(self.Path):
Laurent@2097: os.makedirs(self.Path)
Laurent@2097: self.ParentLibrary = parent_library
andrej@2355:
Laurent@2098: if parent_library is not None:
Laurent@2098: self.LoadModules()
Laurent@2098: else:
Laurent@2098: self.Library = None
Laurent@2137: self.LoadModulesExtraParams()
andrej@2355:
Laurent@2097: def GetPath(self):
Laurent@2097: return self.Path
andrej@2355:
Laurent@2137: def GetModulesExtraParamsFilePath(self):
Laurent@2137: return os.path.join(self.Path, "modules_extra_params.cfg")
andrej@2355:
Laurent@2157: groups_xpath = EtherCATInfo_XPath("Descriptions/Groups/Group")
Laurent@2157: devices_xpath = EtherCATInfo_XPath("Descriptions/Devices/Device")
andrej@2387:
Laurent@2097: def LoadModules(self):
Laurent@2097: self.Library = {}
andrej@2355:
Laurent@2097: files = os.listdir(self.Path)
laurent@2022: for file in files:
Laurent@2097: filepath = os.path.join(self.Path, file)
laurent@2022: if os.path.isfile(filepath) and os.path.splitext(filepath)[-1] == ".xml":
Laurent@2157: self.modules_infos = None
andrej@2355:
laurent@2022: xmlfile = open(filepath, 'r')
Laurent@2157: try:
Laurent@2160: self.modules_infos, error = EtherCATInfoParser.LoadXMLString(xmlfile.read())
Laurent@2160: if error is not None:
Laurent@2160: self.GetCTRoot().logger.write_warning(
Laurent@2160: XSDSchemaErrorMessage % (filepath + error))
andrej@2418: except Exception as exc:
andrej@2434: self.modules_infos, error = None, text(exc)
laurent@2022: xmlfile.close()
andrej@2355:
Edouard@2152: if self.modules_infos is not None:
Edouard@2152: vendor = self.modules_infos.getVendor()
andrej@2355:
Laurent@2157: vendor_category = self.Library.setdefault(
andrej@2355: ExtractHexDecValue(vendor.getId()),
andrej@2355: {"name": ExtractName(vendor.getName(), _("Miscellaneous")),
Laurent@2157: "groups": {}})
andrej@2355:
Laurent@2157: for group in self.groups_xpath(self.modules_infos):
laurent@2022: group_type = group.getType()
andrej@2355:
andrej@2381: vendor_category["groups"].setdefault(
andrej@2381: group_type,
andrej@2381: {
andrej@2381: "name": ExtractName(group.getName(), group_type),
andrej@2381: "parent": group.getParentGroup(),
andrej@2381: "order": group.getSortOrder(),
andrej@2381: # "value": group.getcontent()["value"],
andrej@2381: "devices": [],
andrej@2381: })
andrej@2359:
Laurent@2157: for device in self.devices_xpath(self.modules_infos):
laurent@2022: device_group = device.getGroupType()
andrej@2377: if device_group not in vendor_category["groups"]:
andrej@2412: raise ValueError("Not such group \"%s\"" % device_group)
Laurent@2157: vendor_category["groups"][device_group]["devices"].append(
Laurent@2157: (device.getType().getcontent(), device))
andrej@2355:
Laurent@2160: else:
andrej@2355:
Laurent@2160: self.GetCTRoot().logger.write_error(
andrej@2425: _("Couldn't load {a1} XML file:\n{a2}").format(a1=filepath, a2=error))
andrej@2355:
Laurent@2160: return self.Library
Edouard@2152:
laurent@2041: def GetModulesLibrary(self, profile_filter=None):
Laurent@2098: if self.Library is None:
Laurent@2098: self.LoadModules()
laurent@2022: library = []
Laurent@2097: for vendor_id, vendor in self.Library.iteritems():
laurent@2022: groups = []
Laurent@2073: children_dict = {}
laurent@2022: for group_type, group in vendor["groups"].iteritems():
laurent@2022: group_infos = {"name": group["name"],
laurent@2022: "order": group["order"],
laurent@2022: "type": ETHERCAT_GROUP,
laurent@2041: "infos": None,
laurent@2022: "children": children_dict.setdefault(group_type, [])}
laurent@2022: device_dict = {}
laurent@2022: for device_type, device in group["devices"]:
laurent@2041: if profile_filter is None or profile_filter in device.GetProfileNumbers():
Laurent@2097: product_code = device.getType().getProductCode()
Laurent@2097: revision_number = device.getType().getRevisionNo()
Laurent@2137: module_infos = {"device_type": device_type,
Laurent@2137: "vendor": vendor_id,
Laurent@2137: "product_code": product_code,
Laurent@2137: "revision_number": revision_number}
Laurent@2137: module_infos.update(self.GetModuleExtraParams(vendor_id, product_code, revision_number))
laurent@2041: device_infos = {"name": ExtractName(device.getName()),
laurent@2041: "type": ETHERCAT_DEVICE,
Laurent@2137: "infos": module_infos,
laurent@2041: "children": []}
laurent@2041: group_infos["children"].append(device_infos)
laurent@2041: device_type_occurrences = device_dict.setdefault(device_type, [])
laurent@2041: device_type_occurrences.append(device_infos)
laurent@2022: for device_type_occurrences in device_dict.itervalues():
laurent@2022: if len(device_type_occurrences) > 1:
laurent@2022: for occurrence in device_type_occurrences:
laurent@2022: occurrence["name"] += _(" (rev. %s)") % occurrence["infos"]["revision_number"]
laurent@2041: if len(group_infos["children"]) > 0:
laurent@2041: if group["parent"] is not None:
laurent@2041: parent_children = children_dict.setdefault(group["parent"], [])
laurent@2041: parent_children.append(group_infos)
laurent@2041: else:
laurent@2041: groups.append(group_infos)
laurent@2041: if len(groups) > 0:
laurent@2041: library.append({"name": vendor["name"],
laurent@2041: "type": ETHERCAT_VENDOR,
laurent@2041: "infos": None,
laurent@2041: "children": groups})
laurent@2031: library.sort(lambda x, y: cmp(x["name"], y["name"]))
laurent@2022: return library
Laurent@2097:
Laurent@2098: def GetVendors(self):
Laurent@2098: return [(vendor_id, vendor["name"]) for vendor_id, vendor in self.Library.items()]
andrej@2355:
Laurent@2097: def GetModuleInfos(self, module_infos):
Laurent@2097: vendor = ExtractHexDecValue(module_infos["vendor"])
Laurent@2097: vendor_infos = self.Library.get(vendor)
Laurent@2097: if vendor_infos is not None:
andrej@2406: for _group_name, group_infos in vendor_infos["groups"].iteritems():
Laurent@2097: for device_type, device_infos in group_infos["devices"]:
Laurent@2097: product_code = ExtractHexDecValue(device_infos.getType().getProductCode())
Laurent@2097: revision_number = ExtractHexDecValue(device_infos.getType().getRevisionNo())
andrej@2379: if product_code == ExtractHexDecValue(module_infos["product_code"]) and \
andrej@2379: revision_number == ExtractHexDecValue(module_infos["revision_number"]):
andrej@2355: self.cntdevice = device_infos
andrej@2355: self.cntdeviceType = device_type
Laurent@2137: return device_infos, self.GetModuleExtraParams(vendor, product_code, revision_number)
Laurent@2097: return None, None
andrej@2355:
Laurent@2097: def ImportModuleLibrary(self, filepath):
Laurent@2097: if os.path.isfile(filepath):
Laurent@2097: shutil.copy(filepath, self.Path)
Laurent@2097: self.LoadModules()
Laurent@2097: return True
Laurent@2097: return False
andrej@2355:
Laurent@2137: def LoadModulesExtraParams(self):
Laurent@2137: self.ModulesExtraParams = {}
andrej@2355:
Laurent@2137: csvfile_path = self.GetModulesExtraParamsFilePath()
Laurent@2097: if os.path.exists(csvfile_path):
Laurent@2097: csvfile = open(csvfile_path, "rb")
Laurent@2097: sample = csvfile.read(1024)
Laurent@2097: csvfile.seek(0)
Laurent@2097: dialect = csv.Sniffer().sniff(sample)
Laurent@2097: has_header = csv.Sniffer().has_header(sample)
Laurent@2097: reader = csv.reader(csvfile, dialect)
Laurent@2097: for row in reader:
Laurent@2097: if has_header:
Laurent@2097: has_header = False
Laurent@2097: else:
Laurent@2138: params_values = {}
andrej@2406: for (param, _param_infos), value in zip(
andrej@2379: self.MODULES_EXTRA_PARAMS, row[3:]):
Laurent@2138: if value != "":
Laurent@2138: params_values[param] = int(value)
Laurent@2138: self.ModulesExtraParams[
Laurent@2138: tuple(map(int, row[:3]))] = params_values
Laurent@2097: csvfile.close()
andrej@2355:
Laurent@2137: def SaveModulesExtraParams(self):
Laurent@2137: csvfile = open(self.GetModulesExtraParamsFilePath(), "wb")
andrej@2406: extra_params = [param for param, _params_infos in self.MODULES_EXTRA_PARAMS]
Laurent@2097: writer = csv.writer(csvfile, delimiter=';')
Laurent@2137: writer.writerow(['Vendor', 'product_code', 'revision_number'] + extra_params)
Laurent@2137: for (vendor, product_code, revision_number), module_extra_params in self.ModulesExtraParams.iteritems():
andrej@2355: writer.writerow([vendor, product_code, revision_number] +
andrej@2355: [module_extra_params.get(param, '')
Laurent@2137: for param in extra_params])
Laurent@2097: csvfile.close()
andrej@2355:
Laurent@2137: def SetModuleExtraParam(self, vendor, product_code, revision_number, param, value):
Laurent@2097: vendor = ExtractHexDecValue(vendor)
Laurent@2097: product_code = ExtractHexDecValue(product_code)
Laurent@2097: revision_number = ExtractHexDecValue(revision_number)
andrej@2355:
Laurent@2138: module_infos = (vendor, product_code, revision_number)
Laurent@2138: self.ModulesExtraParams.setdefault(module_infos, {})
Laurent@2138: self.ModulesExtraParams[module_infos][param] = value
andrej@2355:
Laurent@2137: self.SaveModulesExtraParams()
andrej@2355:
Laurent@2137: def GetModuleExtraParams(self, vendor, product_code, revision_number):
Laurent@2097: vendor = ExtractHexDecValue(vendor)
Laurent@2097: product_code = ExtractHexDecValue(product_code)
Laurent@2097: revision_number = ExtractHexDecValue(revision_number)
andrej@2355:
Laurent@2097: if self.ParentLibrary is not None:
Laurent@2137: extra_params = self.ParentLibrary.GetModuleExtraParams(vendor, product_code, revision_number)
Laurent@2137: else:
Laurent@2137: extra_params = {}
andrej@2355:
Laurent@2138: extra_params.update(self.ModulesExtraParams.get((vendor, product_code, revision_number), {}))
andrej@2355:
Laurent@2137: for param, param_infos in self.MODULES_EXTRA_PARAMS:
Laurent@2137: extra_params.setdefault(param, param_infos["default"])
andrej@2355:
Laurent@2137: return extra_params
Laurent@2097:
andrej@2370:
Laurent@2097: USERDATA_DIR = wx.StandardPaths.Get().GetUserDataDir()
Laurent@2097: if wx.Platform != '__WXMSW__':
Laurent@2097: USERDATA_DIR += '_files'
Laurent@2097:
Laurent@2097: ModulesDatabase = ModulesLibrary(
Laurent@2097: os.path.join(USERDATA_DIR, "ethercat_modules"))
Laurent@2097:
andrej@2360:
andrej@2397: class RootClass(object):
andrej@2355:
andrej@2363: CTNChildrenTypes = [("EthercatNode", _EthercatCTN, "Ethercat Master")]
Laurent@2097: EditorType = LibraryEditor
andrej@2355:
Laurent@2097: def __init__(self):
Laurent@2097: self.ModulesLibrary = None
Laurent@2097: self.LoadModulesLibrary()
andrej@2355:
Laurent@2149: def GetIconName(self):
Laurent@2149: return "Ethercat"
andrej@2355:
Laurent@2133: def GetModulesLibraryPath(self, project_path=None):
Laurent@2133: if project_path is None:
Laurent@2133: project_path = self.CTNPath()
andrej@2355: return os.path.join(project_path, "modules")
andrej@2355:
Laurent@2133: def OnCTNSave(self, from_project_path=None):
Laurent@2133: if from_project_path is not None:
Laurent@2133: shutil.copytree(self.GetModulesLibraryPath(from_project_path),
Laurent@2133: self.GetModulesLibraryPath())
Laurent@2133: return True
andrej@2355:
Laurent@2097: def CTNGenerate_C(self, buildpath, locations):
andrej@2363: return [], "", False
andrej@2355:
Laurent@2097: def LoadModulesLibrary(self):
Laurent@2097: if self.ModulesLibrary is None:
Laurent@2097: self.ModulesLibrary = ModulesLibrary(self.GetModulesLibraryPath(), ModulesDatabase)
Laurent@2097: else:
Laurent@2097: self.ModulesLibrary.LoadModulesLibrary()
andrej@2355:
Laurent@2097: def GetModulesDatabaseInstance(self):
Laurent@2097: return ModulesDatabase
andrej@2355:
Laurent@2097: def GetModulesLibraryInstance(self):
Laurent@2097: return self.ModulesLibrary
andrej@2355:
Laurent@2097: def GetModulesLibrary(self, profile_filter=None):
Laurent@2097: return self.ModulesLibrary.GetModulesLibrary(profile_filter)
andrej@2355:
Laurent@2098: def GetVendors(self):
Laurent@2098: return self.ModulesLibrary.GetVendors()
andrej@2355:
Laurent@2097: def GetModuleInfos(self, module_infos):
Laurent@2097: return self.ModulesLibrary.GetModuleInfos(module_infos)