Thread-launched function for orientating the robot around. It gets commands from the keyboard as well as data from the sensor through the sensor_queue queue. :param trigger: CTRL-C event. When it's set, it means CTRL-C was pressed and the thread needs to stop. :param simultaneous_l
(trigger, simultaneous_launcher, motor_command_queue, sensor_queue)
| 192 | |
| 193 | |
| 194 | def robotControl(trigger, simultaneous_launcher, motor_command_queue, sensor_queue): |
| 195 | """ |
| 196 | Thread-launched function for orientating the robot around. It gets commands from the keyboard as well |
| 197 | as data from the sensor through the sensor_queue queue. |
| 198 | |
| 199 | :param trigger: CTRL-C event. When it's set, it means CTRL-C was pressed and the thread needs to stop. |
| 200 | :param simultaneous_launcher: It's a barrier used for synchronizing all threads together. |
| 201 | :param motor_command_queue: Queue containing commands from the keyboard. The commands are read from within main. |
| 202 | :param sensor_queue: Processed data off of the IMU. The queue is intended to be read. |
| 203 | :return: Nothing. |
| 204 | |
| 205 | """ |
| 206 | |
| 207 | time_to_wait_in_queue = 0.1 # measured in |
| 208 | |
| 209 | # try to connect to the GoPiGo3 |
| 210 | try: |
| 211 | gopigo3_robot = EasyGoPiGo3() |
| 212 | except IOError: |
| 213 | print("GoPiGo3 robot not detected") |
| 214 | simultaneous_launcher.abort() |
| 215 | except gopigo3.FirmwareVersionError: |
| 216 | print("GoPiGo3 board needs to be updated") |
| 217 | simultaneous_launcher.abort() |
| 218 | except Exception: |
| 219 | print("Unknown error occurred while instantiating GoPiGo3") |
| 220 | simultaneous_launcher.abort() |
| 221 | |
| 222 | # synchronizing point between all threads |
| 223 | # if abort method was called, then the synch will fail |
| 224 | try: |
| 225 | simultaneous_launcher.wait() |
| 226 | except threading.BrokenBarrierError as msg: |
| 227 | print("[robotControl] thread couldn't be launched") |
| 228 | |
| 229 | # if threads were successfully synchronized |
| 230 | # then set the GoPiGo3 appropriately |
| 231 | if not simultaneous_launcher.broken: |
| 232 | gopigo3_robot.stop() |
| 233 | gopigo3_robot.set_speed(MOTORS_SPEED) |
| 234 | |
| 235 | direction_degrees = None |
| 236 | move = False |
| 237 | acceptable_error_percent = 8 |
| 238 | command = "stop" |
| 239 | rotational_factor = 0.30 |
| 240 | accepted_minimum_by_drivers = 6 |
| 241 | |
| 242 | # while CTRL-C is not pressed, the synchronization between threads didn't fail and while the batteries' voltage isn't too low |
| 243 | while not (trigger.is_set() or simultaneous_launcher.broken or gopigo3_robot.volt() <= MINIMUM_VOLTAGE): |
| 244 | # read from the queue of the keyboard |
| 245 | try: |
| 246 | command = motor_command_queue.get(timeout = time_to_wait_in_queue) |
| 247 | motor_command_queue.task_done() |
| 248 | except queue.Empty: |
| 249 | pass |
| 250 | |
| 251 | # make some selection depending on what every command represents |
nothing calls this directly
no test coverage detected