Skip to content

eds: fix LowLimit/HighLimit parsing for all signed integer types #658

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
friederschueler wants to merge 1 commit into
base: master
Choose a base branch
from
+237 −120
Open

eds: fix LowLimit/HighLimit parsing for all signed integer types #658

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 86 additions & 64 deletions canopen/objectdictionary/eds.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,9 @@

logger = logging.getLogger(__name__)


def import_eds(source, node_id):
eds = RawConfigParser(inline_comment_prefixes=(';',))
eds = RawConfigParser(inline_comment_prefixes=(";",))
eds.optionxform = str
opened_here = False
try:
Expand All @@ -42,25 +43,26 @@ def import_eds(source, node_id):

if eds.has_section("FileInfo"):
od.__edsFileInfo = {
opt: eds.get("FileInfo", opt)
for opt in eds.options("FileInfo")
opt: eds.get("FileInfo", opt) for opt in eds.options("FileInfo")
}

if eds.has_section("Comments"):
linecount = int(eds.get("Comments", "Lines"), 0)
od.comments = '\n'.join([
eds.get("Comments", f"Line{line}")
for line in range(1, linecount+1)
])
od.comments = "\n".join(
[eds.get("Comments", f"Line{line}") for line in range(1, linecount + 1)]
)

if not eds.has_section("DeviceInfo"):
logger.warn("eds file does not have a DeviceInfo section. This section is mandatory")
logger.warn(
"eds file does not have a DeviceInfo section. This section is mandatory"
)
else:
for rate in [10, 20, 50, 125, 250, 500, 800, 1000]:
baudPossible = int(
eds.get("DeviceInfo", f"BaudRate_{rate}", fallback='0'), 0)
eds.get("DeviceInfo", f"BaudRate_{rate}", fallback="0"), 0
)
if baudPossible != 0:
od.device_information.allowed_baudrates.add(rate*1000)
od.device_information.allowed_baudrates.add(rate * 1000)

for t, eprop, odprop in [
(str, "VendorName", "vendor_name"),
Expand All @@ -80,13 +82,13 @@ def import_eds(source, node_id):
]:
try:
if t in (int, bool):
setattr(od.device_information, odprop,
t(int(eds.get("DeviceInfo", eprop), 0))
)
setattr(
od.device_information,
odprop,
t(int(eds.get("DeviceInfo", eprop), 0)),
)
elif t is str:
setattr(od.device_information, odprop,
eds.get("DeviceInfo", eprop)
)
setattr(od.device_information, odprop, eds.get("DeviceInfo", eprop))
except NoOptionError:
pass

Expand Down Expand Up @@ -131,10 +133,11 @@ def import_eds(source, node_id):
if object_type in (objectcodes.VAR, objectcodes.DOMAIN):
var = build_variable(eds, section, node_id, index)
od.add_object(var)
elif object_type == objectcodes.ARRAY and eds.has_option(section, "CompactSubObj"):
elif object_type == objectcodes.ARRAY and eds.has_option(
section, "CompactSubObj"
):
arr = ODArray(name, index)
last_subindex = ODVariable(
"Number of entries", index, 0)
last_subindex = ODVariable("Number of entries", index, 0)
last_subindex.data_type = datatypes.UNSIGNED8
arr.add_member(last_subindex)
arr.add_member(build_variable(eds, section, node_id, index, 1))
Expand Down Expand Up @@ -178,7 +181,7 @@ def import_eds(source, node_id):


def import_from_node(node_id: int, network: canopen.network.Network):
""" Download the configuration from the remote node
"""Download the configuration from the remote node
:param int node_id: Identifier of the node
:param network: network object
"""
Expand All @@ -192,25 +195,23 @@ def import_from_node(node_id: int, network: canopen.network.Network):
with sdo_client.open(0x1021, 0, "rt") as eds_fp:
od = import_eds(eds_fp, node_id)
except Exception as e:
logger.error("No object dictionary could be loaded for node %d: %s",
node_id, e)
logger.error("No object dictionary could be loaded for node %d: %s", node_id, e)
od = None
finally:
network.unsubscribe(0x580 + node_id)
return od


def _calc_bit_length(data_type):
if data_type == datatypes.INTEGER8:
return 8
elif data_type == datatypes.INTEGER16:
return 16
elif data_type == datatypes.INTEGER32:
return 32
elif data_type == datatypes.INTEGER64:
return 64
else:
raise ValueError(f"Invalid data_type '{data_type}', expecting a signed integer data_type.")
_SIGNED_BIT_LENGTHS = {
datatypes.INTEGER8: 8,
datatypes.INTEGER16: 16,
datatypes.INTEGER24: 24,
datatypes.INTEGER32: 32,
datatypes.INTEGER40: 40,
datatypes.INTEGER48: 48,
datatypes.INTEGER56: 56,
datatypes.INTEGER64: 64,
}


def _signed_int_from_hex(hex_str, bit_length):
Expand All @@ -233,8 +234,8 @@ def _convert_variable(node_id, var_type, value):
else:
# COB-ID can contain '$NODEID+' so replace this with node_id before converting
value = value.replace(" ", "").upper()
if '$NODEID' in value and node_id is not None:
return int(re.sub(r'\+?\$NODEID\+?', '', value), 0) + node_id
if "$NODEID" in value and node_id is not None:
return int(re.sub(r"\+?\$NODEID\+?", "", value), 0) + node_id
else:
return int(value, 0)

Expand Down Expand Up @@ -276,7 +277,9 @@ def build_variable(eds, section, node_id, index, subindex=0):
try:
var.data_type = int(eds.get(f"{var.data_type:X}sub1", "DefaultValue"), 0)
except NoSectionError:
logger.warning("%s has an unknown or unsupported data type (0x%X)", name, var.data_type)
logger.warning(
"%s has an unknown or unsupported data type (0x%X)", name, var.data_type
)
# Assume DOMAIN to force application to interpret the byte data
var.data_type = datatypes.DOMAIN

Expand All @@ -286,32 +289,40 @@ def build_variable(eds, section, node_id, index, subindex=0):
try:
min_string = eds.get(section, "LowLimit")
if var.data_type in datatypes.SIGNED_TYPES:
var.min = _signed_int_from_hex(min_string, _calc_bit_length(var.data_type))
var.min = _signed_int_from_hex(
min_string, _SIGNED_BIT_LENGTHS[var.data_type]
)
else:
var.min = int(min_string, 0)
except ValueError:
pass
logger.warning("Failed to parse LowLimit for %s: %r", name, min_string)
if eds.has_option(section, "HighLimit"):
try:
max_string = eds.get(section, "HighLimit")
if var.data_type in datatypes.SIGNED_TYPES:
var.max = _signed_int_from_hex(max_string, _calc_bit_length(var.data_type))
var.max = _signed_int_from_hex(
max_string, _SIGNED_BIT_LENGTHS[var.data_type]
)
else:
var.max = int(max_string, 0)
except ValueError:
pass
logger.warning("Failed to parse HighLimit for %s: %r", name, max_string)
if eds.has_option(section, "DefaultValue"):
try:
var.default_raw = eds.get(section, "DefaultValue")
if '$NODEID' in var.default_raw:
if "$NODEID" in var.default_raw:
var.relative = True
var.default = _convert_variable(node_id, var.data_type, eds.get(section, "DefaultValue"))
var.default = _convert_variable(
node_id, var.data_type, eds.get(section, "DefaultValue")
)
except ValueError:
pass
if eds.has_option(section, "ParameterValue"):
try:
var.value_raw = eds.get(section, "ParameterValue")
var.value = _convert_variable(node_id, var.data_type, eds.get(section, "ParameterValue"))
var.value = _convert_variable(
node_id, var.data_type, eds.get(section, "ParameterValue")
)
except ValueError:
pass
# Factor, Description and Unit are not standard according to the CANopen specifications, but they are implemented in the python canopen package, so we can at least try to use them
Expand Down Expand Up @@ -376,32 +387,36 @@ def export_variable(var, eds):
if var.access_type:
eds.set(section, "AccessType", var.access_type)

if getattr(var, 'default_raw', None) is not None:
if getattr(var, "default_raw", None) is not None:
eds.set(section, "DefaultValue", var.default_raw)
elif getattr(var, 'default', None) is not None:
eds.set(section, "DefaultValue", _revert_variable(
var.data_type, var.default))
elif getattr(var, "default", None) is not None:
eds.set(
section, "DefaultValue", _revert_variable(var.data_type, var.default)
)

if device_commisioning:
if getattr(var, 'value_raw', None) is not None:
if getattr(var, "value_raw", None) is not None:
eds.set(section, "ParameterValue", var.value_raw)
elif getattr(var, 'value', None) is not None:
eds.set(section, "ParameterValue",
_revert_variable(var.data_type, var.value))
elif getattr(var, "value", None) is not None:
eds.set(
section,
"ParameterValue",
_revert_variable(var.data_type, var.value),
)

eds.set(section, "DataType", f"0x{var.data_type:04X}")
eds.set(section, "PDOMapping", hex(var.pdo_mappable))

if getattr(var, 'min', None) is not None:
if getattr(var, "min", None) is not None:
eds.set(section, "LowLimit", var.min)
if getattr(var, 'max', None) is not None:
if getattr(var, "max", None) is not None:
eds.set(section, "HighLimit", var.max)

if getattr(var, 'description', '') != '':
if getattr(var, "description", "") != "":
eds.set(section, "Description", var.description)
if getattr(var, 'factor', 1) != 1:
if getattr(var, "factor", 1) != 1:
eds.set(section, "Factor", var.factor)
if getattr(var, 'unit', '') != '':
if getattr(var, "unit", "") != "":
eds.set(section, "Unit", var.unit)

def export_record(var, eds):
Expand All @@ -420,6 +435,7 @@ def export_record(var, eds):
eds.optionxform = str

from datetime import datetime as dt

defmtime = dt.utcnow()

try:
Expand Down Expand Up @@ -469,10 +485,13 @@ def export_record(var, eds):

# we are also adding out of spec baudrates here.
for rate in od.device_information.allowed_baudrates.union(
{10e3, 20e3, 50e3, 125e3, 250e3, 500e3, 800e3, 1000e3}):
{10e3, 20e3, 50e3, 125e3, 250e3, 500e3, 800e3, 1000e3}
):
eds.set(
"DeviceInfo", f"BaudRate_{int(rate//1000)}",
int(rate in od.device_information.allowed_baudrates))
"DeviceInfo",
f"BaudRate_{int(rate//1000)}",
int(rate in od.device_information.allowed_baudrates),
)

if device_commisioning and (od.bitrate or od.node_id):
eds.add_section("DeviceComissioning")
Expand Down Expand Up @@ -500,11 +519,13 @@ def manufacturer_indices(x):
return 0x2000 <= x < 0x6000

def optional_indices(x):
return all((
x > 0x1001,
not mandatory_indices(x),
not manufacturer_indices(x),
))
return all(
(
x > 0x1001,
not mandatory_indices(x),
not manufacturer_indices(x),
)
)

supported_mantatory_indices = list(filter(mandatory_indices, od))
supported_optional_indices = list(filter(optional_indices, od))
Expand All @@ -524,6 +545,7 @@ def add_list(section, list):

if not dest:
import sys

dest = sys.stdout

eds.write(dest, False)
36 changes: 36 additions & 0 deletions test/sample.eds
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1017,3 +1017,39 @@ PDOMapping=0x0
Factor=ERROR
Description=
Unit=

[3031]
ParameterName=INTEGER24 value range -1 to 0
ObjectType=0x7
DataType=0x10
AccessType=rw
HighLimit=0x000000
LowLimit=0xFFFFFF
PDOMapping=0

[3032]
ParameterName=INTEGER40 value range -1 to 0
ObjectType=0x7
DataType=0x12
AccessType=rw
HighLimit=0x0000000000
LowLimit=0xFFFFFFFFFF
PDOMapping=0

[3033]
ParameterName=INTEGER48 value range -1 to 0
ObjectType=0x7
DataType=0x13
AccessType=rw
HighLimit=0x000000000000
LowLimit=0xFFFFFFFFFFFF
PDOMapping=0

[3034]
ParameterName=INTEGER56 value range -1 to 0
ObjectType=0x7
DataType=0x14
AccessType=rw
HighLimit=0x00000000000000
LowLimit=0xFFFFFFFFFFFFFF
PDOMapping=0
Loading
Loading