make scanning process multithreaded

Author: deckbsd

glouton/commands/module/observationModuleCommand.py

@@ -1,8 +1,10 @@
 from glouton.commands.module.moduleCommand import ModuleCommand
 
+
 class ObservationModuleCommand(ModuleCommand):
     def __init__(self, params):
         ModuleCommand.__init__(self, params)
 
     def process(self):
-        self.params.module.runAfterDownload(self.params.file_name, self.params.full_path, self.params.observation)
+        self.params.module.runAfterDownload(
+            self.params.file_name, self.params.full_path, self.params.observation)

glouton/commands/module/observationModuleCommandParams.py

@@ -1,5 +1,6 @@
 from glouton.commands.module.moduleCommandParams import ModuleCommandParams
 
+
 class ObservationModuleCommandParams(ModuleCommandParams):
     def __init__(self, file_name=None, full_path=None, observation=None, module=None):
         ModuleCommandParams.__init__(self, file_name, full_path, module)

glouton/commands/module/telemetryModuleCommand.py

@@ -1,8 +1,10 @@
 from glouton.commands.module.moduleCommand import ModuleCommand
 
+
 class TelemetryModuleCommand(ModuleCommand):
     def __init__(self, params):
         ModuleCommand.__init__(self, params)
 
     def process(self):
-        self.params.module.runAfterDownload(self.params.file_name, self.params.full_path, self.params.telemetry)
+        self.params.module.runAfterDownload(
+            self.params.file_name, self.params.full_path, self.params.telemetry)

glouton/commands/module/telemetryModuleCommandParams.py

@@ -1,5 +1,6 @@
 from glouton.commands.module.moduleCommandParams import ModuleCommandParams
 
+
 class TelemetryModuleCommandParams(ModuleCommandParams):
     def __init__(self, file_name=None, full_path=None, telemetry=None, module=None):
         ModuleCommandParams.__init__(self, file_name, full_path, module)

glouton/infrastructure/satnogClient.py

@@ -5,7 +5,7 @@
 
 class SatnogClient:
     def __init__(self):
-        self.config  = config.read()
+        self.config = config.read()
         self.proxies = self._set_proxy()
 
     def _set_proxy(self):

glouton/infrastructure/satnogDbClient.py

@@ -12,4 +12,4 @@ def get(self, url, params=None):
         return requests.get(url, params=params, proxies=self.proxies)
 
     def get_from_base(self, url, params=None):
-        return requests.get(self._url + url, params=params, proxies=self.proxies)
+        return requests.get(self._url + url, params=params, proxies=self.proxies)

glouton/infrastructure/satnogNetworkClient.py

@@ -12,4 +12,4 @@ def get(self, url, params=None):
         return requests.get(url, params=params, proxies=self.proxies)
 
     def get_from_base(self, url, params=None):
-        return requests.get(self._url + url, params=params, proxies=self.proxies)
+        return requests.get(self._url + url, params=params, proxies=self.proxies)

glouton/repositories/demoddata/demoddataRepo.py

@@ -8,6 +8,7 @@
 from glouton.shared import threadHelper
 from threading import Event
 
+
 class DemoddataRepo(Downloadable):
     def __init__(self, working_dir, modules):
         self.__working_dir = working_dir
@@ -25,13 +26,15 @@ def register_command(self, observation, start_date, end_date):
 
     def create_worker(self):
         threads = []
-        downloadWorker = DownloadWorker(self.__demoddata_commands, self.__download_status)
+        downloadWorker = DownloadWorker(
+            self.__demoddata_commands, self.__download_status)
         threads.append(threadHelper.create_thread(downloadWorker.execute))
         if self.__modules is not None:
-            moduleWorker = ModuleWorker(self.__demoddate_modules_commands, self.__download_status)
+            moduleWorker = ModuleWorker(
+                self.__demoddate_modules_commands, self.__download_status)
             threads.append(threadHelper.create_thread(moduleWorker.execute))
 
-        return threads 
+        return threads
 
     def __create_dir_name(self, target, start_date, end_date):
         return target + '__' + start_date.strftime('%m-%d-%YT%H-%M-%S') + '__' + end_date.strftime('%m-%d-%YT%H-%M-%S')

glouton/repositories/observation/observationsRepo.py

@@ -1,63 +1,60 @@
 from queue import Queue
-from glouton.infrastructure.satnogNetworkClient import SatnogNetworkClient
 from glouton.shared.logger import logger
+from glouton.shared import dateHelper
+from glouton.workers.pageScanWorker import PageScanWorker
+from glouton.shared import threadHelper
+from glouton.infrastructure.satnogNetworkClient import SatnogNetworkClient
 
 
 class ObservationRepo:
     def __init__(self, cmd, repos):
         self.OBSERVATION_URL = 'observations/'
-        self.__client = SatnogNetworkClient()
         self.__repos = repos
         self.__cmd = cmd
         self.__threads = []
 
     def extract(self):
-        params = self.__create_request_params()
-        page = 1
-        while True:
-            r = self.__client.get_from_base(
-                self.OBSERVATION_URL, params)
-            if r.status_code != 200:
-                break
-
-            logger.Info('scanning page...' + params['page'])
-            self.__read_page(r.json(), self.__cmd.start_date, self.__cmd.end_date)
-            page += 1
-            params['page'] = str(page)
-
+        client = SatnogNetworkClient()
+        diff_days = dateHelper.diff_days(
+            self.__cmd.start_date, self.__cmd.end_date)
+        if diff_days < 8:
+            # no thread needed
+            url_params = self.__url_param_builder(
+                self.__cmd.start_date, self.__cmd.end_date)
+            pageScanner = PageScanWorker(
+                client, self.__cmd, self.__repos, self.OBSERVATION_URL, url_params, 1)
+            pageScanner.scan()
+        else:
+            threads = []
+            job = 1
+            for from_datetime, to_datetime in dateHelper.split_date(self.__cmd.start_date, self.__cmd.end_date, 4):
+                print(str(from_datetime) + " " + str(to_datetime))
+                url_params = self.__url_param_builder(
+                    from_datetime, to_datetime)
+                pageScanner = PageScanWorker(
+                    client, self.__cmd, self.__repos, self.OBSERVATION_URL, url_params, job)
+                t = threadHelper.create_thread(pageScanner.scan)
+                threads.append(t)
+                job += 1
+
+            threadHelper.wait(threads)
         print('\ndownloading started (Ctrl + C to stop)...\t~(  ^o^)~')
         self.__create_workers_and_wait()
 
     def __create_workers_and_wait(self):
         for repo in self.__repos:
             self.__threads.extend(repo.create_worker())
-        while self.__is_one_thread_alive():
-            for t in self.__threads:
-                # let's control to main thread every seconds (in order to be able to capture Ctrl + C if needed)
-                t.join(1)
-
-    def __is_one_thread_alive(self):
-        for thread in self.__threads:
-            if thread.is_alive():
-                return True
-
-        return False
-
-    def __read_page(self, observations, start_date, end_date):
-        for observation in observations:
-            for repo in self.__repos:
-                repo.register_command(
-                    observation, start_date, end_date)
+        threadHelper.wait(self.__threads)
 
-    def __create_request_params(self):
+    def __url_param_builder(self, start_date, end_date):
         return {'satellite__norad_cat_id': self.__cmd.norad_id,
-        'ground_station': self.__cmd.ground_station_id,
-        'start': self.__cmd.start_date.isoformat(),
-        'end': self.__cmd.end_date.isoformat(),
-        'vetted_status': self.__cmd.observation_status,
-        'vetted_user': self.__cmd.user,
-        'transmitter_uuid': self.__cmd.transmitter_uuid,
-        'transmitter_mode': self.__cmd.transmitter_mode,
-        'transmitter_type': self.__cmd.transmitter_type,
-        'page': '1',
-        'format': 'json'}
+                'ground_station': self.__cmd.ground_station_id,
+                'start': start_date.isoformat(),
+                'end': end_date.isoformat(),
+                'vetted_status': self.__cmd.observation_status,
+                'vetted_user': self.__cmd.user,
+                'transmitter_uuid': self.__cmd.transmitter_uuid,
+                'transmitter_mode': self.__cmd.transmitter_mode,
+                'transmitter_type': self.__cmd.transmitter_type,
+                'page': '1',
+                'format': 'json'}

glouton/repositories/payload/payloadRepo.py

@@ -8,6 +8,7 @@
 from glouton.shared import threadHelper
 from threading import Event
 
+
 class PayloadRepo(Downloadable):
     def __init__(self, working_dir, modules):
         self.__working_dir = working_dir
@@ -25,10 +26,12 @@ def register_command(self, observation, start_date, end_date):
 
     def create_worker(self):
         threads = []
-        downloadWorker = DownloadWorker(self.__payload_commands, self.__download_status)
+        downloadWorker = DownloadWorker(
+            self.__payload_commands, self.__download_status)
         threads.append(threadHelper.create_thread(downloadWorker.execute))
         if self.__modules is not None:
-            moduleWorker = ModuleWorker(self.__payload_modules_commands, self.__download_status)
+            moduleWorker = ModuleWorker(
+                self.__payload_modules_commands, self.__download_status)
             threads.append(threadHelper.create_thread(moduleWorker.execute))
 
         return threads

glouton/repositories/telemetry/telemetryRepo.py

@@ -1,60 +1,55 @@
 from queue import Queue
+from glouton.shared import dateHelper
+from glouton.workers.pageScanWorker import PageScanWorker
 from glouton.infrastructure.satnogDbClient import SatnogDbClient
+from glouton.shared import threadHelper
 from glouton.shared.logger import logger
 
 
 class TelemetryRepo:
     def __init__(self, cmd, repos):
         self.TELEMETRY_URL = 'telemetry/'
-        self.__client = SatnogDbClient()
         self.__repos = repos
         self.__cmd = cmd
         self.__threads = []
 
     def extract(self):
-        params = self.__create_request_params()
-        page = 1
-        while True:
-            r = self.__client.get_from_base(
-                self.TELEMETRY_URL, params)
-            if r.status_code != 200:
-                break
-
-            logger.Info('scanning page...' + params['page'])
-            self.__read_page(r.json(), self.__cmd.start_date,
-                             self.__cmd.end_date)
-            page += 1
-
-            params['page'] = str(page)
-
+        client = SatnogDbClient()
+        diff_days = dateHelper.diff_days(
+            self.__cmd.start_date, self.__cmd.end_date)
+        if diff_days < 8:
+            # no thread needed
+            url_params = self.__url_param_builder(
+                self.__cmd.start_date, self.__cmd.end_date)
+            pageScanner = PageScanWorker(
+                client, self.__cmd, self.__repos, self.TELEMETRY_URL, url_params, 1)
+            pageScanner.scan()
+        else:
+            threads = []
+            job = 1
+            for from_datetime, to_datetime in dateHelper.split_date(self.__cmd.start_date, self.__cmd.end_date, 4):
+                print(str(from_datetime) + " " + str(to_datetime))
+                url_params = self.__url_param_builder(
+                    from_datetime, to_datetime)
+                pageScanner = PageScanWorker(
+                    client, self.__cmd, self.__repos, self.TELEMETRY_URL, url_params, job)
+                t = threadHelper.create_thread(pageScanner.scan)
+                threads.append(t)
+                job += 1
+
+            threadHelper.wait(threads)
         print('\ndownloading started (Ctrl + C to stop)...\t~(  ^o^)~')
         self.__create_workers_and_wait()
 
     def __create_workers_and_wait(self):
         for repo in self.__repos:
             self.__threads.extend(repo.create_worker())
-        while self.__is_one_thread_alive():
-            for t in self.__threads:
-                # let's control to main thread every seconds (in order to be able to capture Ctrl + C if needed)
-                t.join(1)
-
-    def __is_one_thread_alive(self):
-        for thread in self.__threads:
-            if thread.is_alive():
-                return True
-
-        return False
-
-    def __read_page(self, telemetries, start_date, end_date):
-        for telemetry in telemetries:
-            for repo in self.__repos:
-                repo.register_command(
-                    telemetry, start_date, end_date)
+        threadHelper.wait(self.__threads)
 
-    def __create_request_params(self):
+    def __url_param_builder(self, start_date, end_date):
         return {'satellite': self.__cmd.norad_id,
-                'start': self.__cmd.start_date.isoformat(),
-                'end': self.__cmd.end_date.isoformat(),
+                'start': start_date.isoformat(),
+                'end': end_date.isoformat(),
                 'observer': self.__cmd.observer,
                 'transmitter': self.__cmd.transmitter,
                 'app_source': self.__cmd.app_source,

glouton/repositories/waterfall/waterfallRepo.py

@@ -8,6 +8,7 @@
 from glouton.shared import threadHelper
 from threading import Event
 
+
 class WaterfallRepo(Downloadable):
     def __init__(self, working_dir, modules):
         self.__working_dir = working_dir
@@ -25,10 +26,12 @@ def register_command(self, observation, start_date, end_date):
 
     def create_worker(self):
         threads = []
-        downloadWorker = DownloadWorker(self.__waterfall_commands, self.__download_status)
+        downloadWorker = DownloadWorker(
+            self.__waterfall_commands, self.__download_status)
         threads.append(threadHelper.create_thread(downloadWorker.execute))
         if self.__modules is not None:
-            moduleWorker = ModuleWorker(self.__waterfall_modules_commands, self.__download_status)
+            moduleWorker = ModuleWorker(
+                self.__waterfall_modules_commands, self.__download_status)
             threads.append(threadHelper.create_thread(moduleWorker.execute))
 
         return threads