|
| 1 | +import struct |
1 | 2 | import time |
2 | 3 | import unittest |
3 | 4 | from datetime import datetime |
|
9 | 10 |
|
10 | 11 | class TestTime(unittest.TestCase): |
11 | 12 |
|
| 13 | + def test_epoch(self): |
| 14 | + """Verify that the epoch matches the standard definition.""" |
| 15 | + epoch = datetime.strptime( |
| 16 | + "1984-01-01 00:00:00 +0000", "%Y-%m-%d %H:%M:%S %z" |
| 17 | + ).timestamp() |
| 18 | + self.assertEqual(int(epoch), canopen.timestamp.OFFSET) |
| 19 | + |
12 | 20 | def test_time_producer(self): |
13 | 21 | network = canopen.Network() |
14 | 22 | network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 |
15 | 23 | network.connect(interface="virtual", receive_own_messages=True) |
16 | 24 | producer = canopen.timestamp.TimeProducer(network) |
17 | 25 |
|
18 | | - # Test that the epoch is correct |
19 | | - epoch = datetime.strptime( |
20 | | - "1984-01-01 00:00:00 +0000", "%Y-%m-%d %H:%M:%S %z" |
21 | | - ).timestamp() |
22 | | - self.assertEqual(int(epoch), canopen.timestamp.OFFSET) |
| 26 | + # Provide a specific time to verify the proper encoding |
| 27 | + producer.transmit(1_927_999_438) # 2031-02-04T19:23:58+00:00 |
| 28 | + msg = network.bus.recv(1) |
| 29 | + self.assertEqual(msg.arbitration_id, 0x100) |
| 30 | + self.assertEqual(msg.dlc, 6) |
| 31 | + self.assertEqual(msg.data, b"\xb0\xa4\x29\x04\x31\x43") |
23 | 32 |
|
| 33 | + # Test again with the current time as implicit timestamp |
24 | 34 | current = time.time() |
25 | 35 | with patch("canopen.timestamp.time.time", return_value=current): |
26 | | - current_in_epoch = current - epoch |
27 | | - |
28 | | - # Test it looking up the current time |
| 36 | + current_from_epoch = current - canopen.timestamp.OFFSET |
29 | 37 | producer.transmit() |
30 | 38 | msg = network.bus.recv(1) |
31 | 39 | self.assertEqual(msg.arbitration_id, 0x100) |
32 | 40 | self.assertEqual(msg.dlc, 6) |
33 | | - ms, days = canopen.timestamp.TIME_OF_DAY_STRUCT.unpack(msg.data) |
34 | | - self.assertEqual(days, int(current_in_epoch) // canopen.timestamp.ONE_DAY) |
35 | | - self.assertEqual(ms, int((current_in_epoch % canopen.timestamp.ONE_DAY) * 1000)) |
36 | | - |
37 | | - # Provide a specific time to verify the proper encoding |
38 | | - faketime = 1_927_999_438 # 2031-02-04 19:23:58 |
39 | | - producer.transmit(faketime) |
40 | | - msg = network.bus.recv(1) |
41 | | - self.assertEqual(msg.arbitration_id, 0x100) |
42 | | - self.assertEqual(msg.dlc, 6) |
43 | | - self.assertEqual(msg.data, b"\xb0\xa4\x29\x04\x31\x43") |
| 41 | + ms, days = struct.unpack("<LH", msg.data) |
| 42 | + self.assertEqual(days, int(current_from_epoch) // 86400) |
| 43 | + self.assertEqual(ms, int(current_from_epoch % 86400 * 1000)) |
44 | 44 |
|
45 | 45 | network.disconnect() |
46 | 46 |
|
|
0 commit comments