Merge pull request #227 from d-bohls/signalDBCValueTable

feat(API): Add DBC signal value table

docs/api_reference/database.rst

@@ -10,3 +10,4 @@ nixnet.database
    database/frame
    database/signal
    database/dbc_attributes
+   database/dbc_signal_value_table

docs/api_reference/database/dbc_signal_value_table.rst

@@ -0,0 +1,7 @@
+nixnet.database.dbc_signal_value_table
+======================================
+
+.. automodule:: nixnet.database._dbc_signal_value_table
+    :members:
+    :inherited-members:
+    :show-inheritance:

nixnet/database/_dbc_signal_value_table.py

@@ -0,0 +1,102 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import collections
+import six
+import typing  # NOQA: F401
+
+from nixnet import _funcs
+from nixnet import constants
+
+
+class DbcSignalValueTable(collections.Mapping):
+    """Collection for accessing a DBC signal value table."""
+
+    def __init__(self, handle):
+        # type: (int) -> None
+        self._handle = handle
+
+    def __repr__(self):
+        return '{}(handle={})'.format(type(self).__name__, self._handle)
+
+    def __eq__(self, other):
+        if isinstance(other, self.__class__):
+            return self._handle == typing.cast(DbcSignalValueTable, other)._handle
+        else:
+            return NotImplemented
+
+    def __ne__(self, other):
+        result = self.__eq__(other)
+        if result is NotImplemented:
+            return result
+        else:
+            return not result
+
+    def __hash__(self):
+        return hash(self._handle)
+
+    def __len__(self):
+        return len(self._value_table)
+
+    def __iter__(self):
+        return self.keys()
+
+    def __getitem__(self, key):
+        # type: (typing.Text) -> typing.Text
+        """Return the value.
+
+            Args:
+                Value description.
+            Returns:
+                Value
+        """
+        if isinstance(key, six.string_types):
+            return self._value_table[key]
+        else:
+            raise TypeError(key)
+
+    def keys(self):
+        """Return all value descriptions in the collection.
+
+            Yields:
+                An iterator to all value descriptions in the collection.
+        """
+        return iter(self._value_table.keys())
+
+    def values(self):
+        """Return all values in the collection.
+
+            Yields:
+                An iterator to all values in the collection.
+        """
+        return iter(self._value_table.values())
+
+    def items(self):
+        """Return all value descriptions and values in the collection.
+
+            Yields:
+                An iterator to tuple pairs of value descriptions and values in the collection.
+        """
+        return iter(self._value_table.items())
+
+    @property
+    def _value_table(self):
+        # type: () -> typing.Dict[typing.Text, typing.Text]
+        mode = constants.GetDbcAttributeMode.VALUE_TABLE_LIST
+        attribute_size = _funcs.nxdb_get_dbc_attribute_size(self._handle, mode, '')
+        attribute_info = _funcs.nxdb_get_dbc_attribute(self._handle, mode, '', attribute_size)
+        table_string = attribute_info[0]
+        if not table_string:
+            return {}
+
+        table_list = table_string.split(',')
+        if len(table_list) % 2:
+            raise ValueError('Value tables require an even number of items: %s' % table_list)
+
+        # table_string is of the format 'value1, key1, value2, key2, ...'
+        # convert to a dict: { 'key1': int('value1'), 'key2': int('value2'), ... }
+        table_dict = dict(
+            (key, value)
+            for value, key in zip(table_list[0::2], table_list[1::2]))
+        return table_dict

nixnet/database/_signal.py

@@ -7,6 +7,7 @@
 from nixnet import _props
 from nixnet import constants
 from nixnet.database import _dbc_attributes
+from nixnet.database import _dbc_signal_value_table
 
 
 class Signal(object):
@@ -16,6 +17,7 @@ def __init__(self, handle):
         # type: (int) -> None
         self._handle = handle
         self._dbc_attributes = None  # type: typing.Optional[_dbc_attributes.DbcAttributeCollection]
+        self._dbc_signal_value_table = _dbc_signal_value_table.DbcSignalValueTable(self._handle)
 
     def __eq__(self, other):
         if isinstance(other, self.__class__):
@@ -72,6 +74,12 @@ def dbc_attributes(self):
             self._dbc_attributes = _dbc_attributes.DbcAttributeCollection(self._handle)
         return self._dbc_attributes
 
+    @property
+    def dbc_signal_value_table(self):
+        # type: () -> _dbc_signal_value_table.DbcSignalValueTable
+        """:any:`nixnet.database._dbc_signal_value_table.DbcSignalValueTable`: Access the signal's DBC value table."""
+        return self._dbc_signal_value_table
+
     @property
     def default(self):
         return _props.get_signal_default(self._handle)

tests/databases/attributes.dbc

@@ -83,4 +83,5 @@ BA_ "MsgAttr3" BO_ 1 "MsgAttr3String";
 BA_ "MsgAttr2" BO_ 1 -23.33;
 BA_ "MsgAttr1" BO_ 1 22;
 BA_ "SigAttr1" SG_ 0 Sig2 11;
+VAL_ 0 Sig1 4 "High" 0 "Zero" -10 "Low" ;
 

tests/test_dbc_attributes.py

@@ -12,9 +12,6 @@
 from nixnet.database import _dbc_attributes
 
 MockXnetLibrary = mock.create_autospec(_cfuncs.XnetLibrary, spec_set=True, instance=True)
-MockXnetLibrary.nxdb_open_database.return_value = _ctypedefs.u32(0)
-MockXnetLibrary.nxdb_close_database.return_value = _ctypedefs.u32(0)
-MockXnetLibrary.nxdb_find_object.return_value = _ctypedefs.u32(0)
 MockXnetLibrary.nxdb_get_dbc_attribute_size.return_value = _ctypedefs.u32(0)
 MockXnetLibrary.nxdb_get_dbc_attribute.return_value = _ctypedefs.u32(0)
 
@@ -60,7 +57,7 @@ def test_cluster_dbc_attributes():
         for key in cluster.dbc_attributes:
             print(cluster.dbc_attributes[key])
 
-        # test containers
+        # test container
         assert sorted(cluster.dbc_attributes.keys()) == ['BusType', 'NetworkAttr1']
         assert sorted(cluster.dbc_attributes.values()) == [('CAN', True), ('abc', True)]
         assert sorted(cluster.dbc_attributes.items()) == [('BusType', ('CAN', True)), ('NetworkAttr1', ('abc', True))]
@@ -87,7 +84,7 @@ def test_ecu_dbc_attributes():
         for key in ecu1.dbc_attributes:
             print(ecu1.dbc_attributes[key])
 
-        # test mapping overrides
+        # test container
         assert sorted(ecu1.dbc_attributes.keys()) == ['EcuAttr1']
         assert sorted(ecu1.dbc_attributes.items()) == [('EcuAttr1', ('xEcu1', True))]
         assert sorted(ecu1.dbc_attributes.values()) == [('xEcu1', True)]
@@ -114,7 +111,7 @@ def test_frame_dbc_attributes():
         for key in frame1.dbc_attributes:
             print(frame1.dbc_attributes[key])
 
-        # test mapping overrides
+        # test container
         assert sorted(frame1.dbc_attributes.keys()) == ['MsgAttr1', 'MsgAttr2', 'MsgAttr3', 'MsgAttr4']
         assert sorted(frame1.dbc_attributes.values()) == [('-11.1', True),
                                                           ('2', True),
@@ -154,7 +151,7 @@ def test_signal_dbc_attributes():
         for key in sig1.dbc_attributes:
             print(sig1.dbc_attributes[key])
 
-        # test mapping overrides
+        # test container
         assert sorted(sig2.dbc_attributes.keys()) == ['SigAttr1']
         assert sorted(sig2.dbc_attributes.values()) == [('11', False)]
         assert sorted(sig2.dbc_attributes.items()) == [('SigAttr1', ('11', False))]

tests/test_dbc_signal_value_table.py

@@ -0,0 +1,71 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import mock  # type: ignore
+import os
+import pytest  # type: ignore
+
+from nixnet import _cfuncs
+from nixnet import _ctypedefs
+from nixnet import database
+from nixnet.database import _dbc_signal_value_table
+
+MockXnetLibrary = mock.create_autospec(_cfuncs.XnetLibrary, spec_set=True, instance=True)
+MockXnetLibrary.nxdb_get_dbc_attribute_size.return_value = _ctypedefs.u32(0)
+MockXnetLibrary.nxdb_get_dbc_attribute.return_value = _ctypedefs.u32(0)
+
+
+@mock.patch('nixnet._cfuncs.lib', MockXnetLibrary)
+def test_dbc_signal_value_table():
+    dbc_sig_val_table120 = _dbc_signal_value_table.DbcSignalValueTable(120)
+    dbc_sig_val_table130_1 = _dbc_signal_value_table.DbcSignalValueTable(130)
+    dbc_sig_val_table130_2 = _dbc_signal_value_table.DbcSignalValueTable(130)
+
+    # test dunders
+    assert str(dbc_sig_val_table120) == 'DbcSignalValueTable(handle=120)'
+    assert dbc_sig_val_table130_1 == dbc_sig_val_table130_2
+    assert dbc_sig_val_table130_1 != dbc_sig_val_table120
+    assert len({dbc_sig_val_table120, dbc_sig_val_table130_1, dbc_sig_val_table130_2}) == 2  # testing `__hash__`
+    assert len(dbc_sig_val_table120) == 0
+    for key in dbc_sig_val_table120:
+        print(dbc_sig_val_table120[key])
+    with pytest.raises(KeyError):
+        print(dbc_sig_val_table120[''])
+    with pytest.raises(TypeError):
+        print(dbc_sig_val_table120[5])
+
+    # test container
+    assert sorted(dbc_sig_val_table120.keys()) == []
+    assert sorted(dbc_sig_val_table120.values()) == []
+    assert sorted(dbc_sig_val_table120.items()) == []
+
+
+@pytest.mark.integration
+def test_signal_dbc_signal_value_table():
+    database_filepath = os.path.join(os.path.dirname(__file__), 'databases\\attributes.dbc')
+    with database.Database(database_filepath) as db:
+        cluster = db.clusters['Cluster']
+        frame1 = cluster.frames['Msg1']
+        sig1 = frame1.mux_static_signals['Sig1']
+        sig2 = frame1.mux_static_signals['Sig2']
+
+        # test dunders
+        assert len(sig1.dbc_signal_value_table) == 3
+        assert len(sig2.dbc_signal_value_table) == 0
+        assert len({sig1.dbc_signal_value_table, sig1.dbc_signal_value_table}) == 1  # testing `__hash__`
+        assert len({sig1.dbc_signal_value_table, sig2.dbc_signal_value_table}) == 2  # testing `__hash__`
+        assert sig1.dbc_signal_value_table == sig1.dbc_signal_value_table
+        assert sig1.dbc_signal_value_table != sig2.dbc_signal_value_table
+        for key in sig1.dbc_signal_value_table:
+            print(sig1.dbc_signal_value_table[key])
+
+        # test container
+        assert sorted(sig1.dbc_signal_value_table.keys()) == ['High', 'Low', 'Zero']
+        assert sorted(sig1.dbc_signal_value_table.values()) == ['-10', '0', '4']
+        assert sorted(sig1.dbc_signal_value_table.items()) == [('High', '4'), ('Low', '-10'), ('Zero', '0')]
+
+        # test values
+        assert sig1.dbc_signal_value_table['Low'] == '-10'
+        assert sig1.dbc_signal_value_table['High'] == '4'
+        assert sig1.dbc_signal_value_table['Zero'] == '0'