#!/usr/bin/env python3 import time import sys ##import utils.Logger as L ##from .Agent import Agent ##sys.path.append("..") ###from agent.Agent import Agent, build_agent sys.path.append("../../../..") from src.core.pyros_django.agent.Agent import Agent, build_agent from typing import List, Tuple, Union, Any ##log = L.setupLogger("AgentXTaskLogger", "AgentX") class AgentBasic(Agent): # FOR TEST ONLY # Run this agent in simulator mode TEST_MODE = False # Run the assertion tests at the end TEST_WITH_FINAL_TEST = False #TEST_MAX_DURATION_SEC = None TEST_MAX_DURATION_SEC = 300 # Who should I send commands to ? TEST_COMMANDS_DEST = "myself" #TEST_COMMANDS_DEST = "AgentB" # Scenario to be executed # @override AGENT_SPECIFIC_COMMANDS = [ # Format : (“cmd_name”, timeout, exec_mode) ("do_specific10", 10, 0), #("set_specific2", 5, 0), ("do_specific30", 3, 0), ###("existing_but_unimplemented_cmd", 3, 0), ] # @override TEST_COMMANDS_LIST = [ # Format : ("self cmd_name cmd_args", timeout, "expected_result", expected_status), #("self do_stop now", 200, '', Agent.CMD_STATUS.CMD_EXECUTED), # do_stop #("self do_stop now", 200, 'STOPPING now', Agent.CMD_STATUS.CMD_EXECUTED), #("self do_stop", 200, 'STOPPING asap', Agent.CMD_STATUS.CMD_EXECUTED), #("self do_stop asap", 200, 'STOPPING asap', Agent.CMD_STATUS.CMD_EXECUTED), # do_restart ("self do_restart now", 200, 'RESTARTING now', Agent.CMD_STATUS.CMD_EXECUTED), #("self do_stop", 200, 'STOPPING asap', Agent.CMD_STATUS.CMD_EXECUTED), # ------------------------------ # ------ A - ERROR CASES ------- # ------------------------------ # - Error case 1 - unimplemented command in agent specific commands list # get_specific_cmds KO (with existing_but_unimplemented_cmd) #("self get_specific_cmds", 200, 'EXCEPTION - One specific cmd is unimplemented: existing_but_unimplemented_cmd', Agent.CMD_STATUS.CMD_EXEC_ERROR), # get_specific_cmds OK (all commands implemented) ("self get_specific_cmds", 200, 'do_specific10(arg1:int,arg2:int,arg3:float,arg4:str,arg5:typing.Tuple[int, str, int],arg6:typing.List[int]);do_specific30()', Agent.CMD_STATUS.CMD_EXECUTED), # - Error case 2 - CMD_INVALID # unknwon command ##FIXME: ("self unexisting_cmd", 200, '', Agent.CMD_STATUS.CMD_UNKNOWN), ("self unexisting_cmd", 200, '', Agent.CMD_STATUS.CMD_INVALID), # bad parameters (missing or too many, or bad type) # missing args ("self do_specific10", 200, '', Agent.CMD_STATUS.CMD_INVALID), # missing args ("self do_specific10 2", 200, '', Agent.CMD_STATUS.CMD_INVALID), # - too many args ("self do_specific10 2 2 3.5 titi (3,'titi',5) [1,3,5,7,9] 4", 200, '', Agent.CMD_STATUS.CMD_INVALID), # - bad type ("self do_specific10 'toto' 2 3.5 titi (3,'titi',5) [1,3,5,7,9]", 200, '', Agent.CMD_STATUS.CMD_INVALID), # - OK #("self do_specific10 2 2 3.5 titi (3,'titi',5) [1,3,5,7,9]", 200, '16.5', Agent.CMD_STATUS.CMD_EXECUTED), # - Error case 3 - CMD_UNIMPLEMENTED ("self existing_but_unimplemented_cmd", 200, '', Agent.CMD_STATUS.CMD_UNIMPLEMENTED), # - Error case 4 - CMD_EXPIRED # This command has a validity of 0s and thus should be tagged as "expired" ("self set_mode ATTENTIVE", 0, "MODE is ATTENTIVE", Agent.CMD_STATUS.CMD_EXPIRED), # - Error case 5 - CMD_SKIPPED # a) In mode ROUTINE, SPECIFIC commands should be skipped as the agent is not ATTENTIVE ("self set_mode ROUTINE", 200, "MODE is ROUTINE", Agent.CMD_STATUS.CMD_EXECUTED), ("self do_specific10 2 2 3.5 titi (3,'titi',5) [1,3,5,7,9]", 200, None, Agent.CMD_STATUS.CMD_SKIPPED), # b) Back to mode ATTENTIVE, SPECIFIC commands should be executed ("self set_mode ATTENTIVE", 200, "MODE is ATTENTIVE", Agent.CMD_STATUS.CMD_EXECUTED), ("self do_specific10 2 2 3.5 titi (3,'titi',5) [1,3,5,7,9]", 200, '16.5', Agent.CMD_STATUS.CMD_EXECUTED), # ------------------------------ # ------ B - NORMAL CASES ------- # ------------------------------ # do_restart => #iteration should restart at 1 (so at least it should be <= 4) #("self do_restart now", 200, 'RESTARTING now', Agent.CMD_STATUS.CMD_EXECUTED), # FIXME: ajouter la possibilité d'utiliser des expressions pour tester le résultat : #("self get_iteration_number", 200, 'EXPR:<=4', Agent.CMD_STATUS.CMD_EXECUTED), # 1) First, 3 EXCEPTION CASES (uncomment to activate exception) # Each of these lines will stop execution with an exception # ------------------------------------------------------------ # - Agent command, unknown => ko, UnknownCmdException #("self do_unknown", 10, None,None), # - Agent general command malformed (missing arg) => ko, AgentCmdBadArgsException #("Agent set_mode", 10, None,None), # - Agent specific command, known but not implemented => ko, AgentCmdUnimplementedException #("self set_specific2", 10, None,None), # - Agent specific command, implemented but missing args => ko, AgentCmdBadArgsException #(" self do_specific1 1 ", 10, None,None), # 2) NORMAL CASES (test scenario) # All these commands should be executed without error, from the 1st to the last one # ------------------------------- # Agent general command ("self set_mode ROUTINE", 200, "MODE is ROUTINE", Agent.CMD_STATUS.CMD_EXECUTED), ("self get_mode", 200, "MODE is ROUTINE", Agent.CMD_STATUS.CMD_EXECUTED), ("self set_mode ATTENTIVE", 200, "MODE is ATTENTIVE", Agent.CMD_STATUS.CMD_EXECUTED), ("self get_mode", 200, "MODE is ATTENTIVE", Agent.CMD_STATUS.CMD_EXECUTED), # End test ("self do_stop", 200, 'STOPPING asap', Agent.CMD_STATUS.CMD_EXECUTED), ] """ ================================================================= FUNCTIONS RUN INSIDE MAIN THREAD ================================================================= """ # @override ''' #def __init__(self, name:str=None, config_filename=None, RUN_IN_THREAD=True): def __init__(self, config_filename=None): ##if name is None: name = self.__class__.__name__ #super().__init__(name, config_filename, RUN_IN_THREAD) super().__init__(config_filename) #self._log.print(f"init done for {name}") self._log.print("init done") ''' def __init__(self, name:str=None): if name is None: name = self.__class__.__name__ super().__init__() #super().__init__(RUN_IN_THREAD) # @override def init(self): #TODO: a faire super().init() # --- Set the mode according the startmode value ##agent_alias = self.__class__.__name__ ##self.set_mode_from_config(agent_alias) self.set_delay(3) # @override def main_loop_start(self): print("AgentBasic LOOP START") # @override def main_loop_end(self): if self.CURRENT_CMD and self.CURRENT_CMD.name != "get_specific_cmds": self.sleep(3) print("AgentBasic LOOP END"); # @override def routine_process_before_body(self): if self.CURRENT_CMD and self.CURRENT_CMD.name != "get_specific_cmds": self.sleep(2) # @override def routine_process_after_body(self): if self.CURRENT_CMD and self.CURRENT_CMD.name != "get_specific_cmds": self.sleep(2) ''' # @override def load_config(self): super().load_config() ''' ''' # @override def update_survey(self): super().update_survey() ''' ''' # @override def get_next_command(self): return super().get_next_command() ''' # @override def do_log(self): super().do_log() # @override def _sleep_as_soon_as_running(self): print("Make agent sleep as soon as a command has started running") print("so that we can see this command in the 'current command' column of agent web pages") if self.CURRENT_CMD and self.CURRENT_CMD.name != "get_specific_cmds": self.sleep(5) # @override def do_things_before_exit(self, stopper_agent_name=None): print("AgentBasic fait quelques trucs à lui avant de stopper...") """ ================================================================= FUNCTIONS RUN INSIDE A SUB-THREAD (OR A PROCESS) (thread_*()) ================================================================= """ # Define your own command step(s) here def cmd_step1(self, step:int): cmd = self._current_specific_cmd cmd.result = f"in step #{step}/{self._thread_total_steps_number}" cmd.save() """ if self.RUN_IN_THREAD: print("(save from thread)") cmd.save() else: #@transaction.atomic print("(save from process)") with transaction.atomic(): cmd.save() #Command.objects.select_for_update() """ def cmd_step2(self, step:int): self.cmd_step1(step) def cmd_step3(self, step:int): self.cmd_step1(step) def cmd_step4(self, step:int): self.cmd_step1(step) """ # @override def thread_exec_specific_cmd_step(self, step:int, sleep_time:float=1.0): self.thread_stop_if_asked() cmd = self._current_specific_cmd print(f">>>>> Thread (cmd {cmd.name}): step #{step}/5") self.sleep(sleep_time) """ ''' # @override def exec_specific_cmd_start(self, cmd:Command, from_thread=True): super().exec_specific_cmd_start(cmd, from_thread) ''' # @override def simulator_test_results_main(self, commands): nb_asserted = 0 for cmd in commands: if cmd.name == "flush_commands": assert cmd.is_executed() nb_asserted+=1 # 2 times if cmd.name == "go_active": assert cmd.is_executed() nb_asserted+=1 # 2 times if cmd.name == "go_idle": assert cmd.is_executed() nb_asserted+=1 """ if cmd.name == "specific0": assert cmd.is_skipped() assert cmd.result == "in step #5/5" nb_asserted+=1 """ if cmd.name == "specific1": assert cmd.is_killed() nb_asserted+=1 if cmd.name == "specific2": assert cmd.is_executed() assert cmd.result == "in step #5/5" nb_asserted+=1 if cmd.name == "eval 4+3": assert cmd.is_executed() assert cmd.get_result() == "7" nb_asserted+=1 if cmd.name in ("abort"): assert cmd.is_executed() nb_asserted+=1 if cmd.name in ("exit"): assert cmd.is_executed() nb_asserted+=1 return nb_asserted ### # ================================================================================================ # AGENT SPECIFIC COMMANDS (functions) # (here just to serve as examples) # ================================================================================================ ### #def do_specific1(self, arg1:int, arg2:int=2, arg3:int=3) -> int: #def do_specific1(self, arg1:int, arg2:int=2, arg3:float=3.1, arg4:list=[1,2,3]) -> float: def do_specific10(self, arg1:int, arg2:int, arg3:float=3.1, arg4:str='toto', arg5:Tuple[int,str,int]=(1,'toto',3), #arg5:Tuple[int,int,int]=(1,2,3), arg6:List[int]=[] ) -> float: ''' arg1 = int(arg1) arg2 = int(arg2) arg3 = float(arg3) arg5 = ast.literal_eval(arg5) print(arg5[1]) arg6 = ast.literal_eval(arg6) ''' #print(arg4) res = arg1 + arg2 + arg3 + arg5[0] + arg5[2] if arg6: res += arg6[0] return res # Undefined specific cmd #def set_specific2(self, arg1:str, arg2:int): pass def do_specific30(self): pass """ ================================================================= MAIN FUNCTION ================================================================= """ if __name__ == "__main__": agent = build_agent(AgentBasic) ''' TEST_MODE, configfile = extract_parameters() #agent = AgentX() agent = AgentA("AgentA", configfile, RUN_IN_THREAD) agent.setSimulatorMode(TEST_MODE) print(agent) ''' agent.run()