Bye Bye Celery (1 - from pyros.py)

Etienne Pallier
1 parent ddc7106a
Showing 1 changed file with 32 additions and 150 deletions Show diff stats
pyros.py
@@ -9,10 +9,6 @@ import argparse
 import time
 import signal
-# For USE_CELERY global variable
-#from src.pyros import settings
-USE_CELERY = False
-#USE_CELERY = True
 DEBUG = True
 SIMULATOR_CONFIG_FILE = "conf.json" #default config file
@@ -54,28 +50,27 @@ class Utils:
             print("Current directory : " + str(os.getcwd()))
         return 0
-    def replacePatternInFile(self, pattern, replace, file_path):
+    def replacePatternInFile(self, pattern, replacement, file_path):
         #try:
         with fileinput.FileInput(file_path, inplace=True, backup='.bak') as file:
             for line in file:
-                print(line.replace(pattern, replace), end='')
+                print(line.replace(pattern, replacement), end='')
         '''
         except:
             return 1
         return 0
         '''
         # Now, check that replacement has been done or else ERROR !!!
-        #pattern = "CELERY_TEST = "+str(set_it)
         FOUND=False
         #with fileinput.FileInput(file_path) as file:
         with open(file_path) as file:
             for line in file:
-                #if replace in line:
-                if line.startswith(replace):
+                #if replacement in line:
+                if line.startswith(replacement):
                     FOUND = True
                     break
         #if FOUND: print("pattern "+pattern+" found in file "+file_path)
-        if not FOUND: raise(Exception("pattern "+replace+" not found in file "+file_path))
+        if not FOUND: raise(Exception("pattern "+replacement+" not found in file "+file_path))
     def printColor(self, color, message, file=sys.stdout, eol=os.linesep, forced=False):
@@ -113,7 +108,6 @@ class AManager(Utils):
     python_version = sys.version_info
     bin_dir = ""
-    celery = "celery"
     venv_pip = "pip"
     venv_bin = "python"
     wait = True
@@ -141,34 +135,29 @@ class AManager(Utils):
             self.bin_dir = "Scripts"
             self.bin_name = "python.exe"
             self.pip_name = "pip.exe"
-            self.celery = "celery.exe"
         else:
             self.bin_dir = "bin"
             self.bin_name = "python"
             self.pip_name = "pip"
-            self.celery = "celery"
         self.venv_pip = self.path_dir_file + os.sep + config["path"] + os.sep + config["env"] + os.sep + self.bin_dir + os.sep + self.pip_name
         self.venv_bin = self.path_dir_file + os.sep + config["path"] + os.sep + config["env"] + os.sep + self.bin_dir + os.sep + self.bin_name
-        self.venv_cel = self.path_dir_file + os.sep + config["path"] + os.sep + config["env"] + os.sep + self.bin_dir + os.sep + self.celery
     def help(self):
         print("This function must be implemented")
         raise(NotImplementedError("Function not implemented"))
-    def set_celery_test_and_simulator_modes_to(self, set_it:bool, simulator:bool=True):
+    def set_simulator_mode_to(self, set_it:bool):
         file_path = "pyros/settings.py"
         '''
-        self.replacePatternInFile("CELERY_TEST = False", "CELERY_TEST = True", file_path)
         if simulator: self.replacePatternInFile("SIMULATOR = False", "SIMULATOR = True", file_path)
         '''
-        self.replacePatternInFile("CELERY_TEST = "+str(not set_it), "CELERY_TEST = "+str(set_it), file_path)
-        if simulator: self.replacePatternInFile("SIMULATOR = "+str(not set_it), "SIMULATOR = "+str(set_it), file_path)
+        self.replacePatternInFile("SIMULATOR = "+str(not set_it), "SIMULATOR = "+str(set_it), file_path)
     def signal_handler(self, signal, frame):
         self.printFullTerm(Colors.WARNING, "Ctrl-c catched")
-        self.set_celery_test_and_simulator_modes_to(False)
+        self.set_simulator_mode_to(False)
         for p in self.subproc:
             proc, name = p
             self.printColor(Colors.BLUE, "Killing process " + str(name))
@@ -448,12 +437,10 @@ class Pyros(AManager):
         return self.execProcess("rm logs/*.log")
     def test(self):
-        ##self.singleWorker("scheduling")
         self.changeDirectory("src")
         ##self.execProcessFromVenv(self.venv_bin + " manage.py test")
         self.execProcessFromVenvAsync(self.venv_bin + " manage.py test")
         self.changeDirectory("..")
-        ##self.stop_workers()
         return 0
     def migrate(self):
@@ -487,7 +474,7 @@ class Pyros(AManager):
     def reset_config(self):
         self.changeDirectory("src")
-        self.set_celery_test_and_simulator_modes_to(False)
+        self.set_simulator_mode_to(False)
         self.addExecuted(self.current_command, "reset configuration")
         self.changeDirectory("..")
         return 0
@@ -514,50 +501,6 @@ class Pyros(AManager):
         return 0
-    def celery_on(self, TOTAL=True):
-        self.changeDirectory("src")
-
-        if TOTAL: self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q alert_listener_q -n pyros@alert_listener -c 1")
-        if TOTAL: self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q monitoring_q -n pyros@monitoring -c 1")
-
-        #self.singleWorker("majordome")
-        self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q majordome_q -n pyros@majordome -c 1")
-
-        #self.singleWorker("scheduling")
-        self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q scheduling_q --purge -n pyros@scheduling -c 1")
-        
-        #self.singleWorker("execute_plan_vis")
-        self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q execute_plan_vis_q --purge -n pyros@execute_plan_vis -c 1")
-
-        if TOTAL: self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q night_calibrations_q --purge -n pyros@night_calibrations -c 1")
-        
-        #self.singleWorker("execute_plan_nir")
-        self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q execute_plan_nir_q --purge -n pyros@execute_plan_nir -c 1")
-        
-        #self.singleWorker("create_calibrations")
-        self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q create_calibrations_q --purge -n pyros@create_calibrations -c 1")
-
-        #self.singleWorker("analysis")
-        self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q analysis_q --purge -n pyros@analysis -c 1")
-        
-        self.changeDirectory("..")
-        return 0
-
-    # ex start()
-    def start_workers(self):
-        self.stop_workers()
-        self.celery_on()
-        return 0
-
-    # ex stop()
-    def stop_workers(self):
-        if (self.system == "Windows"):
-            self.execProcessAsync("taskkill /f /im celery.exe")
-            self.execProcessAsync("taskkill /f /im python.exe")
-        else:
-            self.execProcessAsync("ps aux | grep \"celery worker\" | awk '{print $2}' | xargs kill -9")
-        return 0
-
     def init_database(self):
         self.makemigrations()
         self.migrate()
@@ -596,15 +539,6 @@ class Pyros(AManager):
         what="all"
         '''
-        ''' OLD CODE
-        self.stop_workers()
-        #self.celery_on()
-        # Start the 3 main agents (via their celery worker) :
-        self.startAgent("monitoring")
-        self.startAgent("majordome")
-        self.startAgent("alert_listener")
-        '''
-
         # Go into src/
         self.changeDirectory("src")
         #print("Current directory : " + str(os.getcwd()))
@@ -655,17 +589,10 @@ class Pyros(AManager):
-    def startAgent(self, agent_name:str):
-        # Start Celery Worker
-        self.singleWorker(agent_name)
-        time.sleep(3)
-        # Put the agent task run() into the Worker queue
-
-        
     # Reset the database content
     def reset_database_sim(self):
         self.changeDirectory("src")
-        self.set_celery_test_and_simulator_modes_to(True, False)
+        self.set_simulator_mode_to(True, False)
         '''
         Supprime toutes les données de la base de données 
         et réexécute tout gestionnaire de post-synchronisation. 
@@ -673,11 +600,11 @@ class Pyros(AManager):
         '''
         #TODO: remplacer par manage.py --noinput flush pour eviter le "echo yes"...
         self.execProcess( self.venv_bin + " manage.py flush  --noinput")
-        self.set_celery_test_and_simulator_modes_to(True, False)
+        self.set_simulator_mode_to(True, False)
         self.changeDirectory("..")
-    # Simulation for the scheduler ONLY (and only some celery workers, not all of them)
+    # Simulation for the scheduler ONLY
     def simulator_development(self):
         self.simulator(False)
@@ -691,12 +618,11 @@ class Pyros(AManager):
     delete from request;
     '''
-    # Simulation (by default, with ALL simulators and ALL celery workers)
+    # Simulation (by default, with ALL simulators)
     def simulator(self, TOTAL=True):
         self.changeDirectory("src")
-        # Set CELERY_TEST mode ON
-        self.replacePatternInFile("CELERY_TEST = False", "CELERY_TEST = True", "pyros/settings.py")
+        # Set SIMULATOR mode ON
         self.replacePatternInFile("SIMULATOR = False", "SIMULATOR = True", "pyros/settings.py")
         # 0) Reset the database (Empty the database from any data)
@@ -714,7 +640,7 @@ class Pyros(AManager):
         self.loaddata()
         #
-        # 1) Launch Django web server
+        # 1) Launch web server
         #
         self.server()
         self.sleep(2)
@@ -727,18 +653,11 @@ class Pyros(AManager):
         self.printColor(Colors.GREEN, "If you want to shutdown the simulation, please run :")
         self.printColor(Colors.GREEN, "CTRL-C or pyros.py kill_simulation")
         self.printColor(Colors.GREEN, "If the simulation isn't correctly killed, please switch the variable")
-        self.printColor(Colors.GREEN, "CELERY_TEST in src/pyros/settings.py to False")
+        self.printColor(Colors.GREEN, "SIMULATOR in src/pyros/settings.py to False")
         self.printFullTerm(Colors.WARNING, "SUMMARY")
         #
-        # 2) (if USE_CELERY) Start Celery workers
-        #
-        if USE_CELERY:
-            self.celery_on(TOTAL)
-            self.sleep(3)
-
-        #
-        # 3) Start simulator(s) :
+        # 2) Start simulator(s) :
         #
         self.sims_launch(TOTAL)
@@ -774,20 +693,6 @@ class Pyros(AManager):
             self.changeDirectory("..")
         '''
-        '''
-        #
-        # 2) (if not USE_CELERY) Start Agents
-        #
-        if TOTAL and not USE_CELERY :
-            #pass
-            # Start Environment Monitoring Agent
-            # Start Majordome Agent
-            # Start Alert Manager Agent
-            self.start_agents()
-            # Needed ?
-            self.sleep(3)
-        '''
-
         # When simulators are finished:
         #self.kill_simulation()
         return 0
@@ -799,8 +704,8 @@ class Pyros(AManager):
     def test_majordome(self):
         self.changeDirectory("src")
-        # Set CELERY_TEST mode ON
-        self.set_celery_test_and_simulator_modes_to(True)
+        # Set SIMULATOR mode ON
+        self.set_simulator_mode_to(True)
         # 0) Reset the database (Empty the database from any data)
         self.changeDirectory("..")
@@ -824,19 +729,11 @@ class Pyros(AManager):
         self.printColor(Colors.GREEN, "If you want to shutdown the simulation, please run :")
         self.printColor(Colors.GREEN, "CTRL-C or pyros.py kill_simulation")
         self.printColor(Colors.GREEN, "If the simulation isn't correctly killed, please switch the variable")
-        self.printColor(Colors.GREEN, "CELERY_TEST in src/pyros/settings.py to False")
+        self.printColor(Colors.GREEN, "SIMULATOR in src/pyros/settings.py to False")
         self.printFullTerm(Colors.WARNING, "SUMMARY")
-        # 2) (if USE_CELERY) Start Celery workers
         #
-        '''
-        if USE_CELERY:
-            self.celery_on(TOTAL)
-            self.sleep(3)
-        '''
-
-        #
-        # 3) Start simulator(s) :
+        # 2) Start simulator(s) :
         #
         #self.sims_launch(True)
         agent='majordome'
@@ -852,7 +749,7 @@ class Pyros(AManager):
         # When simulators are finished:
         #self.kill_simulation()
         #self.changeDirectory("src")
-        self.set_celery_test_and_simulator_modes_to(False)
+        self.set_simulator_mode_to(False)
         self.changeDirectory("..")
         self.execProcessAsync("ps aux | grep \"start_agent_majordome.py\" | awk '{ print $2 }' | xargs kill")
@@ -870,7 +767,7 @@ class Pyros(AManager):
     def kill_simulation(self):
         self.changeDirectory("src")
-        self.set_celery_test_and_simulator_modes_to(False)
+        self.set_simulator_mode_to(False)
         self.changeDirectory("..")
         if (self.system == "Windows"):
             self.execProcessAsync("taskkill /f /im python.exe")
@@ -884,12 +781,12 @@ class Pyros(AManager):
         self.kill_server()
-        # (if not using celery) Kill all agents :
-        if not USE_CELERY:
-            self.execProcessAsync("ps aux | grep \"start_agent_alert_manager.py\" | awk '{ print $2 }' | xargs kill")
-            self.execProcessAsync("ps aux | grep \"start_agent_monitoring.py\" | awk '{ print $2 }' | xargs kill")
-            self.execProcessAsync("ps aux | grep \"start_agent_majordome.py\" | awk '{ print $2 }' | xargs kill")
+        # Kill all agents :
+        self.execProcessAsync("ps aux | grep \"start_agent_alert_manager.py\" | awk '{ print $2 }' | xargs kill")
+        self.execProcessAsync("ps aux | grep \"start_agent_monitoring.py\" | awk '{ print $2 }' | xargs kill")
+        self.execProcessAsync("ps aux | grep \"start_agent_majordome.py\" | awk '{ print $2 }' | xargs kill")
         #self.execProcessAsync("ps aux | grep \"start_agent_majordome.py\" | awk '{ print $2 }' | xargs kill")
+
         # Kill all simulators :
         self.execProcessAsync("ps aux | grep \" domeSimulator.py\" | awk '{ print $2 }' | xargs kill")
         self.execProcessAsync("ps aux | grep \" userSimulator.py\" | awk '{ print $2 }' | xargs kill")
@@ -899,7 +796,6 @@ class Pyros(AManager):
         self.execProcessAsync("ps aux | grep \" cameraNIRSimulator.py\" | awk '{ print $2 }' | xargs kill")
         self.execProcessAsync("ps aux | grep \" cameraVISSimulator.py\" | awk '{ print $2 }' | xargs kill")
         self.changeDirectory("..")
-        if USE_CELERY: self.stop_workers()
         self.printFullTerm(Colors.GREEN, "simulation ended")
         return 0
@@ -1001,8 +897,8 @@ class Pyros(AManager):
         # Get back to project root folder
         self.changeDirectory("..")
-        # (if not celery) Launch agents (env monitor, major, alert mgr)
-        if not USE_CELERY: self.start_agents()
+        # Launch agents (env monitor, major, alert mgr)
+        self.start_agents()
         # Wait for end of simulators :
         for p in procs: p.wait()
@@ -1012,12 +908,6 @@ class Pyros(AManager):
         return 0
-    def singleWorker(self, worker):
-        self.changeDirectory("src")
-        self.execProcessFromVenvAsync(self.venv_cel + " worker -A pyros -Q "+ worker +"_q -n pyros@"+worker+" -c 1")
-        self.changeDirectory("..")
-        return 0
-
     def mysql_on(self):
         self.changeDirectory("src")
         self.replacePatternInFile("MYSQL = False", "MYSQL = True", "pyros/settings.py")
@@ -1050,7 +940,6 @@ class Pyros(AManager):
             "unittest": self.unittest,
             "reset_config": self.reset_config,
             "test_all": self.test_all,
-            "celery_on": self.celery_on,
             "reset_database_sim": self.reset_database_sim,
             "init_database": self.init_database,
             "kill_server": self.kill_server,
@@ -1065,10 +954,6 @@ class Pyros(AManager):
             "start_agent_alertmanager": self.start_agent_alertmanager,
             "test_majordome": self.test_majordome,
-            # ex start
-            "start_workers": self.start_workers,
-            # ex stop
-            "stop_workers": self.stop_workers,
             "simulator": self.simulator,
             "kill_simulation": self.kill_simulation,
             "sims_launch": self.sims_launch,
@@ -1092,14 +977,11 @@ class Pyros(AManager):
             "updatedb": "Update the database",
             "kill_server": "Kill the web server on port 8000",
             "init_database": "Create a standard context for pyros in db",
-            "unittest": "Runs the tests that don't need celery",
+            "unittest": "Runs the unit tests",
             "test_all": "Run all the existing tests (this command needs to be updated when tests are added in the project",
-            "celery_on": "Starts celery workers",
-            "start_workers": "Starts the celery workers (then the web server ???)",
-            "stop_workers": "Stops the celery workers",
             "simulator": "Launch a simulation",
             "simulator_development": "Simulation for the scheduler only",
-            "kill_simulation": "kill the simulators / celery workers / web server",
+            "kill_simulation": "kill the simulators & web server",
             "sims_launch": "Launch only the simulators",
         }