Agent : agentX working with a command simulator

Etienne Pallier
1 parent 082ccda3
Showing 5 changed files with 222 additions and 98 deletions Show diff stats
README.md
pyros.py
src/agent/Agent.py
src/agent/AgentX.py
src/common/models.py
@@ -67,14 +67,14 @@ This software has been tested and validated with the following configurations :
 --------------------------------------------------------------------------------------------
 ## LAST VERSION
  
-Date: 04/03/2019
+Date: 05/03/2019
  
 Author: E. Pallier
  
-VERSION: 0.20.5
+VERSION: 0.20.6
  
 Comment: 
-	pyros.py : enrichi commande "update" (fait aussi la mise à jour des requirements python)
+	Agent : agentX working with a command simulator
  
  
 --------------------------------------------------------------------------------------------
@@ -271,11 +271,14 @@ def install():
 '''
 TODO:
 '''
-@pyros_launcher.command(help="Update the pyros software (git pull + update DB if necessary)")
+@pyros_launcher.command(help="Update (only if necessary) the python packages AND the source code AND the DB structure")
 def update():
     print("Running update command")
-    _gitpull() or die()
+    # 1) Update python packages (pip install requirements)
     _update_python_packages_from_requirements() or die()
+    # 2) Update source code (git pull)
+    _gitpull() or die()
+    # 3) Update database structure (make migrations + migrate)
     _updatedb() or die()
     return True
  
-VERSION = "0.3"
+VERSION = "0.4"
+
+
+"""
+=================================================================
+    IMPORT PYTHON PACKAGES
+=================================================================
+"""
  
 #from __future__ import absolute_import
  
@@ -36,7 +43,7 @@ import alert_manager.tasks
 """
 # from common.models import *
 from common.models import Config, Log, PlcDeviceStatus
-from common.models import AgentSurvey, AgentCommand
+from common.models import AgentSurvey, Command
 from dashboard.views import get_sunelev
 from devices.TelescopeRemoteControlDefault import TelescopeRemoteControlDefault
  
@@ -50,29 +57,60 @@ from majordome.MajordomeDecorators import *
 from utils.JDManipulator import *
 """
  
-
 from threading import Thread
  
-DEBUG_FILE = False
-log = L.setupLogger("MajordomeTaskLogger", "Majordome")
  
-"""
-    Task to handle the execution of the program
  
-    check the environment status in database
-    check the devices status (telescope / cameras)
-    check if the last schedule made has to be planned
-    launch schedule's sequences
+
+"""
+=================================================================
+    GENERAL MODULE CONSTANT DEFINITIONS
+=================================================================
 """
  
+DEBUG_FILE = False
+
+log = L.setupLogger("AgentLogger", "Agent")
+
  
  
+
+"""
+=================================================================
+    class Agent
+=================================================================
+"""
+
 class Agent:
     # (EP) do this so that Majordome can be run from a thread, and called with thread.start():
     # class Majordome(Task, Thread):
  
-    name = "Generic Agent"
+    # FOR TEST ONLY
+    # Run this agent in simulator mode
+    SIMULATOR_MODE = True
+    SIMULATOR_COMMANDS = iter([
+        "go_active",
+        "go_idle",
+
+        "go_active",
+        "go_idle",
+
+        "go_active",
+
+        "specific_1",
+        "specific_2",
+
+        "stop"
+    ])
+    #_simulator_current_cmd_idx = 0
+    #_current_test_cmd = None
+    #_nb_test_cmds = 0
+
+
+    # Run for real, otherwise just print messages without really doing anything
     FOR_REAL = True
+
+    name = "Generic Agent"
     mainloop_waittime = 3
     subloop_waittime = 2
     status = None
@@ -94,7 +132,7 @@ class Agent:
     CONFIG_DIR = "config"
  
     _agent_survey = None
-    _agent_command = None
+
  
     def __init__(self, name:str=None, config_filename:str=None):
         self.set_mode(self.MODE_IDLE)
@@ -115,15 +153,17 @@ class Agent:
         self.config = ConfigPyros(config_filename)
         if self.config.get_last_errno() != self.config.NO_ERROR:
             raise Exception(f"Bad config file name '{config_filename}', error {str(self.config.get_last_errno())}: {str(self.config.get_last_errmsg())}")
+
+        # Create 1st survey if none
         #tmp = AgentSurvey.objects.filter(name=self.name)
         #if len(tmp) == 0:
         #nb_agents = AgentSurvey.objects.filter(name=self.name).count()
         #if nb_agents == 0:
         if not AgentSurvey.objects.filter(name=self.name).exists():
             self._agent_survey = AgentSurvey.objects.create(name=self.name, validity_duration_sec=60, mode=self.mode, status=self.status)
+            print("Agent survey is", self._agent_survey)
             #self._agent_survey = AgentSurvey(name=self.name, validity_duration_sec=60, mode=self.mode, status=self.status)
             #self._agent_survey.save()
-        print("agent is", self._agent_survey)
  
  
     def __str__(self):
@@ -170,6 +210,7 @@ class Agent:
             """
  
             print()
+            print()
             print("Starting main loop iteration...")
             self.show_mode_and_status()
  
@@ -177,12 +218,13 @@ class Agent:
  
             self.update_survey()
  
-            self.test_send_command_to_myself()
+            if self.SIMULATOR_MODE: self.simulator_send_command_to_myself()
  
             # generic cmd in json format
-            cmd = self.read_next_command()
+            print("---")
+            cmd = self.get_next_command()
  
-            self.general_process(cmd)
+            if cmd: cmd = self.general_process(cmd)
             '''
             if self.config.majordome_state == "STOP":
                 break
@@ -196,7 +238,8 @@ class Agent:
             # Sub-level loop (only if ACTIVE)
             if self.is_active():
                 self.set_status(self.STATUS_PROCESS_LOOP)
-                self.specific_process(cmd)
+                if cmd: self.specific_process(cmd)
+            print("---")
  
             self.waitfor(self.mainloop_waittime)
  
@@ -206,14 +249,6 @@ class Agent:
  
  
  
-    def general_process(self, cmd):
-        pass
-
-    # @abstract
-    # to be implemented by subclasses
-    def specific_process(self, cmd):
-        raise NotImplemented()
-
     def waitfor(self, nbsec):
             print(f"Now, waiting for {nbsec} seconds...")
             time.sleep(nbsec)
@@ -221,6 +256,7 @@ class Agent:
     def set_status(self, status:str):
         print(f"Switching from status {self.status} to status {status}")
         self.status = status
+        return False
  
     def set_mode(self, mode:str):
         print(f"Switching from mode {self.mode} to mode {mode}")
@@ -233,12 +269,11 @@ class Agent:
         self.set_mode(self.MODE_ACTIVE)
  
     def set_idle(self):
-        self.mode = self.MODE_IDLE
+        self.set_mode(self.MODE_IDLE)
  
     def show_mode_and_status(self):
         print(f"CURRENT MODE is {self.mode} (with status {self.status})")
  
-
     def die(self):
         self.set_status(self.STATUS_EXIT)
  
@@ -263,7 +298,7 @@ class Agent:
         self.set_active()
         return True
  
-    def set_mode_from_config(self,agent_alias):
+    def set_mode_from_config(self, agent_alias):
         # --- Get the startmode of the AgentX
         modestr = self.config.get_paramvalue(agent_alias,'general','startmode')
         if self.config.get_last_errno() != self.config.NO_ERROR:
@@ -314,7 +349,6 @@ class Agent:
                     namevalue = self.config.get_paramvalue(alias,unit_subtag,'name')
                     print("Unit {} alias is {}. Name is {}".format(unit_subtag,alias,namevalue))
             print("------------------------------------------")
-                
             #params = self.config.get_params(unit_alias)
             #for param in params:
             #    print("Unit component is {}".format(param))
@@ -337,52 +371,114 @@ class Agent:
             return False
         """
  
-    def update_db_survey(self):
+    def update_survey(self):
         print("Updating the survey database table...")
         self._agent_survey = AgentSurvey.objects.get(name=self.name)
         self._agent_survey.mode = self.mode
         self._agent_survey.status = self.status
         self._agent_survey.save()
  
-    def test_send_command_to_myself(self):
-        agent_command = AgentCommand.objects.create(sender=self.name, receiver=self.name, command="toto")
-        #(sender=self.name, receiver=self.name, receiver_error_code=1)
  
-    def read_next_command(self):
+    def simulator_send_command_to_myself(self):
+        #self._current_test_cmd = "go_idle" if self._current_test_cmd=="go_active" else "go_active"
+        #if self._nb_test_cmds == 4: self._current_test_cmd = "stop"
+        cmd = next(self.SIMULATOR_COMMANDS, None)
+        if not cmd: return
+        agent_command = Command.objects.create(sender=self.name, receiver=self.name, command=cmd)
+        #self._simulator_current_cmd_idx += 1
+        #self._nb_test_cmds += 1
+
+
+    def get_next_command(self)->Command:
+        """
+        Return next command (read from the DB command table) which is relevant to this agent
+        Commands are read in chronological order
+        """
+
         print("Looking for new commands from the database ...")
-        agent_commands = AgentCommand.objects.filter(receiver=self.name, receiver_status_code=AgentCommand.RECEIVER_STATUS_CODES.RSCODE_PENDING)
-        # Is there any command for me to execute ?
-        if agent_commands:
-            print(f"I have received {len(agent_commands)} new command(s) to execute")
-            for agent_command in agent_commands:
-                self.exec_cmd(agent_command)
-
-    def exec_cmd(self, ac:AgentCommand):
-        print(f"Executing command {ac.command} sent by agent {ac.sender} at {ac.sender_deposit_time}")
-        print(ac)
-        # Already read this command ?
-        if ac.receiver_read_time:
-            print("I have already read this command before, so remove it")
-            #ac.delete()
-            return
-        # Update table to say that the command has been READ
-        ac.receiver_read_time = datetime.datetime.now()
-        '''
-        if (ac.receiver_read_time - ac.sender_deposit_time) > ac.validity_duration_sec:
-            print("This command is expired, so remove it")
-            #ac.delete()
-            return
-        '''
-        # Execute this command
-        # TODO:
-        print("...command being executed...")
+
+        # 1) Is there a command currently being processed (status CMD_RUNNING) ?
+        # If so, return
+        if Command.objects.filter(
+            # only commands for me
+            receiver = self.name,
+            # only pending commands
+            receiver_status_code = Command.CMD_STATUS_CODES.CMD_RUNNING,
+        ).exists():
+            print("There is currently a running command, so I do nothing (wait for end of execution)")
+            return None
+
+        # 2) Get only the oldest PENDING command which I am recipient of
+        cmd = Command.objects.filter(
+            # only commands for me
+            receiver = self.name,
+            # only pending commands
+            receiver_status_code = Command.CMD_STATUS_CODES.CMD_PENDING,
+        ).order_by('sender_deposit_time').first()
+        #).order_by('sender_deposit_time')
+        #print("all commands", next_command)
+        #next_command = next_command.first()
+        if not cmd: return None
+        print(f"Got command {cmd.command} sent by agent {cmd.sender} at {cmd.sender_deposit_time}")
+        print(cmd)
+
+        # 3) Update read time to say that the command has been READ
+        assert cmd.receiver_read_time is None # f"Command {cmd} should not have been already read !!"
+        cmd.receiver_read_time = datetime.datetime.now()
+        cmd.save()
+        return cmd
+
+
+    def general_process(self, cmd:Command)->Command:
+
+        # Precondition : command cmd has already been read
+        assert cmd.receiver_read_time is not None # f"Command {cmd} should have been already read !!"
+
+        print(f"Starting general processing of command {cmd.command} sent by agent {cmd.sender} at {cmd.sender_deposit_time}")
+        #print(cmd)
+
+        # If expired command, change its status to expired and return
+        elapsed_time = cmd.receiver_read_time - cmd.sender_deposit_time
+        max_time = datetime.timedelta(seconds = cmd.validity_duration_sec)
+        print(f"Elapsed time is {elapsed_time}, (max is {max_time})")
+        if elapsed_time > max_time:
+            print("This command is expired, so mark it as expired, and ignore it")
+            #cmd.delete()
+            cmd.receiver_status_code = Command.CMD_STATUS_CODES.CMD_OUTOFDATE
+            cmd.save()
+            return None
+
+        # If cmd is generic, execute it, change its status to executed, and return
+        if cmd.is_generic():
+            print("This command is generic, execute it...")
+            self.exec_generic_cmd(cmd)
+            print("...Generic cmd has been executed")
+            cmd.receiver_status_code = Command.CMD_STATUS_CODES.CMD_EXECUTED
+            cmd.save()
+            if cmd.command == "stop": exit(0)
+            return None
+
+        # cmd is not generic
+        else:
+            # cmd is not generic but, as I am idle, change its status to SKIPPED, ignore it, and return
+            if self.mode == self.MODE_IDLE:
+                print("This command is not generic but, as I am IDLE, I mark it SKIPPED and ignore it")
+                cmd.receiver_status_code = Command.CMD_STATUS_CODES.CMD_SKIPPED
+                cmd.save()
+                return None
+
+        # Je suis pas idle et cde pas générique: je la traite pas, elle sera traitée par core_process :
+        # attendre que cette commande soit exécutée avant de passer à la commande suivante (situation “bloquante” normale)
+        print("This command is not generic and, as I am not IDLE, I pass it to the specific processing")
+        print("(then I will wait for this command to be EXECUTED before going to next command)")
+        return cmd
+
+
+    def exec_generic_cmd(self, cmd:Command):
         time.sleep(1)
-        print("Command is executed")
-        # Now, update table to say that the command has been EXECUTED
-        ac.receiver_error_code = 0
-        # Update table
-        ####ac.save()
-        # TODO: cleanup table from OLD lines already executed
+        if cmd.command == "go_active": self.set_active()
+        elif cmd.command == "go_idle": self.set_idle()
+        elif cmd.command == "stop": pass
  
  
     def do_log(self):
@@ -394,19 +490,16 @@ class Agent:
         print("Logging data...")
  
  
-    def core_process(self):
+    # @abstract
+    # to be implemented by subclasses
+    def specific_process(self, cmd:Command):
+        #raise NotImplemented()
         """
         Sublevel Loop (only if ACTIVE) :
         PLUS TARD, maybe :start_process_thread() dans un thread : ensuite, à chaque tour de boucle il regarde si c'est fini ou pas, et si fini recommence
         """
-        assert(self.is_active())
-
-        print("Starting the core process subloop...")
-
-        self.waitfor(self.subloop_waittime)
-
-        print("Ending core process subloop...")
-
+        assert self.is_active()
+        # TODO: LOG
  
  
  
@@ -2,6 +2,8 @@
 import utils.Logger as L
  
 from .Agent import Agent
+from common.models import Command
+
  
  
 log = L.setupLogger("AgentXTaskLogger", "AgentX")
@@ -27,23 +29,40 @@ class AgentX(Agent):
         # --- Set the mode according the startmode value
         agent_alias = self.__class__.__name__
         self.set_mode_from_config(agent_alias)
+        # TODO: remove
+        self.set_idle()
  
     # @override
+    '''
     def load_config(self):
         super().load_config()
+    '''
  
     # @override
+    '''
     def update_survey(self):
-        super().update_db_survey()
+        super().update_survey()
+    '''
  
     # @override
-    def read_next_command(self):
-        return super().read_next_command()
+    '''
+    def get_next_command(self):
+        return super().get_next_command()
+    '''
  
     # @override
     def do_log(self):
         super().do_log()
  
     # @override
-    def specific_process(self, cmd):
-        pass
 \ No newline at end of file
+    def specific_process(self, cmd:Command):
+        super().specific_process(cmd)
+
+        print("Starting specific process subloop...")
+        cmd.receiver_status_code = Command.CMD_STATUS_CODES.CMD_RUNNING
+
+        self.waitfor(self.subloop_waittime)
+
+        cmd.receiver_status_code = Command.CMD_STATUS_CODES.CMD_EXECUTED
+        cmd.save()
+        print("Ending specific process subloop...")
@@ -210,7 +210,7 @@ class Request(models.Model):
 ------------------------
 """
  
-class AgentCommand(models.Model):
+class Command(models.Model):
     """
     | id | sender | receiver | command | validity_duration_sec (default=60) | sender_deposit_time | receiver_read_time
     (l'agent destinataire en profite pour supprimer les commandes périmées qui le concernent)
@@ -223,7 +223,7 @@ class AgentCommand(models.Model):
     RSCODE_PENDING = 1 # cde en attente d'exécution
     RSCODE_SKIPPED = 2 # cde ignorée (je suis idle… et j’ai ignoré cette commande, et je passe à la cde suivante) 
     RSCODE_OUTOFDATE = 3 # cde périmée 
-    RECEIVER_STATUS_CODES = (
+    CMD_STATUS_CODES = (
         RSCODE_RUNNING, # en cours d’exécution => une fois la cde lue
         RSCODE_EXECUTED, # cde exécutée => simulé par un sleep(3) dans AgentX.core_process())
         RSCODE_PENDING, # cde en attente d'exécution
@@ -231,14 +231,16 @@ class AgentCommand(models.Model):
         RSCODE_OUTOFDATE, # cde périmée 
     )
     """
-    RECEIVER_STATUS_CODES = Choices(
-        "RSCODE_RUNNING", # en cours d’exécution => une fois la cde lue
-        "RSCODE_EXECUTED", # cde exécutée => simulé par un sleep(3) dans AgentX.core_process())
-        "RSCODE_PENDING", # cde en attente d'exécution
-        "RSCODE_SKIPPED", # cde ignorée (je suis idle… et j’ai ignoré cette commande, et je passe à la cde suivante) 
-        "RSCODE_OUTOFDATE" # cde périmée 
+    CMD_STATUS_CODES = Choices(
+        "CMD_RUNNING", # en cours d’exécution => une fois la cde lue
+        "CMD_EXECUTED", # cde exécutée => simulé par un sleep(3) dans AgentX.core_process())
+        "CMD_PENDING", # cde en attente d'exécution
+        "CMD_SKIPPED", # cde ignorée (je suis idle… et j’ai ignoré cette commande, et je passe à la cde suivante) 
+        "CMD_OUTOFDATE" # cde périmée 
     )
  
+    GENERIC_COMMANDS = ["go_idle", "go_active", "stop"]
+
     #sender = models.CharField(max_length=50, blank=True, null=True, unique=True)
     sender = models.CharField(max_length=50, help_text='sender agent name')
     receiver = models.CharField(max_length=50)
@@ -253,19 +255,26 @@ class AgentCommand(models.Model):
     receiver_read_time = models.DateTimeField(null=True)
     # - after execution
     receiver_processed_time = models.DateTimeField(null=True)
-    receiver_status_code = models.CharField(choices = RECEIVER_STATUS_CODES, default=RECEIVER_STATUS_CODES.RSCODE_PENDING, max_length=20)
-    #receiver_status_code = models.IntegerField(choices=RECEIVER_STATUS_CODES, default=RSCODE_PENDING)
+    receiver_status_code = models.CharField(choices = CMD_STATUS_CODES, default=CMD_STATUS_CODES.CMD_PENDING, max_length=20)
+    #receiver_status_code = models.IntegerField(choices=CMD_STATUS_CODES, default=RSCODE_PENDING)
     # TODO: maybe à mettre au format json (key:value)
     result = models.CharField(max_length=400, blank=True)
  
     class Meta:
         managed = True
-        db_table = 'agent_command'
+        db_table = 'command'
         #verbose_name = "agent survey"
         #verbose_name_plural = "agents survey"
  
     def __str__(self):
-        return (f"Agent {self.sender} sent commmand {self.command} to {self.receiver} at {self.sender_deposit_time}")
+        return (f"Agent {self.sender} sent commmand '{self.command}' to {self.receiver} at {self.sender_deposit_time}")
+
+    def is_generic(self):
+        """
+        Is this a generic command ?
+        It is the case if command is of style "go_idle" or "go_active" or "stop"...
+        """
+        return self.command in self.GENERIC_COMMANDS
  
  
 class AgentSurvey(models.Model):