(self, helper)
| 60 | # helper: 参考算法接入文档 |
| 61 | # 返回值: |
| 62 | def on_step(self, helper): |
| 63 | print("========== my module on_step called. =========") |
| 64 | t = helper.timestamp() |
| 65 | print("current simulation timestamp: {}ms".format(t)) |
| 66 | if self.pub: |
| 67 | msg = 'msg of py pub module on {} ms.'.format(t) |
| 68 | helper.publish_message(TEST_TCP_TOPIC, msg.encode('utf-8')) |
| 69 | buf = helper.get_published_shmem_buffer(TEST_SHMEM_TOPIC) |
| 70 | if buf: |
| 71 | print("got shared memory buffer with size of {} bytes.".format(len(buf))) |
| 72 | struct.pack_into('I{}s'.format(len(msg)), buf, 0, |
| 73 | len(msg), msg.encode('utf-8')) |
| 74 | else: |
| 75 | msg = helper.get_subscribed_message(TEST_TCP_TOPIC) |
| 76 | print('received {} msg: {}'.format( |
| 77 | TEST_TCP_TOPIC, msg.decode('utf-8'))) |
| 78 | buf = helper.get_subscribed_shmem_data(TEST_SHMEM_TOPIC) |
| 79 | if buf: |
| 80 | msg_size = struct.unpack_from('I', buf, 0)[0] |
| 81 | print("found {} bytes data in shared memory.".format(msg_size)) |
| 82 | shmem_data = struct.unpack_from( |
| 83 | '{}s'.format(msg_size), buf, 4)[0] |
| 84 | print("data in shared memory: {}".format(shmem_data)) |
| 85 | if self.max_step_time > 0 and t > self.max_step_time: |
| 86 | helper.stop_scenario("max step time reached!") |
| 87 | |
| 88 | # 备注: 该方法在结束仿真系统时候被回调 |
| 89 | # 参数: |
nothing calls this directly
no test coverage detected