docs(example): Signal XY example

Author: epage

nixnet_examples/can_signal_xy_io.py

@@ -0,0 +1,112 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import datetime
+import pprint
+import six
+import sys
+import time
+
+import nixnet
+from nixnet import constants
+
+
+pp = pprint.PrettyPrinter(indent=4)
+
+
+def convert_timestamp(timestamp):
+    system_epoch = time.gmtime(0)
+    system_epock_datetime = datetime.datetime(system_epoch.tm_year, system_epoch.tm_mon, system_epoch.tm_mday)
+    xnet_epoch_datetime = datetime.datetime(1601, 1, 1)
+    delta = system_epock_datetime - xnet_epoch_datetime
+    date = datetime.datetime.fromtimestamp(timestamp * 100e-9) - delta
+    return date
+
+
+def convert_datetime(date):
+    system_epoch = time.gmtime(0)
+    system_epock_datetime = datetime.datetime(system_epoch.tm_year, system_epoch.tm_mon, system_epoch.tm_mday)
+    xnet_epoch_datetime = datetime.datetime(1601, 1, 1)
+    delta = system_epock_datetime - xnet_epoch_datetime
+    timestamp = (date + delta - datetime.datetime(1970, 1, 1)) / datetime.timedelta(seconds=1) * 100e9
+    return int(timestamp)
+
+
+def main():
+    database_name = 'NIXNET_example'
+    cluster_name = 'CAN_Cluster'
+    input_signals = ['CANEventSignal1', 'CANEventSignal2']
+    output_signals = ['CANEventSignal1', 'CANEventSignal2']
+    interface1 = 'CAN1'
+    interface2 = 'CAN2'
+
+    with nixnet.SignalInXYSession(
+            interface1,
+            database_name,
+            cluster_name,
+            input_signals) as input_session:
+        with nixnet.SignalOutXYSession(
+                interface2,
+                database_name,
+                cluster_name,
+                output_signals) as output_session:
+            terminated_cable = six.moves.input('Are you using a terminated cable (Y or N)? ')
+            if terminated_cable.lower() == "y":
+                input_session.intf.can_term = constants.CanTerm.ON
+                output_session.intf.can_term = constants.CanTerm.OFF
+            elif terminated_cable.lower() == "n":
+                input_session.intf.can_term = constants.CanTerm.ON
+                output_session.intf.can_term = constants.CanTerm.ON
+            else:
+                print("Unrecognised input ({}), assuming 'n'".format(terminated_cable))
+                input_session.intf.can_term = constants.CanTerm.ON
+                output_session.intf.can_term = constants.CanTerm.ON
+
+            # Start the input session manually to make sure that the first
+            # signal value sent before the initial read will be received.
+            input_session.start()
+
+            out_waveforms = []
+            for out_signal in output_signals:
+                user_value = six.moves.input('Enter "{}" signal values [float, ...]: '.format(out_signal))
+                try:
+                    out_waveforms.append([float(x.strip()) for x in user_value.split(",")])
+                except ValueError:
+                    out_waveforms.append([24.5343, 77.0129])
+                    print('Unrecognized input ({}). Setting waveform to {}'.format(user_value, out_waveforms[-1]))
+
+            print('The same values should be received. Press q to quit')
+            i = 0
+            while True:
+                for waveform_index, waveform in enumerate(out_waveforms):
+                    for value_index, value in enumerate(waveform):
+                        out_waveforms[waveform_index][value_index] = value + i
+                output_session.signals.write(out_waveforms)
+                print('Sent signal values: {}'.format(pp.pformat(out_waveforms)))
+
+                # Wait 5 s and then read the received values.
+                # They should be the same as the ones sent.
+                time_limit = convert_datetime(datetime.datetime.now() + datetime.timedelta(seconds=5))
+
+                signals = input_session.signals.read(len(out_waveforms[0]), time_limit)
+                print('Received signals:')
+                for signal_values in signals:
+                    signal_values = [(value, convert_timestamp(timestamp)) for (value, timestamp) in signal_values]
+                    print('    {}', signal_values)
+
+                i += 1
+                for waveform in out_waveforms:
+                    if max(waveform) + i > sys.float_info.max:
+                        i = 0
+                        break
+
+                inp = six.moves.input('Hit enter to continue (q to quit): ')
+                if inp.lower() == 'q':
+                    break
+
+            print('Data acquisition stopped.')
+
+
+if __name__ == '__main__':
+    main()

tests/test_examples.py

@@ -13,6 +13,7 @@
 from nixnet_examples import can_frame_stream_io
 from nixnet_examples import can_signal_conversion
 from nixnet_examples import can_signal_single_point_io
+from nixnet_examples import can_signal_xy_io
 
 
 MockXnetLibrary = mock.create_autospec(_cfuncs.XnetLibrary, spec_set=True, instance=True)
@@ -24,6 +25,8 @@
 MockXnetLibrary.nx_read_frame.return_value = _ctypedefs.u32(0)
 MockXnetLibrary.nx_write_signal_single_point.return_value = _ctypedefs.u32(0)
 MockXnetLibrary.nx_read_signal_single_point.return_value = _ctypedefs.u32(0)
+MockXnetLibrary.nx_write_signal_xy.return_value = _ctypedefs.u32(0)
+MockXnetLibrary.nx_read_signal_xy.return_value = _ctypedefs.u32(0)
 MockXnetLibrary.nx_convert_frames_to_signals_single_point.return_value = _ctypedefs.u32(0)
 MockXnetLibrary.nx_convert_signals_to_frames_single_point.return_value = _ctypedefs.u32(0)
 MockXnetLibrary.nx_stop.return_value = _ctypedefs.u32(0)
@@ -84,6 +87,21 @@ def test_can_signal_single_point_empty_session(input_values):
         can_signal_single_point_io.main()
 
 
+@pytest.mark.parametrize("input_values", [
+    ['y', '1, 2', '3, 4', 'q'],
+    ['n', '1, 2', '3, 4', 'q'],
+    ['invalid', '1, 2', '3, 4', 'q'],
+    ['y', '1', '3, 4', 'q'],
+    ['y', 'invalid', '3, 4', 'q'],
+    ['y', 'invalid', '3, 4'] + 0x100 * [''] + ['q'],
+])
+@mock.patch('nixnet._cfuncs.lib', MockXnetLibrary)
+@mock.patch('time.sleep', lambda time: None)
+def test_can_signal_xy_empty_session(input_values):
+    with mock.patch('six.moves.input', six_input(input_values)):
+        can_signal_xy_io.main()
+
+
 @pytest.mark.parametrize("input_values", [
     ['1, 2'],
     ['1'],