Run the main simulation loop. Args: update_rate: Simulation update frequency in Hz print_status: Whether to print status to console
(self, update_rate: float = 100.0, print_status: bool = True)
| 1005 | # ========================================================================= |
| 1006 | |
| 1007 | def run(self, update_rate: float = 100.0, print_status: bool = True): |
| 1008 | """ |
| 1009 | Run the main simulation loop. |
| 1010 | |
| 1011 | Args: |
| 1012 | update_rate: Simulation update frequency in Hz |
| 1013 | print_status: Whether to print status to console |
| 1014 | """ |
| 1015 | if not self.can_sender: |
| 1016 | raise RuntimeError( |
| 1017 | "CAN sender not configured. Call set_can_sender() first." |
| 1018 | ) |
| 1019 | |
| 1020 | self.running = True |
| 1021 | dt = 1.0 / update_rate |
| 1022 | |
| 1023 | # Message transmission rates (in Hz) |
| 1024 | rates = { |
| 1025 | "engine": 100, |
| 1026 | "transmission": 50, |
| 1027 | "bms": 10, |
| 1028 | "chassis": 100, |
| 1029 | "body": 10, |
| 1030 | "cluster": 5, |
| 1031 | } |
| 1032 | |
| 1033 | # Counters for rate control |
| 1034 | counters = {key: 0 for key in rates} |
| 1035 | |
| 1036 | try: |
| 1037 | last_time = time.time() |
| 1038 | frame_count = 0 |
| 1039 | |
| 1040 | while self.running: |
| 1041 | current_time = time.time() |
| 1042 | elapsed = current_time - last_time |
| 1043 | |
| 1044 | if elapsed >= dt: |
| 1045 | last_time = current_time |
| 1046 | frame_count += 1 |
| 1047 | |
| 1048 | # Update simulation state |
| 1049 | self.update_driving_simulation(dt) |
| 1050 | |
| 1051 | # Transmit messages based on their rates |
| 1052 | for msg_type, rate in rates.items(): |
| 1053 | counters[msg_type] += 1 |
| 1054 | interval = int(update_rate / rate) |
| 1055 | |
| 1056 | if counters[msg_type] >= interval: |
| 1057 | counters[msg_type] = 0 |
| 1058 | |
| 1059 | if msg_type == "engine": |
| 1060 | self.transmit_engine_messages() |
| 1061 | elif msg_type == "transmission": |
| 1062 | self.transmit_transmission_messages() |
| 1063 | elif msg_type == "bms": |
| 1064 | self.transmit_bms_messages() |
no test coverage detected