PYROS-IRAP / pyros

Commit baada9d5aeaf8df20462bde463600e39083b693a

Authored by Etienne Pallier
1 parent 9b40508f
Exists in dev

New PRIO CMD management

Showing 10 changed files with 411 additions and 116 deletions   Show diff stats
Inline Side-by-side
cp_private_dev_to_private.sh
  Diff comments   View file @baada9d
1 1  
2 2 cp -p src/core/pyros_django/agent/AgentBasic.py privatedev/plugin/agent/AgentBasic.py
  3 +cp -p src/core/pyros_django/agent/AgentTriton.py privatedev/plugin/agent/AgentTriton.py
3 4  
4 5 rsync -avh --delete --exclude '.gitignore' --exclude 'plugin/README' --exclude 'config/README' privatedev/ private/
5 6  
... ...
privatedev/plugin/agent/AgentBasic.py
  Diff comments   View file @baada9d
... ... @@ -36,6 +36,7 @@ class AgentBasic(Agent):
36 36  
37 37  
38 38 # @override
  39 + '''
39 40 _AGENT_SPECIFIC_COMMANDS = [
40 41 # Format : (“cmd_name”, timeout, exec_mode)
41 42  
... ... @@ -51,6 +52,22 @@ class AgentBasic(Agent):
51 52 ("do_cmd_with_long_exec_time", 50, Agent.EXEC_MODE.THREAD),
52 53  
53 54 ]
  55 + '''
  56 + _AGENT_SPECIFIC_COMMANDS = {
  57 + # Format : (“cmd_name”, timeout, exec_mode)
  58 +
  59 + # Error raising commands
  60 + "do_cmd_unimplemented_and_declared": (3, Agent.EXEC_MODE.SEQUENTIAL),
  61 + "cmd_misnamed_and_declared": (3, Agent.EXEC_MODE.SEQUENTIAL),
  62 + "do_cmd_raising_some_exception": (3, Agent.EXEC_MODE.SEQUENTIAL),
  63 +
  64 + # Normal Commands
  65 + #("set_specific2", 5, 0),
  66 + "do_specific10": (1, Agent.EXEC_MODE.SEQUENTIAL),
  67 + "do_specific30": (3, Agent.EXEC_MODE.SEQUENTIAL),
  68 + "do_cmd_with_long_exec_time": (50, Agent.EXEC_MODE.SEQUENTIAL),
  69 +
  70 + }
54 71  
55 72 # Deactivate some tests, so that test scenario runs faster during DEV
56 73 # on DEV
... ... @@ -73,7 +90,10 @@ class AgentBasic(Agent):
73 90 #("self do_stop asap", 200, 'STOPPING asap', Agent.CMD_STATUS.CMD_EXECUTED),
74 91  
75 92 # get_specific_cmds
76   - (True, " self get_all_cmds ", 100,
  93 + # prio
  94 + #(True, " self get_all_cmds ", 100,
  95 + # noprio
  96 + (True, " self get_all_cmds noprio ", 100,
77 97 None,
78 98 #'do_specific10(arg1:int,arg2:int,arg3:float,arg4:str,arg5:typing.Tuple[int, str, int],arg6:typing.List[int]);do_specific30();do_cmd_raising_some_exception();do_cmd_unimplemented(U)',
79 99 Agent.CMD_STATUS.CMD_EXECUTED
... ... @@ -375,7 +395,7 @@ class AgentBasic(Agent):
375 395  
376 396 # Long time execution command
377 397 def do_cmd_with_long_exec_time(self):
378   - nbsec=8
  398 + nbsec=2
379 399  
380 400 res = f"1 - now sleeping {nbsec} sec"
381 401 self.CC.set_result(res, True)
... ...
privatedev/plugin/agent/AgentTriton.py
  Diff comments   View file @baada9d
... ... @@ -31,13 +31,15 @@ class AgentTriton(Agent):
31 31 #TEST_COMMANDS_DEST = "AgentB"
32 32 # Scenario to be executed
33 33  
34   - _AGENT_SPECIFIC_COMMANDS = [
35   - ("do_process_image",10,0),
36   - ]
  34 + _AGENT_SPECIFIC_COMMANDS = {
  35 + # Format : “cmd_name” : (timeout, exec_mode)
  36 +
  37 + "do_process_image" : (10, Agent.EXEC_MODE.SEQUENTIAL),
  38 + }
37 39  
38 40 _TEST_COMMANDS_LIST = [
39   - ("self do_process_image i.fit", 200, None, "CMD_EXECUTED"),
40   - ("self do_exit", 500, "STOPPING", "CMD_EXECUTED"),
  41 + (True, "self do_process_image i.fit", 200, None, Agent.CMD_STATUS.CMD_EXECUTED),
  42 + (True, "self do_stop", 500, "STOPPING", Agent.CMD_STATUS.CMD_EXECUTED),
41 43 ]
42 44 """
43 45 =================================================================
... ... @@ -45,7 +47,7 @@ class AgentTriton(Agent):
45 47 =================================================================
46 48 """
47 49  
48   - # @override
  50 + # @Override
49 51 '''
50 52 #def __init__(self, name:str=None, config_filename=None, RUN_IN_THREAD=True):
51 53 def __init__(self, config_filename=None):
... ... @@ -81,9 +83,8 @@ class AgentTriton(Agent):
81 83  
82 84 os.environ.update(env)
83 85  
84   - # @override
  86 + # @Override
85 87 def _init(self):
86   - super()._init()
87 88 triton_folder = self.get_config().get_agent_information(self.get_config().unit_name,self.name)["path"]
88 89 triton_folder_abs_path = pwd + "/" + triton_folder + "/"
89 90 self.triton_folder_abs_path = triton_folder_abs_path
... ...
src/core/pyros_django/agent/Agent.py
  Diff comments   View file @baada9d
... ... @@ -42,7 +42,7 @@ and execute them on reception at each iteration :
42 42 # For cmd parsing
43 43 from array import array
44 44 from datetime import datetime
45   -from typing import List, Tuple, Union, Any, Optional, Literal
  45 +from typing import Dict, List, Tuple, Union, Any, Optional, Literal
46 46 import ast
47 47 from inspect import signature
48 48  
... ... @@ -128,8 +128,8 @@ log.debug("DB2 used is:" + djangosettings.DATABASES["default"]["NAME"])
128 128 import platform
129 129 import random
130 130 import threading
131   -#import multiprocessing
132 131 from threading import Thread
  132 +import multiprocessing
133 133 #import multiprocessing
134 134 import time
135 135 import re
... ... @@ -320,8 +320,6 @@ class Agent:
320 320 - cmd (str) : the command name
321 321 - timeout (int) : the command timeout (in sec ; special values : 0=not executed ; -1=no timeout)
322 322 - exec_mode (EXEC_MODE) : EXEC_MODE.SEQUENTIAL, EXEC_MODE.THREAD, or EXEC_MODE.PROCESS
323   - '''
324   - #_AGENT_SPECIFIC_COMMANDS: List[ Tuple[str, int, int] ] = [
325 323 _AGENT_SPECIFIC_COMMANDS: List[ Tuple[str, int, EXEC_MODE] ] = [
326 324 # Format : (“cmd_name”, timeout, exec_mode)
327 325  
... ... @@ -330,6 +328,16 @@ class Agent:
330 328 ("do_specific3", 3, EXEC_MODE.THREAD),
331 329 ("do_specific4", 3, EXEC_MODE.PROCESS),
332 330 ]
  331 + '''
  332 + _AGENT_SPECIFIC_COMMANDS: Dict [ str, Tuple[int, EXEC_MODE] ] = {
  333 + # Format : “cmd_name” : (timeout, exec_mode)
  334 +
  335 + #"do_specific1" : (10, EXEC_MODE.SEQUENTIAL),
  336 + #"do_specific3" : (3, EXEC_MODE.THREAD),
  337 + #"do_specific4" : (3, EXEC_MODE.PROCESS),
  338 + }
  339 +
  340 +
333 341  
334 342 #
335 343 # --- FOR TEST ONLY ---
... ... @@ -538,7 +546,7 @@ class Agent:
538 546 >>>>> Thread: PID: 2690, Process Name: Process-3, Thread Name: MainThread
539 547 """
540 548 # with thread
541   - RUN_IN_THREAD = True
  549 + #RUN_IN_THREAD = True
542 550 # with process
543 551 #RUN_IN_THREAD = False
544 552  
... ... @@ -579,11 +587,6 @@ class Agent:
579 587 __agent_survey:AgentSurvey = None
580 588 __pending_commands:QuerySet = None # []
581 589  
582   - '''
583   - _current_device_cmd = None
584   - _current_device_cmd_thread = None
585   - '''
586   -
587 590 # List of agents I will send commands to
588 591 _my_client_agents_aliases = []
589 592 _my_client_agents = {}
... ... @@ -610,6 +613,7 @@ class Agent:
610 613 self.__test_cmd_received_num:int = 0 # only for tests
611 614 # Current Command running
612 615 self.__CC :Optional[AgentCmd] = None
  616 + self.__CC_thread :Union[StoppableThreadEvenWhenSleeping, multiprocessing.Process] = None
613 617 # Previous Command running
614 618 self.__CC_prev :Optional[AgentCmd] = None
615 619 # Current Command exception (if occurs)
... ... @@ -1082,7 +1086,7 @@ class Agent:
1082 1086 # MAIN loop #
1083 1087 ############@
1084 1088 self.__iter_num = 1
1085   - self.__CC, self.__CC_prev, self.__CCE = None, None, None
  1089 + self.__CC, self.__CC_thread, self.__CC_prev, self.__CCE = None, None, None, None
1086 1090 self.DO_MAIN_LOOP = True
1087 1091 while self.DO_MAIN_LOOP:
1088 1092 # EXIT because of nb of iterations ?
... ... @@ -1129,68 +1133,79 @@ class Agent:
1129 1133 #print("general cmds:", self.get_general_cmds())
1130 1134 ##print("all cmds:", self.get_all_cmds())
1131 1135  
1132   - # ROUTINE BEFORE (only if not IDLE)
  1136 + # I - ROUTINE BEFORE (only if not IDLE)
1133 1137 #if not self.IS_IDLE():
1134 1138 self.__routine_process_iter_start()
1135 1139  
1136   - # NEW CMD EXEC (if possible and exists)
1137   - #
1138   - # ONLY possible if :
1139   - # - no current command self.__CC
1140   - # OR
1141   - # - current command self.__CC finished (can be run in parallel)
1142   - # OR
1143   - # - PRIORITY cmd received
1144   - # => Then start and return next received cmd (only if exists)
1145   - # => Otherwise, return None
1146   - # This may throw Exception => stored in self.__CCE
1147   - # Not good because started (running) commands cannot access the current cmd (not yet in self.__CC)
1148   - ###cmd = self.__start_next_received_cmd_if_possible_and_exists()
1149   - # Better :
1150   - # Side effect : set as expired, unimplemented or invalid all affected commands
1151   - cmd = self.__get_next_received_valid_cmd_if_possible_and_exists()
1152   - if cmd:
1153   - self.__CCE, self.__CC_prev, self.__CC = None, self.__CC, cmd
1154   - # This way, started commands will be able to access the current command via self.__CC
1155   - #self.__CC = cmd
1156   - # Can throw exception => will be stored in self.__CCE
1157   - self.__start_next_received_cmd(cmd)
1158   -
1159   - # Check if running cmd timeout
1160   - if self.__CC and self.__CC.is_running() and self.__CC.is_exec_timeout():
1161   - self.__CC.set_as_exec_timeout()
1162   -
1163   - # If Current cmd is finished (SEQUENTIAL OR PARALLEL)
1164   - # => process exception if exists (and test result if in testing mode)
1165   - if ( self.__CC and self.__CC.is_finished() ) or self.__CCE :
1166   - self.__process_finished_cmd()
1167   - #self.__CC = None
1168   - self.__CCE, self.__CC_prev, self.__CC = None, self.__CC, cmd
1169   -
1170   -
1171   - # if no EXIT condition (no do_stop or do_restart command received)
1172   - if self.DO_MAIN_LOOP:
1173   -
1174   - # ROUTINE AFTER (only if not IDLE)
1175   - #if not self.IS_IDLE():
1176   - # Wait end of routine_process_iter_start before running new routine
1177   - while self.ROUTINE_ITER_START_IS_RUNNING: self.sleep(1)
1178   - self.__routine_process_iter_end()
  1140 + # II - CURRENT COMMAND MANAGEMENT (CC)
  1141 +
  1142 + # 1) PRIORITY cmd received ? => exec it immediately & sequentially (even if CC still running)
  1143 + while True:
  1144 + cmd_prio = self.__get_received_priority_cmd_if_exists()
  1145 + if cmd_prio is None: break
  1146 + print("*** PRIO CMD start ***")
  1147 + self.__start_next_received_cmd(cmd_prio)
  1148 + print("*** PRIO CMD end ***")
  1149 + # Prio cmd is run sequentially so must be finished
  1150 + assert cmd_prio.is_finished()
  1151 + # if STOP/RESTART condition: break
  1152 + if not self.DO_MAIN_LOOP: break
  1153 +
  1154 + # if EXIT condition (do_stop or do_restart command received)
  1155 + if not self.DO_MAIN_LOOP:
  1156 + self.__main_loop_end()
  1157 + return
1179 1158  
1180   - # TEST MODE only : send next command from test scenario (not waiting for current cmd to be finished)
1181   - #if self.is_in_test_mode():
1182   - self.__TEST_send_next_test_cmd()
  1159 + # 2) Start a new command (if no current cmd running and new command received)
  1160 + if self.__CC is None:
  1161 + # Side effect : set as expired, unimplemented or invalid all affected commands
  1162 + cmd = self.__get_next_received_valid_cmd_if_exists()
  1163 + if cmd:
  1164 + self.__CC = cmd
  1165 + # This way, started commands will be able to access the current command via self.__CC
  1166 + # Can throw exception => will be stored in self.__CCE
  1167 + self.__start_next_received_cmd(cmd)
  1168 +
  1169 + # 3) Check if CC finished
  1170 + # Current command running ?
  1171 + if self.__CC:
  1172 +
  1173 + # a) Check if exec timeout
  1174 + if self.__CC.is_running() and self.__CC.is_exec_timeout():
  1175 + self.__CC.set_as_exec_timeout()
  1176 +
  1177 + # b) If Current cmd is finished (SEQUENTIAL OR PARALLEL)
  1178 + # => process exception if exists (and test result if in testing mode)
  1179 + if self.__CC.is_finished() or self.__CCE :
  1180 + self.__process_finished_cmd()
  1181 + #self.__CC = None
  1182 + self.__CC, self.__CCE = None, None
  1183 +
  1184 + # if EXIT condition (do_stop or do_restart command received)
  1185 + if not self.DO_MAIN_LOOP:
  1186 + self.__main_loop_end()
  1187 + return
  1188 +
  1189 + # IV - ROUTINE AFTER (only if not IDLE)
  1190 + #if not self.IS_IDLE():
  1191 + # Wait end of routine_process_iter_start before running new routine
  1192 + while self.ROUTINE_ITER_START_IS_RUNNING: self.sleep(1)
  1193 + self.__routine_process_iter_end()
  1194 +
  1195 + # TEST MODE only : send next command from test scenario (not waiting for current cmd to be finished)
  1196 + #if self.is_in_test_mode():
  1197 + self.__TEST_send_next_test_cmd()
1183 1198  
1184   - #self.printd("====== END COMMMANDS PROCESSING ======")
  1199 + #self.printd("====== END COMMMANDS PROCESSING ======")
1185 1200  
1186   - #self.waitfor(self.mainloop_waittime)
  1201 + #self.waitfor(self.mainloop_waittime)
1187 1202  
1188 1203 # Wait end of routine_process_iter_end before starting new iteration
1189 1204 while self.ROUTINE_ITER_END_IS_RUNNING: self.sleep(1)
1190 1205 self.__main_loop_end()
1191 1206  
1192 1207  
1193   - def __get_next_received_valid_cmd_if_possible_and_exists(self)->Optional[AgentCmd]:
  1208 + def __get_next_received_valid_cmd_if_exists(self)->Optional[AgentCmd]:
1194 1209 self.__set_and_log_status(self.AGT_STATUS.IN_MAIN_LOOP_GET_NEXT_CMD)
1195 1210 cmd = None
1196 1211  
... ... @@ -1262,9 +1277,7 @@ class Agent:
1262 1277 commands = AgentCmd.get_pending_and_running_commands_for_agent(self.name)
1263 1278 if not commands.exists(): return None
1264 1279 for cmd in commands:
1265   - #if cmd.name in ("do_exit", "do_abort", "do_flush_pending_commands"): break
1266   - #if cmd.name in ("do_exit", "do_abort"): break
1267   - if cmd.is_agent_general_priority_cmd():
  1280 + if cmd.is_agent_general_priority_cmd_name():
1268 1281 # if prio cmd but "noprio" => do not consider it as prio
1269 1282 if 'noprio' in cmd.args:
1270 1283 continue
... ... @@ -1276,6 +1289,7 @@ class Agent:
1276 1289 if cmd.is_running():
1277 1290 return None
1278 1291 if not cmd.is_expired():
  1292 + #print("FOUND PRIO COMMAND", cmd)
1279 1293 return cmd
1280 1294 return None
1281 1295  
... ... @@ -1354,6 +1368,7 @@ class Agent:
1354 1368 log.info(f"The previous command {self.__CC_prev.name} is still running => Waiting for it to finish (waiting {wait_nbsec} sec...)")
1355 1369 self.waitfor(wait_nbsec)
1356 1370  
  1371 + '''
1357 1372 # Waiting for current ROUTINE BEFORE process to finish if still running
1358 1373 while self.ROUTINE_ITER_START_IS_RUNNING:
1359 1374 log.info(f"The ROUTINE BEFORE process is still running => Waiting for it to finish (waiting {wait_nbsec} sec...)")
... ... @@ -1363,6 +1378,7 @@ class Agent:
1363 1378 while self.ROUTINE_ITER_END_IS_RUNNING:
1364 1379 log.info(f"The ROUTINE AFTER process is still running => Waiting for it to finish (waiting {wait_nbsec} sec...)")
1365 1380 self.waitfor(wait_nbsec)
  1381 + '''
1366 1382  
1367 1383 log.info(f"{self.name}: Before exiting, calling do_things_before_exit()")
1368 1384 self._do_things_before_exit(stopper_agent_name)
... ... @@ -1670,6 +1686,7 @@ class Agent:
1670 1686 def sleep(self, nbsec:float):
1671 1687 log.info(f"Now, waiting (sleeping) for {nbsec} second(s)...")
1672 1688 time.sleep(nbsec)
  1689 +
1673 1690 # alias to sleep
1674 1691 def waitfor(self, nbsec:float):
1675 1692 '''
... ... @@ -2334,6 +2351,180 @@ class Agent:
2334 2351 self.printd("Logging data...")
2335 2352 '''
2336 2353  
  2354 +
  2355 +
  2356 +
  2357 +
  2358 +
  2359 + ######################################################################################################################
  2360 +
  2361 +
  2362 + def __run_thread_or_process(self, cmd_name:str, target)->Union[StoppableThreadEvenWhenSleeping,multiprocessing.Process]:
  2363 +
  2364 + # - THREAD exec
  2365 + if self.is_to_be_executed_in_thread(cmd_name):
  2366 + print("(run cmd in a thread)")
  2367 + #self.__CC_thread = StoppableThreadEvenWhenSleeping(target=self._thread_exec_cmd)
  2368 + return StoppableThreadEvenWhenSleeping(target=self.__thread_exec_cmd)
  2369 +
  2370 + # - PROCESS exec
  2371 + else:
  2372 + print("(run cmd in a process)")
  2373 + # close the database connection first, it will be re-opened in each process
  2374 + db.connections.close_all()
  2375 + #self.__CC_thread = multiprocessing.Process(target=self._thread_exec_cmd)
  2376 + return multiprocessing.Process(target=self.__thread_exec_cmd)
  2377 +
  2378 +
  2379 + #FIXME: imported from AgentDevice START
  2380 +
  2381 + """
  2382 + =================================================================
  2383 + FUNCTIONS RUN INSIDE A SUB-THREAD (OR A PROCESS) (thread_*())
  2384 + =================================================================
  2385 + """
  2386 +
  2387 + def tprintd(self, *args, **kwargs):
  2388 + print('(THREAD):', *args, *kwargs)
  2389 +
  2390 + def __thread_exec_cmd(self):
  2391 + assert self.__CC_thread is not None
  2392 + cmd = self.__CC
  2393 +
  2394 + # thread execution setting up
  2395 + self.__thread_exec_cmd_start(cmd)
  2396 +
  2397 + # Exit if I am a Thread and was asked to stop
  2398 + if self.RUN_IN_THREAD and threading.current_thread().stopped():
  2399 + self.tprintd(f">>>>> Thread (cmd {cmd.name}): I received the stop signal, so I stop (in error)")
  2400 + exit(1)
  2401 +
  2402 + else:
  2403 + try:
  2404 + res = self.__exec_method(func_name, args, cmd)
  2405 + except (CmdExceptionBadArgs, CmdExceptionExecError) as e:
  2406 + # managed at higher level
  2407 + self.__CC_thread.terminate()
  2408 + #cmd.set_as_killed_by(type(self).__name__)
  2409 + raise
  2410 +
  2411 + cmd.set_result(res)
  2412 +
  2413 + '''
  2414 + # processing body (main)
  2415 + # Specific case of the EVAL command
  2416 + #if self._current_device_cmd.name.startswith("eval"):
  2417 + if self._current_device_cmd.name == "do_eval":
  2418 + self.thread_set_total_steps_number(1)
  2419 + self.thread_exec_device_cmd_step(1, self.cmd_step_eval)
  2420 + #return
  2421 + # to be overriden by subclasses
  2422 + else:
  2423 + self.thread_exec_specific_cmd_main()
  2424 + '''
  2425 +
  2426 + # thread execution tearing down
  2427 + self.__thread_exec_cmd_end(cmd)
  2428 + '''
  2429 + except TimeoutError:
  2430 + pass
  2431 + '''
  2432 +
  2433 + #def exec_specific_cmd_start(self, cmd:Command):
  2434 + def __thread_exec_cmd_start(self, cmd:AgentCmd):
  2435 + #cmd = self.__CC
  2436 + """ specific command execution setting up """
  2437 + #cmd = self.get_current_device_cmd()
  2438 + self.tprintd(">>>>> Thread: starting execution of command", cmd.name)
  2439 + self.tprintd(">>>>> Thread: PID: %s, Process Name: %s, Thread Name: %s" % (
  2440 + os.getpid(),
  2441 + multiprocessing.current_process().name,
  2442 + threading.current_thread().name)
  2443 + )
  2444 + cmd.set_as_running()
  2445 + """
  2446 + if self.RUN_IN_THREAD:
  2447 + cmd.set_as_running()
  2448 + else:
  2449 + with transaction.atomic():
  2450 + cmd.set_as_running()
  2451 + """
  2452 +
  2453 + def __thread_exec_cmd_end(self, cmd:AgentCmd):
  2454 + """ specific command execution tearing down """
  2455 + #cmd = self.__CC
  2456 + cmd.set_as_processed()
  2457 + """
  2458 + if self.RUN_IN_THREAD:
  2459 + cmd.set_as_processed()
  2460 + else:
  2461 + with transaction.atomic():
  2462 + cmd.set_as_processed()
  2463 + """
  2464 + self.tprintd(f">>>>> Thread: ended execution of command '{cmd.name}'")
  2465 + ###cmd = None
  2466 + # No more current thread
  2467 + #self._current_device_cmd_thread = None
  2468 +
  2469 +
  2470 + #FIXME:
  2471 + # NEW A MOI (EP)
  2472 + def __exec_in_thread():
  2473 + self.print("Launching device cmd in a thread (or process)...")
  2474 + # Run in a thread
  2475 + if self.RUN_IN_THREAD:
  2476 + self.printd("(run device cmd in a thread)")
  2477 + self._current_device_cmd_thread = StoppableThreadEvenWhenSleeping(target=self._thread_exec_device_cmd)
  2478 + #self._current_device_cmd_thread = StoppableThreadEvenWhenSleeping(target=self.exec_specific_cmd, args=(cmd,))
  2479 + #self._current_thread = threading.Thread(target=self.exec_command)
  2480 + #self._current_device_cmd_thread = StoppableThread(target=self.exec_specific_cmd, args=(cmd,))
  2481 + #self._current_device_cmd_thread = threading.Thread(target=self.exec_specific_cmd, args=(cmd,))
  2482 + #self._current_device_cmd_thread = thread_with_exception('thread test')
  2483 +
  2484 + # Run in a process
  2485 + else:
  2486 + self.printd("(run cmd in a process)")
  2487 + # close the database connection first, it will be re-opened in each process
  2488 + db.connections.close_all()
  2489 + self._current_device_cmd_thread = multiprocessing.Process(target=self._thread_exec_device_cmd)
  2490 + #self._current_device_cmd_thread = multiprocessing.Process(target=self.exec_specific_cmd, args=(cmd,))
  2491 +
  2492 + self._current_device_cmd_thread.start()
  2493 + #self._current_device_cmd_thread = threading.Thread(target=self.exec_specific_cmd, args=(cmd,))
  2494 + #self._current_device_cmd_thread = thread_with_exception('thread test')
  2495 + #my_thread.join()
  2496 + #self.waitfor(self.subloop_waittime)
  2497 + self.printd("Ending specific process (thread has been launched)")
  2498 +
  2499 +
  2500 + #FIXME: imported from AgentDevice STOP
  2501 +
  2502 + ######################################################################################################################
  2503 +
  2504 + def get_specific_cmd_parameters(self, cmd_name:str)->Tuple[int,EXEC_MODE]:
  2505 + return self._AGENT_SPECIFIC_COMMANDS.get(cmd_name)
  2506 +
  2507 + def get_specific_cmd_timeout(self, cmd_name:str)->int:
  2508 + timeout, _ = self.get_specific_cmd_parameters(cmd_name)
  2509 + return timeout
  2510 +
  2511 + def get_specific_cmd_exec_mode(self, cmd_name:str)->EXEC_MODE:
  2512 + _, exec_mode = self.get_specific_cmd_parameters(cmd_name)
  2513 + return exec_mode
  2514 +
  2515 +
  2516 + def is_to_be_executed_sequentially(self, cmd_name:str)->bool:
  2517 + return self.get_specific_cmd_exec_mode(cmd_name) == self.EXEC_MODE.SEQUENTIAL
  2518 +
  2519 + def is_to_be_executed_concurrently(self, cmd_name:str)->bool:
  2520 + return self.get_specific_cmd_exec_mode(cmd_name) != self.EXEC_MODE.SEQUENTIAL
  2521 +
  2522 + def is_to_be_executed_in_thread(self, cmd_name:str)->bool:
  2523 + return self.get_specific_cmd_exec_mode(cmd_name) == self.EXEC_MODE.THREAD
  2524 +
  2525 + def is_to_be_executed_in_process(self, cmd_name:str)->bool:
  2526 + return self.get_specific_cmd_exec_mode(cmd_name) == self.EXEC_MODE.PROCESS
  2527 +
2337 2528 def __exec_cmd_from_its_name(self, cmd:AgentCmd)->Any:
2338 2529 #print(dir(self))
2339 2530 '''
... ... @@ -2348,8 +2539,8 @@ class Agent:
2348 2539  
2349 2540 methods_list = [method for method in dir(self) if callable(getattr(self, method))]
2350 2541 #print(methodsList)
2351   - func = cmd.name
2352   - if func not in methods_list: raise CmdExceptionUnimplemented(cmd)
  2542 + func_name = cmd.name
  2543 + if func_name not in methods_list: raise CmdExceptionUnimplemented(cmd)
2353 2544 ##f = getattr(self, func)
2354 2545 ###print(func, ' => ', signature(f))
2355 2546  
... ... @@ -2397,13 +2588,38 @@ class Agent:
2397 2588 #except ValueError as e: newarg = arg
2398 2589 args.append(arg)
2399 2590  
  2591 + # SEQUENTIAL OR THREAD/PROCESS exec ?
  2592 + IS_SEQ = True
  2593 +
  2594 + if self.is_agent_specific_cmd(cmd):
  2595 + IS_SEQ = self.is_to_be_executed_sequentially(cmd.name)
  2596 + #FIXME: gérer aussi le timeout, comment ???
  2597 +
2400 2598 cmd.set_as_running()
2401 2599 self._sleep_as_soon_as_running()
2402 2600  
2403 2601 # Command EXECUTION
  2602 + # - SEQUENTIAL exec
  2603 + try:
  2604 + if IS_SEQ:
  2605 + print("*** SEQ EXEC ***")
  2606 + return self.__exec_method(func_name, args, cmd)
  2607 + # - CONCURRENT exec
  2608 + else:
  2609 + print("*** CONCURRENT EXEC ***")
  2610 + self.__CC_thread = self.__run_thread_or_process(cmd.name, target=self.__thread_exec_cmd)
  2611 + self.__CC_thread.start()
  2612 + except (CmdExceptionBadArgs, CmdExceptionExecError) as e:
  2613 + # managed at higher level
  2614 + raise
  2615 +
  2616 +
  2617 +
  2618 + def __exec_method(self, func_name:str, args:List, cmd:AgentCmd)->Any:
  2619 +
2404 2620 try:
2405 2621 # equivalent to calling self.func(*cmd.args)
2406   - return getattr(self, func)(*args)
  2622 + return getattr(self, func_name)(*args)
2407 2623  
2408 2624 # Replace low level exception with high level one (CmdExceptionBadArgs)
2409 2625 except (TypeError, AttributeError, ValueError, CmdExceptionBadArgs, AssertionError) as e:
... ... @@ -2451,11 +2667,14 @@ class Agent:
2451 2667 self._exec_agent_specific_cmd(cmd)
2452 2668 '''
2453 2669  
2454   - def is_agent_specific_cmd(self, cmd:AgentCmd):
  2670 + def is_agent_specific_cmd(self, cmd:AgentCmd)->bool:
2455 2671 #return cmd.name in self.AGENT_SPECIFIC_COMMANDS
2456 2672 #return (cmd.name,) in self.AGENT_SPECIFIC_COMMANDS
  2673 + '''
2457 2674 for (cmd_name,_,_) in self._AGENT_SPECIFIC_COMMANDS:
2458 2675 if cmd.name == cmd_name : return True
  2676 + '''
  2677 + return self._AGENT_SPECIFIC_COMMANDS.get(cmd.name) is not None
2459 2678  
2460 2679 '''
2461 2680 def _exec_agent_specific_cmd(self, cmd:Command):
... ... @@ -2463,7 +2682,7 @@ class Agent:
2463 2682 self.exec_cmd_from_its_name(cmd)
2464 2683 '''
2465 2684  
2466   - def is_in_test_mode(self):
  2685 + def is_in_test_mode(self)->bool:
2467 2686 return self.TEST_MODE
2468 2687  
2469 2688  
... ... @@ -2474,7 +2693,8 @@ class Agent:
2474 2693 # ================================================================================================
2475 2694 ###
2476 2695  
2477   - def get_all_cmds(self)->str:
  2696 + def get_all_cmds(self, noprio:str=None)->str:
  2697 + if noprio: assert noprio=="noprio"
2478 2698 general_cmds = self.get_general_cmds()
2479 2699 specific_cmds = self.get_specific_cmds()
2480 2700 all_cmds = general_cmds
... ... @@ -2482,12 +2702,12 @@ class Agent:
2482 2702 return all_cmds
2483 2703  
2484 2704 def get_general_cmds(self)->str:
2485   - return self.__get_cmds_with_args_from_names_list(AgentCmd._AGENT_GENERAL_COMMANDS)
  2705 + return self.__get_cmds_with_args_from_names(AgentCmd._AGENT_GENERAL_COMMANDS)
2486 2706  
2487 2707 def get_specific_cmds(self)->str:
2488   - return self.__get_cmds_with_args_from_names_list(self._AGENT_SPECIFIC_COMMANDS)
  2708 + return self.__get_cmds_with_args_from_names(self._AGENT_SPECIFIC_COMMANDS)
2489 2709  
2490   - def __get_cmds_with_args_from_names_list(self, names_list:List)->str:
  2710 + def __get_cmds_with_args_from_names(self, cmd_names:Union[List,Dict])->str:
2491 2711 '''
2492 2712 Return the list of all specific cmds, with their arguments type (if exists)
2493 2713 - Each cmd is separated with a ';' and presented with this format :
... ... @@ -2501,9 +2721,9 @@ class Agent:
2501 2721 cmds = ""
2502 2722 # For each command
2503 2723 #for command_tuple in self._AGENT_SPECIFIC_COMMANDS:
2504   - for command_tuple in names_list:
  2724 + for cmd_name in cmd_names:
2505 2725  
2506   - cmd_name = command_tuple[0] if isinstance(command_tuple, tuple) else command_tuple
  2726 + #cmd_name = command_tuple[0] if isinstance(command_tuple, tuple) else command_tuple
2507 2727 #print(cmd_name)
2508 2728 cmds += cmd_name
2509 2729  
... ... @@ -2770,7 +2990,10 @@ class Agent:
2770 2990 #cmd_full_name, validity, res_expected, after_status = next(self.TEST_COMMANDS, (None,None,None,None))
2771 2991 DO_IT = False
2772 2992 while not DO_IT:
2773   - DO_IT, cmd_full_name, validity, expected_res, expected_status = next(self.TEST_COMMANDS, (False,None,None,None,None))
  2993 + cmd_params = next(self.TEST_COMMANDS, (False,None,None,None,None))
  2994 + #print(cmd_params)
  2995 + DO_IT, cmd_full_name, validity, expected_res, expected_status = cmd_params
  2996 + ###DO_IT, cmd_full_name, validity, expected_res, expected_status = next(self.TEST_COMMANDS, (False,None,None,None,None))
2774 2997 ##print(DO_IT, cmd_full_name)
2775 2998 if cmd_full_name is None: return None
2776 2999 #print(expected_final_status)
... ... @@ -2956,16 +3179,21 @@ class Agent:
2956 3179 #print("num", cmd_num)
2957 3180 i=1
2958 3181 cmd_num=0
  3182 + print(self._TEST_COMMANDS_LIST)
2959 3183 while cmd_num <= cmd_searched_num:
2960 3184 # Format : (DO_IT, "self cmd_name cmd_args", timeout, "expected_result", expected_status),
2961 3185 cmd = self._TEST_COMMANDS_LIST[i-1]
2962 3186 #print(cmd)
2963 3187 DO_IT, cmd_name, _, _, _ = cmd
2964   - if DO_IT:
  3188 + # Remove dest (self) from cmd name
  3189 + #cmd_name = re.sub(r"\s+", " ", cmd_name).strip().split(' ')[1]
  3190 + #_, cmd_name_and_args = re.sub(r"\s+", " ", cmd_name).strip().split(' ')
  3191 + #_, _, cmd_name_and_args = re.sub(r"\s+", " ", cmd_name).strip().partition(' ')
  3192 + _, cmd_name, *cmd_args = re.sub(r"\s+", " ", cmd_name).strip().split(' ')
  3193 + if DO_IT and not AgentCmd._is_agent_general_priority_cmd(cmd_name, cmd_args) :
2965 3194 cmd_num+=1
2966 3195 if cmd_num == cmd_searched_num:
2967   - cmd_name = re.sub(r"\s+", " ", cmd_name).strip().split(' ')[1]
2968   - ##print(cmd_name, self.CC.name)
  3196 + print(cmd_name, self.CC.name)
2969 3197 assert cmd_name == self.CC.name
2970 3198 break
2971 3199 i+=1
... ...
src/core/pyros_django/agent/AgentBasic.py
  Diff comments   View file @baada9d
... ... @@ -36,6 +36,7 @@ class AgentBasic(Agent):
36 36  
37 37  
38 38 # @override
  39 + '''
39 40 _AGENT_SPECIFIC_COMMANDS = [
40 41 # Format : (“cmd_name”, timeout, exec_mode)
41 42  
... ... @@ -51,6 +52,22 @@ class AgentBasic(Agent):
51 52 ("do_cmd_with_long_exec_time", 50, Agent.EXEC_MODE.THREAD),
52 53  
53 54 ]
  55 + '''
  56 + _AGENT_SPECIFIC_COMMANDS = {
  57 + # Format : (“cmd_name”, timeout, exec_mode)
  58 +
  59 + # Error raising commands
  60 + "do_cmd_unimplemented_and_declared": (3, Agent.EXEC_MODE.SEQUENTIAL),
  61 + "cmd_misnamed_and_declared": (3, Agent.EXEC_MODE.SEQUENTIAL),
  62 + "do_cmd_raising_some_exception": (3, Agent.EXEC_MODE.SEQUENTIAL),
  63 +
  64 + # Normal Commands
  65 + #("set_specific2", 5, 0),
  66 + "do_specific10": (1, Agent.EXEC_MODE.SEQUENTIAL),
  67 + "do_specific30": (3, Agent.EXEC_MODE.SEQUENTIAL),
  68 + "do_cmd_with_long_exec_time": (50, Agent.EXEC_MODE.SEQUENTIAL),
  69 +
  70 + }
54 71  
55 72 # Deactivate some tests, so that test scenario runs faster during DEV
56 73 # on DEV
... ... @@ -73,7 +90,10 @@ class AgentBasic(Agent):
73 90 #("self do_stop asap", 200, 'STOPPING asap', Agent.CMD_STATUS.CMD_EXECUTED),
74 91  
75 92 # get_specific_cmds
76   - (True, " self get_all_cmds ", 100,
  93 + # prio
  94 + #(True, " self get_all_cmds ", 100,
  95 + # noprio
  96 + (True, " self get_all_cmds noprio ", 100,
77 97 None,
78 98 #'do_specific10(arg1:int,arg2:int,arg3:float,arg4:str,arg5:typing.Tuple[int, str, int],arg6:typing.List[int]);do_specific30();do_cmd_raising_some_exception();do_cmd_unimplemented(U)',
79 99 Agent.CMD_STATUS.CMD_EXECUTED
... ... @@ -375,7 +395,7 @@ class AgentBasic(Agent):
375 395  
376 396 # Long time execution command
377 397 def do_cmd_with_long_exec_time(self):
378   - nbsec=8
  398 + nbsec=2
379 399  
380 400 res = f"1 - now sleeping {nbsec} sec"
381 401 self.CC.set_result(res, True)
... ...
src/core/pyros_django/agent/AgentSST.py
  Diff comments   View file @baada9d
... ... @@ -26,11 +26,13 @@ class AgentSST(Agent):
26 26 subprocess_dict = {}
27 27 agent_in_mode_test = {}
28 28  
29   - _AGENT_SPECIFIC_COMMANDS = [
30   - ("do_stop_agent",10,0),
31   - ("do_restart_agent",20,0),
32   - ("do_start_agent",20,0),
33   - ]
  29 + _AGENT_SPECIFIC_COMMANDS = {
  30 + # Format : “cmd_name” : (timeout, exec_mode)
  31 +
  32 + "do_stop_agent" : (10, Agent.EXEC_MODE.SEQUENTIAL),
  33 + "do_restart_agent" : (20, Agent.EXEC_MODE.SEQUENTIAL),
  34 + "do_start_agent" : (20, Agent.EXEC_MODE.SEQUENTIAL),
  35 + }
34 36  
35 37 _TEST_COMMANDS_LIST = [
36 38 (True, "self get_mode", 200, "MODE = ATTENTIVE", Agent.CMD_STATUS.CMD_EXECUTED),
... ...
src/core/pyros_django/agent/AgentTriton.py
  Diff comments   View file @baada9d
... ... @@ -31,13 +31,15 @@ class AgentTriton(Agent):
31 31 #TEST_COMMANDS_DEST = "AgentB"
32 32 # Scenario to be executed
33 33  
34   - _AGENT_SPECIFIC_COMMANDS = [
35   - ("do_process_image",10,0),
36   - ]
  34 + _AGENT_SPECIFIC_COMMANDS = {
  35 + # Format : “cmd_name” : (timeout, exec_mode)
  36 +
  37 + "do_process_image" : (10, Agent.EXEC_MODE.SEQUENTIAL),
  38 + }
37 39  
38 40 _TEST_COMMANDS_LIST = [
39   - ("self do_process_image i.fit", 200, None, "CMD_EXECUTED"),
40   - ("self do_exit", 500, "STOPPING", "CMD_EXECUTED"),
  41 + (True, "self do_process_image i.fit", 200, None, Agent.CMD_STATUS.CMD_EXECUTED),
  42 + (True, "self do_stop", 500, "STOPPING", Agent.CMD_STATUS.CMD_EXECUTED),
41 43 ]
42 44 """
43 45 =================================================================
... ...
src/core/pyros_django/common/models.py
  Diff comments   View file @baada9d
... ... @@ -999,9 +999,11 @@ class AgentCmd(models.Model):
999 999 # By default, no args
1000 1000 #cmd_args = None
1001 1001 cmd_args = []
  1002 + '''
1002 1003 cmd_name = self.full_name
1003 1004 if ' ' in cmd_name:
1004   - cmd_name, *cmd_args = cmd_name.split(' ')
  1005 + '''
  1006 + cmd_name, *cmd_args = self.full_name.split(' ')
1005 1007 return cmd_name, cmd_args
1006 1008  
1007 1009  
... ... @@ -1078,15 +1080,27 @@ class AgentCmd(models.Model):
1078 1080 return self.name in self._AGENT_GENERAL_COMMANDS
1079 1081 # "CMD_OUTOFDATE" # cde périmée
1080 1082  
  1083 + def is_agent_general_priority_cmd_name(self)->bool:
  1084 + return self.name in self._AGENT_GENERAL_PRIORITY_COMMANDS
  1085 +
1081 1086 def is_agent_general_priority_cmd(self)->bool:
1082   - #return self.name in self._AGENT_GENERAL_PRIORITY_COMMANDS
1083 1087 #return AgentCmd._is_agent_general_priority_cmd(self.name)
1084 1088 #return self.__class__._is_agent_general_priority_cmd(self.name)
1085   - return type(self)._is_agent_general_priority_cmd(self.name)
  1089 + #return type(self)._is_agent_general_priority_cmd(self.name)
  1090 + #return type(self)._is_agent_general_priority_cmd(self.full_name)
  1091 + return type(self)._is_agent_general_priority_cmd(self.name, self.args)
1086 1092  
1087 1093 @classmethod
1088   - def _is_agent_general_priority_cmd(cls, cmd_name)->bool:
1089   - return cmd_name in cls._AGENT_GENERAL_PRIORITY_COMMANDS
  1094 + #def _is_agent_general_priority_cmd(cls, cmd_name_and_args=[])->bool:
  1095 + def _is_agent_general_priority_cmd(cls, cmd_name, cmd_args=[])->bool:
  1096 + #print(cmd_name_and_args)
  1097 + #cmd_name, cmd_args = re.sub(r"\s+", " ", cmd_name_and_args).strip().split(' ')
  1098 + #print(cmd_name, cmd_args)
  1099 + if cmd_name not in cls._AGENT_GENERAL_PRIORITY_COMMANDS: return False
  1100 + if cmd_args and cmd_args[0] == 'noprio': return False
  1101 + #return cmd_name in cls._AGENT_GENERAL_PRIORITY_COMMANDS and cmd_args and cmd_args[0] != 'noprio'
  1102 + return True
  1103 +
1090 1104  
1091 1105 def is_read(self)->bool:
1092 1106 return self.r_read_time is not None
... ...
src/core/pyros_django/observation_manager/AgentImagesCalibrator.py
  Diff comments   View file @baada9d
... ... @@ -38,11 +38,16 @@ class AgentImagesCalibrator(Agent):
38 38 RUNNING_COMPUTE_RON_GAIN = 2
39 39  
40 40 # TODO: Redefine valid timeout
41   - _AGENT_SPECIFIC_COMMANDS = [
42   - ("do_create_test_images_1",60, 0), # self.EXEC_MODE.SEQUENTIAL
43   - ("do_create_test_images_2",60, 0), # self.EXEC_MODE.THREAD
44   - ("do_stop_current_processing",60, 0), # self.EXEC_MODE.PROCESS
45   - ]
  41 + _AGENT_SPECIFIC_COMMANDS = {
  42 + # Format : “cmd_name” : (timeout, exec_mode)
  43 +
  44 + "do_create_test_images_1" : (60, Agent.EXEC_MODE.SEQUENTIAL),
  45 + "do_create_test_images_2" : (60, Agent.EXEC_MODE.SEQUENTIAL),
  46 + "do_stop_current_processing" : (60, Agent.EXEC_MODE.SEQUENTIAL),
  47 + }
  48 +
  49 +
  50 +
46 51  
47 52 # Scenario to be executed
48 53 # "self do_stop_current_processing"
... ...
src/core/pyros_django/observation_manager/AgentImagesProcessor.py
  Diff comments   View file @baada9d
... ... @@ -35,11 +35,13 @@ class AgentImagesProcessor(Agent):
35 35 RUNNING_COMPUTE_RON_GAIN = 2
36 36  
37 37 # TODO: Redefine valid timeout
38   - _AGENT_SPECIFIC_COMMANDS = [
39   - ("do_create_test_images_1",60, 0), # self.EXEC_MODE.SEQUENTIAL
40   - ("do_create_test_images_2",60, 0), # self.EXEC_MODE.THREAD
41   - ("do_stop_current_processing",60, 0), # self.EXEC_MODE.PROCESS
42   - ]
  38 + _AGENT_SPECIFIC_COMMANDS = {
  39 + # Format : “cmd_name” : (timeout, exec_mode)
  40 +
  41 + "do_create_test_images_1" : (60, Agent.EXEC_MODE.SEQUENTIAL),
  42 + "do_create_test_images_2" : (60, Agent.EXEC_MODE.SEQUENTIAL),
  43 + "do_stop_current_processing" : (60, Agent.EXEC_MODE.SEQUENTIAL),
  44 + }
43 45  
44 46 # Scenario to be executed
45 47 # "self do_stop_current_processing"
... ...