Skip to content

Commit 331d09c

Browse files
andchiindandchiind
committed
Make monitor mission an asyncio task
1 parent 4200a93 commit 331d09c

3 files changed

Lines changed: 163 additions & 175 deletions

File tree

‎src/isar/robot/robot_monitor_mission.py‎

Lines changed: 103 additions & 119 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
1+
import asyncio
12
import logging
2-
import time
3-
from threading import Event
43
from typing import Callable, Iterator, List, Optional
54

65
from isar.config.settings import settings
@@ -49,9 +48,7 @@ def should_upload_inspections(task: TASKS) -> bool:
4948
return task.status == TaskStatus.Successful and isinstance(task, InspectionTask)
5049

5150

52-
def get_task_status(
53-
signal_exit: Event,
54-
signal_mission_stopped: Event,
51+
async def get_task_status(
5552
robot: RobotInterface,
5653
task_id: str,
5754
) -> TaskStatus:
@@ -62,10 +59,8 @@ def get_task_status(
6259
while (
6360
request_status_failure_counter < settings.REQUEST_STATUS_FAILURE_COUNTER_LIMIT
6461
):
65-
if signal_exit.wait(0) or signal_mission_stopped.wait(0):
66-
return TaskStatus.Cancelled
6762
if request_status_failure_counter > 0:
68-
time.sleep(settings.REQUEST_STATUS_COMMUNICATION_RECONNECT_DELAY)
63+
await asyncio.sleep(settings.REQUEST_STATUS_COMMUNICATION_RECONNECT_DELAY)
6964

7065
try:
7166
task_status = robot.task_status(task_id)
@@ -103,9 +98,7 @@ def get_task_status(
10398
)
10499

105100

106-
def get_mission_status(
107-
signal_exit: Event,
108-
signal_mission_stopped: Event,
101+
async def get_mission_status(
109102
robot: RobotInterface,
110103
mission_id: str,
111104
) -> MissionStatus:
@@ -116,10 +109,8 @@ def get_mission_status(
116109
while (
117110
request_status_failure_counter < settings.REQUEST_STATUS_FAILURE_COUNTER_LIMIT
118111
):
119-
if signal_exit.wait(0) or signal_mission_stopped.wait(0):
120-
break
121112
if request_status_failure_counter > 0:
122-
time.sleep(settings.REQUEST_STATUS_COMMUNICATION_RECONNECT_DELAY)
113+
await asyncio.sleep(settings.REQUEST_STATUS_COMMUNICATION_RECONNECT_DELAY)
123114

124115
try:
125116
mission_status = robot.mission_status(mission_id)
@@ -188,10 +179,8 @@ def get_mission_status_based_on_task_status(tasks: List[TASKS]) -> MissionStatus
188179
return MissionStatus.Successful
189180

190181

191-
def robot_monitor_mission(
182+
async def robot_monitor_mission(
192183
mission: Mission,
193-
signal_exit: Event,
194-
signal_mission_stopped: Event,
195184
robot: RobotInterface,
196185
request_inspection_upload: Callable[[InspectionTask], None],
197186
mqtt_publisher: MqttClientInterface,
@@ -204,114 +193,109 @@ def robot_monitor_mission(
204193
current_task.status = TaskStatus.NotStarted # type: ignore
205194
current_mission_status = MissionStatus.NotStarted
206195

207-
while True:
208-
if signal_exit.wait(0):
209-
return None
210-
if signal_mission_stopped.wait(0):
211-
if current_task is not None:
212-
current_task.status = TaskStatus.Cancelled
213-
publish_task_status(mqtt_publisher, current_task, mission.id)
214-
publish_mission_status(
215-
mqtt_publisher,
216-
mission.id,
217-
MissionStatus.Cancelled,
218-
ErrorMessage(
219-
ErrorReason.RobotMissionStatusException, "Mission cancelled"
220-
),
221-
)
222-
return None
223-
224-
if current_task:
225-
new_task_status: TaskStatus | None = None
226-
try:
227-
new_task_status = get_task_status(
228-
signal_exit, signal_mission_stopped, robot, current_task.id
229-
)
230-
231-
if current_task.status != new_task_status:
232-
current_task.status = new_task_status
233-
log_task_status(logger, current_task)
234-
publish_task_status(mqtt_publisher, current_task, mission.id)
196+
try:
197+
while True:
198+
if current_task:
199+
new_task_status: TaskStatus | None = None
200+
try:
201+
new_task_status = await get_task_status(robot, current_task.id)
235202

236-
if is_finished(new_task_status):
237-
if should_upload_inspections(current_task):
238-
request_inspection_upload(current_task) # type: ignore
239-
current_task = get_next_task(task_iterator)
240-
if current_task is not None:
241-
# This is not required, but does make reporting more responsive
242-
current_task.status = TaskStatus.InProgress
203+
if current_task.status != new_task_status:
204+
current_task.status = new_task_status
243205
log_task_status(logger, current_task)
244206
publish_task_status(mqtt_publisher, current_task, mission.id)
245-
except RobotTaskStatusException as e:
246-
logger.error(
247-
"Failed to collect task status. Error description: %s",
248-
e.error_description,
249-
)
250-
# Currently we only stop mission monitoring after failing to get mission status
251207

252-
new_mission_status: MissionStatus
253-
try:
254-
new_mission_status = get_mission_status(
255-
signal_exit, signal_mission_stopped, robot, mission.id
256-
)
257-
except RobotMissionStatusException as e:
258-
logger.exception("Failed to collect mission status")
259-
error_message = ErrorMessage(
260-
error_reason=e.error_reason,
261-
error_description=e.error_description,
262-
)
263-
if current_task:
264-
current_task.status = TaskStatus.Failed
265-
publish_task_status(mqtt_publisher, current_task, mission.id)
266-
publish_mission_status(
267-
mqtt_publisher,
268-
mission.id,
269-
MissionStatus.Failed,
270-
error_message,
271-
)
272-
break
273-
274-
if new_mission_status == MissionStatus.Cancelled or (
275-
new_mission_status
276-
not in [MissionStatus.NotStarted, MissionStatus.InProgress]
277-
and current_task is None # We wait for all task statuses
278-
):
279-
if (
280-
new_mission_status == MissionStatus.Cancelled
281-
and current_task is not None
282-
and current_task.status == TaskStatus.InProgress
283-
):
284-
current_task.status = TaskStatus.Cancelled
285-
publish_task_status(mqtt_publisher, current_task, mission.id)
286-
287-
# Standardises final mission status report
288-
new_mission_status = get_mission_status_based_on_task_status(mission.tasks)
289-
if error_message is None and new_mission_status in [
290-
MissionStatus.Failed,
291-
MissionStatus.Cancelled,
292-
]:
208+
if is_finished(new_task_status):
209+
if should_upload_inspections(current_task):
210+
request_inspection_upload(current_task) # type: ignore
211+
current_task = get_next_task(task_iterator)
212+
if current_task is not None:
213+
# This is not required, but does make reporting more responsive
214+
current_task.status = TaskStatus.InProgress
215+
log_task_status(logger, current_task)
216+
publish_task_status(
217+
mqtt_publisher, current_task, mission.id
218+
)
219+
except RobotTaskStatusException as e:
220+
logger.error(
221+
"Failed to collect task status. Error description: %s",
222+
e.error_description,
223+
)
224+
# Currently we only stop mission monitoring after failing to get mission status
225+
226+
new_mission_status: MissionStatus
227+
try:
228+
new_mission_status = await get_mission_status(robot, mission.id)
229+
except RobotMissionStatusException as e:
230+
logger.exception("Failed to collect mission status")
293231
error_message = ErrorMessage(
294-
error_reason=None,
295-
error_description="The mission failed because all tasks in the mission failed",
232+
error_reason=e.error_reason,
233+
error_description=e.error_description,
296234
)
297-
publish_mission_status(
298-
mqtt_publisher,
299-
mission.id,
300-
new_mission_status,
301-
error_message,
302-
)
303-
break
235+
if current_task:
236+
current_task.status = TaskStatus.Failed
237+
publish_task_status(mqtt_publisher, current_task, mission.id)
238+
publish_mission_status(
239+
mqtt_publisher,
240+
mission.id,
241+
MissionStatus.Failed,
242+
error_message,
243+
)
244+
break
304245

305-
if new_mission_status != current_mission_status:
306-
current_mission_status = new_mission_status
307-
publish_mission_status(
308-
mqtt_publisher,
309-
mission.id,
310-
current_mission_status,
311-
error_message,
312-
)
246+
if new_mission_status == MissionStatus.Cancelled or (
247+
new_mission_status
248+
not in [MissionStatus.NotStarted, MissionStatus.InProgress]
249+
and current_task is None # We wait for all task statuses
250+
):
251+
if (
252+
new_mission_status == MissionStatus.Cancelled
253+
and current_task is not None
254+
and current_task.status == TaskStatus.InProgress
255+
):
256+
current_task.status = TaskStatus.Cancelled
257+
publish_task_status(mqtt_publisher, current_task, mission.id)
313258

314-
time.sleep(settings.FSM_SLEEP_TIME)
259+
# Standardises final mission status report
260+
new_mission_status = get_mission_status_based_on_task_status(
261+
mission.tasks
262+
)
263+
if error_message is None and new_mission_status in [
264+
MissionStatus.Failed,
265+
MissionStatus.Cancelled,
266+
]:
267+
error_message = ErrorMessage(
268+
error_reason=None,
269+
error_description="The mission failed because all tasks in the mission failed",
270+
)
271+
publish_mission_status(
272+
mqtt_publisher,
273+
mission.id,
274+
new_mission_status,
275+
error_message,
276+
)
277+
break
278+
279+
if new_mission_status != current_mission_status:
280+
current_mission_status = new_mission_status
281+
publish_mission_status(
282+
mqtt_publisher,
283+
mission.id,
284+
current_mission_status,
285+
error_message,
286+
)
315287

316-
logger.info("Stopped monitoring mission")
317-
return error_message
288+
await asyncio.sleep(settings.FSM_SLEEP_TIME)
289+
logger.info("Stopped monitoring mission")
290+
return error_message
291+
except asyncio.CancelledError:
292+
if current_task is not None:
293+
current_task.status = TaskStatus.Cancelled
294+
publish_task_status(mqtt_publisher, current_task, mission.id)
295+
publish_mission_status(
296+
mqtt_publisher,
297+
mission.id,
298+
MissionStatus.Cancelled,
299+
ErrorMessage(ErrorReason.RobotMissionStatusException, "Mission cancelled"),
300+
)
301+
return None

0 commit comments

Comments
 (0)