(args)
| 1253 | |
| 1254 | |
| 1255 | def control_loop(args): |
| 1256 | try: |
| 1257 | # Wait for crowd service. |
| 1258 | time.sleep(3) |
| 1259 | c = Context(args) |
| 1260 | print('Control loop running.') |
| 1261 | |
| 1262 | pid_integrals = defaultdict(float) |
| 1263 | pid_last_errors = defaultdict(float) |
| 1264 | pid_last_update_time = None |
| 1265 | |
| 1266 | while True: |
| 1267 | start = time.time() |
| 1268 | pid_last_update_time = do_control(c, pid_integrals, pid_last_errors, pid_last_update_time) |
| 1269 | time.sleep(max(0, 1 / CONTROL_MAX_RATE - (time.time() - start))) # 20 Hz |
| 1270 | # print('({}) Control rate: {} Hz'.format(os.getpid(), 1 / max(time.time() - start, 0.001))) |
| 1271 | except Pyro4.errors.ConnectionClosedError: |
| 1272 | pass |
| 1273 | |
| 1274 | |
| 1275 | def gamma_loop(args): |
nothing calls this directly
no test coverage detected