import time import unittest import lcm from test.TestUtils import * lc = lcm.LCM() class LCMPublishDataTest(unittest.TestCase): def test_publish_lcm(self): num_msgs = 100 msg_pub_rate_sec = 1 pub = LcmPublisher(lc) data = read_file() data.pop(0) # drop first line channel_name = 'WetLabsUBAT' var_name = 'raw_data' print("\n") print("publishing {} msgs at {} Hz:".format(num_msgs, 1/msg_pub_rate_sec)) for l in data[:num_msgs]: try: print("publishing msg # {}".format(pub.seq_number)) pub.clear_msg() pub.add_variable(var_name, l) pub.timestamp() pub.publish(channel_name) time.sleep(msg_pub_rate_sec) except KeyboardInterrupt: break self.assertEqual(True, True) class LCMPubTest(unittest.TestCase): def test_get_type(self): pub = LcmPublisher(lc) val = [1, [1, 2]] for v in val: self.assertEqual(pub.get_type(v) is int, True) val = [1.1, [1.1, 2.2]] for v in val: self.assertEqual(pub.get_type(v) is float, True) val = [numpy.float32(1.1), [numpy.float32(1.1), numpy.float32(2.2)]] for v in val: self.assertEqual(pub.get_type(v) is numpy.float32, True) val = ['1.0', ['1.1', '2.2']] for v in val: self.assertEqual(pub.get_type(v) is str, True) val = [[1.1, '2.2'], [1.1, 1], ['1.1', 1]] for v in val: self.assertEqual(pub.get_type(v), False) def test_add_int(self): pub = LcmPublisher(lc) name = 'int_test' unit = 'n/a' print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) pub.add_int(name=name + str(i), unit=unit, val=i) self.assertEqual(len(pub.msg.intVector), i+1) # print("size of intVector: {}, i: {}".format(len(pub.msg.intVector), i+1)) pub.publish('add_int') def test_add_int_same_name(self): pub = LcmPublisher(lc) name = 'int_test' unit = 'n/a' print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) pub.add_int(name, unit=unit, val=i) self.assertEqual(len(pub.msg.intVector), 1) self.assertEqual(pub.msg.intVector[0].val[0], i) # print("size of intVector: {}, i: {}".format(len(pub.msg.intVector), i+1)) pub.publish('add_int_same_name') def test_add_int_array(self): pub = LcmPublisher(lc) name = 'int_test' unit = 'n/a' val = [] print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) val.append(i) pub.add_int(name=name, unit=unit, val=val) self.assertEqual(len(pub.msg.intVector[0].val), i + 1) # print("size of intVector val: {}, i: {}".format(len(pub.msg.intVector[0].val), i+1)) pub.publish('add_int_same_array') pub.clear_msg() self.assertEqual(len(pub.msg.intVector), 0) def test_add_float(self): pub = LcmPublisher(lc) name = 'float_test' unit = 'n/a' print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) pub.add_float(name + str(i), unit=unit, val=numpy.float32(i)) self.assertEqual(len(pub.msg.floatVector), i + 1) # print("size of floatVector: {}, i: {}".format(len(pub.msg.floatVector), i + 1)) pub.publish('add_float') def test_add_float_same_name(self): pub = LcmPublisher(lc) name = 'float_test' unit = 'n/a' print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) pub.add_float(name, unit=unit, val=numpy.float32(i)) self.assertEqual(len(pub.msg.floatVector), 1) # print("size of floatVector: {}, i: {}".format(len(pub.msg.floatVector), i + 1)) pub.publish('add_float_same_name') def test_add_float_array(self): pub = LcmPublisher(lc) name = 'float_test' unit = 'n/a' val = [] print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) val.append(numpy.float32(i)) pub.add_float(name, unit=unit, val=val) self.assertEqual(len(pub.msg.floatVector), 1) self.assertEqual(len(pub.msg.floatVector[0].val), i + 1) # print("size of floatVector val: {}, i: {}".format(len(pub.msg.floatVector[0].val), i + 1)) pub.publish('add_float_array') pub.clear_msg() self.assertEqual(len(pub.msg.floatVector), 0) def test_add_double(self): pub = LcmPublisher(lc) name = 'double_test' unit = 'n/a' print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) pub.add_double(name + str(i), unit=unit, val=float(i)) self.assertEqual(len(pub.msg.doubleVector), i + 1) # print("size of doubleVector: {}, i: {}".format(len(pub.msg.doubleVector), i + 1)) pub.publish('add_double') def test_add_double_same_name(self): pub = LcmPublisher(lc) name = 'double_test' unit = 'n/a' print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) pub.add_double(name, unit=unit, val=float(i)) self.assertEqual(len(pub.msg.doubleVector), 1) # print("size of doubleVector: {}, i: {}".format(len(pub.msg.doubleVector), i + 1)) pub.publish('add_double_same_name') def test_add_double_array(self): pub = LcmPublisher(lc) name = 'double_test' unit = 'n/a' val = [] print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) val.append(float(i)) pub.add_double(name, unit=unit, val=val) self.assertEqual(len(pub.msg.doubleVector), 1) self.assertEqual(len(pub.msg.doubleVector[0].val), i + 1) # print("size of doubleVector val: {}, i: {}".format(len(pub.msg.doubleVector[0].val), i + 1)) pub.publish('add_double_array') pub.clear_msg() self.assertEqual(len(pub.msg.doubleVector), 0) def test_add_string(self): pub = LcmPublisher(lc) name = 'string_test' unit = 'n/a' print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) pub.add_str(name=name+str(i), val='test' + str(i)) self.assertEqual(len(pub.msg.stringVector), i+1) # print("size of stringVector: {}, i: {}".format(len(pub.msg.stringVector), i+1)) pub.publish('add_string') def test_add_string_array(self): pub = LcmPublisher(lc) name = 'string_test' unit = 'n/a' val = [] print("") for i in range(10): pub.msg.epochMillisec = int(time.time() * 1000) val.append('test' + str(i)) pub.add_str(name=name, val=val) self.assertEqual(len(pub.msg.stringVector), 1) self.assertEqual(len(pub.msg.stringVector[0].val), i + 1) # print("size of stringVector val: {}, i: {}".format(len(pub.msg.stringVector[0].val), i+1)) pub.publish('add_string_array') pub.clear_msg() self.assertEqual(len(pub.msg.stringVector), 0) def test_add_variable(self): pub = LcmPublisher(lc) name = 'var_test' unit = 'n/a' val = [['foo', 'bar'], [1, 2], [1.1, 2.2]] pub.add_variable(name, val[0], unit) self.assertEqual(pub.msg.stringVector[0].nVal, 2) pub.clear_msg() pub.add_variable(name, val[1], unit) self.assertEqual(pub.msg.intVector[0].nVal, 2) pub.clear_msg() pub.add_variable(name, val[2], unit) self.assertEqual(pub.msg.doubleVector[0].nVal, 2) pub.clear_msg() print("\ntest mixed val: ", end="") pub.add_variable(name, [1.1, '2.2'], unit) self.assertEqual(len(pub.msg.doubleVector), 0) pub.clear_msg() if __name__ == '__main__': unittest.main()