remove version 1 and add tqdm

Author: KZHIWEI

dpdispatcher/dp_cloud_server.py

@@ -18,9 +18,9 @@ def __init__(self, context):
         self.input_data = context.remote_profile['input_data'].copy()
         self.api_version = 2
         if 'api_version' in self.input_data:
-            self.api_version = self.input_data.get('api_version')
+            self.api_version = self.input_data.get('api_version', 2)
         if 'lebesgue_version' in self.input_data:
-            self.api_version = self.input_data.get('lebesgue_version')
+            self.api_version = self.input_data.get('lebesgue_version', 2)
         self.grouped = self.input_data.get('grouped', False)
         email = context.remote_profile.get("email", None)
         username = context.remote_profile.get('username', None)
@@ -33,7 +33,7 @@ def __init__(self, context):
         if password is None:
             raise ValueError("can not find password in remote_profile, please check your machine file.")
         if self.api_version == 1:
-            warnings.warn('api version 1 is deprecated and will be removed in a future version. Use version 2 instead.', DeprecationWarning)
+            warnings.warn('api version 1 is deprecated. Use version 2 instead.', DeprecationWarning)
         self.api = API(email, password)
         self.group_id = None
 
@@ -88,26 +88,17 @@ def do_submit(self, job):
         # input_data['backward_files'] = self._gen_backward_files_list(job)
         if self.context.remote_profile.get('program_id') is None:
             warnings.warn('program_id will be compulsory in the future.')
-        job_id = None
-        if self.api_version == 2:
-            job_id, group_id = self.api.job_create_v2(
-                job_type=input_data['job_type'],
-                oss_path=input_data['job_resources'],
-                input_data=input_data,
-                program_id=self.context.remote_profile.get('program_id', None),
-                group_id=self.group_id
-            )
-            if self.grouped:
-                self.group_id = group_id
-            job.job_id = str(job_id) + ':job_group_id:' + str(group_id)
-            job_id = job.job_id
-        else:
-            job_id = self.api.job_create(
-                job_type=input_data['job_type'],
-                oss_path=input_data['job_resources'],
-                input_data=input_data,
-                program_id=self.context.remote_profile.get('program_id', None)
-            )
+        job_id, group_id = self.api.job_create(
+            job_type=input_data['job_type'],
+            oss_path=input_data['job_resources'],
+            input_data=input_data,
+            program_id=self.context.remote_profile.get('program_id', None),
+            group_id=self.group_id
+        )
+        if self.grouped:
+            self.group_id = group_id
+        job.job_id = str(job_id) + ':job_group_id:' + str(group_id)
+        job_id = job.job_id
         job.job_state = JobStatus.waiting
         return job_id
 
@@ -126,20 +117,13 @@ def check_status(self, job):
         dlog.debug(f"debug: check_status; job.job_id:{job_id}; job.job_hash:{job.job_hash}")
         check_return = None
         # print("api",self.api_version,self.input_data.get('job_group_id'),job.job_id)
-        if self.api_version == 2:
-            check_return = self.api.get_tasks_v2(job_id,group_id)
-        else:
-            check_return = self.api.get_tasks(job_id)
+        check_return = self.api.get_tasks(job_id,group_id)
         try:
             dp_job_status = check_return[0]["status"]
         except IndexError as e:
             dlog.error(f"cannot find job information in check_return. job {job.job_id}. check_return:{check_return}; retry one more time after 60 seconds")
             time.sleep(60)
-            retry_return = None
-            if self.api_version == 2:
-                retry_return = self.api.get_tasks_v2(job_id, group_id)
-            else:
-                retry_return = self.api.get_tasks(job_id)
+            retry_return = self.api.get_tasks(job_id, group_id)
             try:
                 dp_job_status = retry_return[0]["status"]
             except IndexError as e:

dpdispatcher/dp_cloud_server_context.py

@@ -13,6 +13,7 @@
 from .dpcloudserver.api import API
 from .dpcloudserver import zip_file
 import shutil
+import tqdm
 # from zip_file import zip_files
 DP_CLOUD_SERVER_HOME_DIR = os.path.join(
     os.path.expanduser('~'),
@@ -89,8 +90,8 @@ def upload(self, submission):
 
         # zip_path = "/home/felix/workplace/22_dpdispatcher/dpdispatcher-yfb/dpdispatcher/dpcloudserver/t.txt"
         # zip_path = self.local_root
-
-        for job in submission.belonging_jobs:
+        bar_format = "{l_bar}{bar}| {n:.02f}/{total:.02f} %  [{elapsed}<{remaining}, {rate_fmt}{postfix}]"
+        for job in tqdm.tqdm(submission.belonging_jobs, desc="Uploading to Lebesgue", bar_format=bar_format):
             self.machine.gen_local_script(job)
             zip_filename = job.job_hash + '.zip'
             oss_task_zip = self._gen_oss_path(job, zip_filename)
@@ -124,15 +125,12 @@ def download(self, submission):
         group_id = None
         job_infos = {}
         for job in jobs:
-            if isinstance(job.job_id, str) and ':job_group_id:' in job.job_id:
-                ids = job.job_id.split(":job_group_id:")
-                jid, gid = int(ids[0]), int(ids[1])
-                job_hashs[jid] = job.job_hash
-                group_id = gid
-            else:
-                job_infos[job.job_hash] = self.api.get_tasks(job.job_id)[0]
+            ids = job.job_id.split(":job_group_id:")
+            jid, gid = int(ids[0]), int(ids[1])
+            job_hashs[jid] = job.job_hash
+            group_id = gid
         if group_id is not None:
-            job_result = self.api.get_tasks_v2_list(group_id)
+            job_result = self.api.get_tasks_list(group_id)
             for each in job_result:
                 if 'result_url' in each and each['result_url'] != '' and each['status'] == 2:
                     job_hash = ''
@@ -143,7 +141,8 @@ def download(self, submission):
                     else:
                         job_hash = job_hashs[each['task_id']]
                     job_infos[job_hash] = each
-        for job_hash, info in job_infos.items():
+        bar_format = "{l_bar}{bar}| {n:.02f}/{total:.02f} %  [{elapsed}<{remaining}, {rate_fmt}{postfix}]"
+        for job_hash, info in tqdm.tqdm(job_infos.items(), desc="Downloading to Lebesgue", bar_format=bar_format):
             result_filename = job_hash + '_back.zip'
             target_result_zip = os.path.join(self.local_root, result_filename)
             self.api.download_from_url(info['result_url'], target_result_zip)

dpdispatcher/dpcloudserver/api.py

@@ -170,18 +170,8 @@ def upload(self, oss_task_zip, zip_task_file, endpoint, bucket_name):
         # print('debug:upload_result:', result, dir())
         return result
 
-    def job_create(self, job_type, oss_path, input_data, program_id=None):
-        post_data = {
-            'job_type': job_type,
-            'oss_path': oss_path,
-            'input_data': input_data,
-        }
-        if program_id is not None:
-            post_data["program_id"] = program_id
-        ret = self.post('/data/insert_job', post_data)
-        return ret['job_id']
 
-    def job_create_v2(self, job_type, oss_path, input_data, program_id=None, group_id=None):
+    def job_create(self, job_type, oss_path, input_data, program_id=None, group_id=None):
         post_data = {
             'job_type': job_type,
             'oss_path': oss_path,
@@ -210,17 +200,7 @@ def get_jobs(self, page=1, per_page=10):
         )
         return ret['items']
 
-    def get_tasks(self, job_id, page=1, per_page=10):
-        ret = self.get(
-            f'data/job/{job_id}/tasks',
-            {
-                'page': page,
-                'per_page': per_page,
-            }
-        )
-        return ret['items']
-
-    def get_tasks_v2(self, job_id, group_id, page=1, per_page=10):
+    def get_tasks(self, job_id, group_id, page=1, per_page=10):
         ret = self.get(
             f'data/job/{group_id}/tasks',
             {
@@ -232,10 +212,10 @@ def get_tasks_v2(self, job_id, group_id, page=1, per_page=10):
             if job_id == each["task_id"]:
                 return [each]
         if len(ret['items']) != 0:
-            return self.get_tasks_v2(job_id, group_id, page=page + 1)
+            return self.get_tasks(job_id, group_id, page=page + 1)
         return []
 
-    def get_tasks_v2_list(self, group_id, per_page=30):
+    def get_tasks_list(self, group_id, per_page=30):
         result = []
         page = 0
         while True:

requirements.txt

@@ -1,3 +1,4 @@
 paramiko
 dargs>=0.2.6
 oss2
+tqdm

setup.py

@@ -41,7 +41,7 @@
     install_requires=install_requires,    
     extras_require={
         'docs': ['sphinx', 'recommonmark', 'sphinx_rtd_theme>=1.0.0rc1', 'numpydoc'],
-        "cloudserver": ["oss2"],
+        "cloudserver": ["oss2", "tqdm"],
         ":python_version<'3.7'": ["typing_extensions"],
     },
         entry_points={