Thu, 05 Oct 2023 15:41:41 +0200
Download of products for automatic upload to EARLINET db implemented
ioannis@67 | 1 | """ This is a script that allows interaction with the SCC through the command line. |
ioannis@67 | 2 | |
ioannis@67 | 3 | It is based on the requests module for accessing the server. |
ioannis@67 | 4 | |
ioannis@67 | 5 | Most of the interactions are done through the web interface, i.e. by mimicking user interaction with the |
ioannis@67 | 6 | SCC website (i.e. user login, data submission, etc.). In few cases, the SCC API is also used. |
ioannis@67 | 7 | |
ioannis@67 | 8 | Most of the functionality is included in the SCC class. The class is used to login into the SCC website and automate |
ioannis@67 | 9 | interaction with the site (i.e. upload a file, get measurement status, etc.). |
ioannis@67 | 10 | |
ioannis@67 | 11 | Two other classes (Measurement, AncillaryFile) are used in some cases to handle the output of the SCC API. |
ioannis@67 | 12 | |
ioannis@67 | 13 | Several shortcut functions are defined to perform specific tasks using the SCC class (e.g. process_file, delete_measurements etc). |
ioannis@67 | 14 | """ |
ioannis@43 | 15 | import sys |
ioannis@43 | 16 | |
claudio@71 | 17 | import mysql.connector |
victor@7 | 18 | import requests |
victor@7 | 19 | |
ioannis@43 | 20 | try: |
ioannis@43 | 21 | import urllib.parse as urlparse # Python 3 |
ioannis@43 | 22 | except ImportError: |
ioannis@17 | 23 | import urlparse # Python 2 |
ioannis@43 | 24 | |
victor@7 | 25 | import argparse |
madrouin@24 | 26 | import datetime |
madrouin@24 | 27 | import logging |
victor@7 | 28 | import os |
victor@7 | 29 | import re |
ioannis@32 | 30 | from io import BytesIO |
ioannis@45 | 31 | |
victor@7 | 32 | import time |
i@40 | 33 | |
victor@7 | 34 | from zipfile import ZipFile |
madrouin@24 | 35 | |
i@14 | 36 | import yaml |
victor@7 | 37 | |
ioannis@43 | 38 | import netCDF4 as netcdf |
ioannis@43 | 39 | |
ioannis@43 | 40 | requests.packages.urllib3.disable_warnings() |
i@14 | 41 | logger = logging.getLogger(__name__) |
victor@7 | 42 | |
victor@7 | 43 | # The regex to find the measurement id from the measurement page |
victor@7 | 44 | # This should be read from the uploaded file, but would require an extra NetCDF module. |
i@36 | 45 | regex = "<h3>Measurement (?P<measurement_id>.{12,15}) <small>" # {12, 15} to handle both old- and new-style measurement ids. |
victor@7 | 46 | |
victor@7 | 47 | |
victor@7 | 48 | class SCC: |
ioannis@67 | 49 | """A class that will attempt to interact SCC server. |
i@14 | 50 | |
ioannis@67 | 51 | Most interactions are by simulating a normal browser session. In the current |
ioannis@67 | 52 | version few checks are performed before upload a file, and no feedback is given in case the upload |
ioannis@67 | 53 | fails. |
victor@7 | 54 | """ |
victor@7 | 55 | |
claudio@71 | 56 | def __init__(self, auth, output_dir, base_url, sccdb_host=None, sccdb_credentials=None, sccdb=None, sccquery=None): |
moritz@29 | 57 | |
victor@7 | 58 | self.auth = auth |
victor@7 | 59 | self.output_dir = output_dir |
i@14 | 60 | self.base_url = base_url |
claudio@71 | 61 | self.sccdb_host = sccdb_host |
claudio@71 | 62 | self.sccdb_credentials = sccdb_credentials |
claudio@71 | 63 | self.sccdb = sccdb |
claudio@71 | 64 | self.sccquery = sccquery |
victor@7 | 65 | self.session = requests.Session() |
moritz@29 | 66 | self.session.auth = auth |
moritz@29 | 67 | self.session.verify = False |
victor@7 | 68 | |
ioannis@67 | 69 | # Setup SCC server URLS for later use |
i@14 | 70 | self.login_url = urlparse.urljoin(self.base_url, 'accounts/login/') |
i@31 | 71 | self.logout_url = urlparse.urljoin(self.base_url, 'accounts/logout/') |
i@31 | 72 | self.list_measurements_url = urlparse.urljoin(self.base_url, 'data_processing/measurements/') |
i@31 | 73 | |
i@14 | 74 | self.upload_url = urlparse.urljoin(self.base_url, 'data_processing/measurements/quick/') |
ioannis@54 | 75 | self.measurement_page_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/') |
ioannis@43 | 76 | self.download_hirelpp_pattern = urlparse.urljoin(self.base_url, |
ioannis@67 | 77 | 'data_processing/measurements/{0}/download-hirelpp/') |
ioannis@43 | 78 | self.download_cloudmask_pattern = urlparse.urljoin(self.base_url, |
ioannis@67 | 79 | 'data_processing/measurements/{0}/download-cloudmask/') |
ioannis@34 | 80 | |
ioannis@43 | 81 | self.download_elpp_pattern = urlparse.urljoin(self.base_url, |
ioannis@67 | 82 | 'data_processing/measurements/{0}/download-preprocessed/') |
ioannis@43 | 83 | self.download_elda_pattern = urlparse.urljoin(self.base_url, |
ioannis@67 | 84 | 'data_processing/measurements/{0}/download-optical/') |
ioannis@45 | 85 | self.download_plots_pattern = urlparse.urljoin(self.base_url, |
moritz@29 | 86 | 'data_processing/measurements/{0}/download-plots/') |
ioannis@43 | 87 | self.download_elic_pattern = urlparse.urljoin(self.base_url, |
ioannis@67 | 88 | 'data_processing/measurements/{0}/download-elic/') |
i@14 | 89 | self.delete_measurement_pattern = urlparse.urljoin(self.base_url, 'admin/database/measurements/{0}/delete/') |
i@31 | 90 | |
ioannis@67 | 91 | # Setup API URLs for later use |
i@14 | 92 | self.api_base_url = urlparse.urljoin(self.base_url, 'api/v1/') |
i@31 | 93 | self.api_measurement_pattern = urlparse.urljoin(self.api_base_url, 'measurements/{0}/') |
i@31 | 94 | self.api_measurements_url = urlparse.urljoin(self.api_base_url, 'measurements') |
i@38 | 95 | self.api_sounding_search_pattern = urlparse.urljoin(self.api_base_url, 'sounding_files/?filename={0}') |
i@38 | 96 | self.api_lidarratio_search_pattern = urlparse.urljoin(self.api_base_url, 'lidarratio_files/?filename={0}') |
i@38 | 97 | self.api_overlap_search_pattern = urlparse.urljoin(self.api_base_url, 'overlap_files/?filename={0}') |
ioannis@43 | 98 | |
i@14 | 99 | def login(self, credentials): |
ioannis@67 | 100 | """ Login to the SCC. |
ioannis@67 | 101 | |
ioannis@67 | 102 | Parameters |
ioannis@67 | 103 | ---------- |
ioannis@67 | 104 | credentials : tuple or list |
ioannis@67 | 105 | A list or tuple in the form (username, password). |
ioannis@67 | 106 | """ |
victor@7 | 107 | logger.debug("Attempting to login to SCC, username %s." % credentials[0]) |
moritz@29 | 108 | login_credentials = {'username': credentials[0], |
moritz@29 | 109 | 'password': credentials[1]} |
victor@7 | 110 | |
i@14 | 111 | logger.debug("Accessing login page at %s." % self.login_url) |
victor@7 | 112 | |
ioannis@67 | 113 | # Get login form |
moritz@29 | 114 | login_page = self.session.get(self.login_url) |
i@14 | 115 | |
i@31 | 116 | if not login_page.ok: |
i@31 | 117 | raise self.PageNotAccessibleError('Could not access login pages. Status code %s' % login_page.status_code) |
victor@7 | 118 | |
i@31 | 119 | logger.debug("Submitting credentials.") |
victor@7 | 120 | # Submit the login data |
i@14 | 121 | login_submit = self.session.post(self.login_url, |
moritz@29 | 122 | data=login_credentials, |
victor@7 | 123 | headers={'X-CSRFToken': login_page.cookies['csrftoken'], |
moritz@29 | 124 | 'referer': self.login_url}) |
victor@7 | 125 | return login_submit |
victor@7 | 126 | |
victor@7 | 127 | def logout(self): |
ioannis@67 | 128 | """ Logout from the SCC """ |
i@31 | 129 | return self.session.get(self.logout_url, stream=True) |
victor@7 | 130 | |
claudio@69 | 131 | def upload_file(self, filename, system_id, force_upload, delete_related, delay=0, cloudfree=False, |
claudio@69 | 132 | rs_filename=None, ov_filename=None, lr_filename=None): |
ioannis@67 | 133 | """ Upload a file for processing. |
ioannis@67 | 134 | |
ioannis@67 | 135 | If the upload is successful, it returns the measurement id. |
ioannis@67 | 136 | |
ioannis@67 | 137 | |
ioannis@67 | 138 | Parameters |
ioannis@67 | 139 | ---------- |
ioannis@67 | 140 | filename : str |
ioannis@67 | 141 | File path of the file to upload |
ioannis@67 | 142 | system_id : int |
ioannis@67 | 143 | System id to be used in the processing |
ioannis@67 | 144 | force_upload : bool |
ioannis@67 | 145 | If True, if a measurement with the same ID is found on the server, it will be first deleted and the |
ioannis@67 | 146 | file current file will be uploaded. If False, the file will not be uploaded if the measurement ID is |
ioannis@67 | 147 | already present on the SCC server. |
ioannis@67 | 148 | delete_related : bool |
ioannis@67 | 149 | Answer to delete related question when deleting existing measurements from the SCC. |
claudio@69 | 150 | cloudfree : bool |
claudio@69 | 151 | Manually set the measurement as cloud free |
ioannis@67 | 152 | rs_filename, ov_filename, lr_filename : str |
ioannis@67 | 153 | Ancillary files pahts to be uploaded. |
ioannis@67 | 154 | """ |
ioannis@67 | 155 | # Get the measurement ID from the netcdf file |
ioannis@43 | 156 | measurement_id = self.measurement_id_from_file(filename) |
ioannis@43 | 157 | |
ioannis@67 | 158 | # Handle possible existing measurements with the same ID on the SCC server. |
ioannis@43 | 159 | logger.debug('Checking if a measurement with the same id already exists on the SCC server.') |
ioannis@50 | 160 | existing_measurement, _ = self.get_measurement(measurement_id) |
ioannis@43 | 161 | |
ioannis@43 | 162 | if existing_measurement: |
ioannis@43 | 163 | if force_upload: |
ioannis@43 | 164 | logger.info( |
ioannis@43 | 165 | "Measurement with id {} already exists on the SCC. Trying to delete it...".format(measurement_id)) |
ioannis@43 | 166 | self.delete_measurement(measurement_id, delete_related) |
ioannis@43 | 167 | else: |
ioannis@43 | 168 | logger.error( |
ioannis@43 | 169 | "Measurement with id {} already exists on the SCC. Use --force_upload flag to overwrite it.".format( |
ioannis@43 | 170 | measurement_id)) |
ioannis@67 | 171 | # TODO: Implement handling at the proper place. Exiting here does not allow the SCC class to be |
ioannis@67 | 172 | # used by external programs. Instead an exception should be raised. |
ioannis@43 | 173 | sys.exit(1) |
ioannis@43 | 174 | |
ioannis@67 | 175 | # Upload the file(s) |
ioannis@67 | 176 | |
victor@7 | 177 | # Get submit page |
moritz@29 | 178 | upload_page = self.session.get(self.upload_url) |
victor@7 | 179 | |
victor@7 | 180 | # Submit the data |
ioannis@54 | 181 | upload_data = {'system': system_id, |
claudio@69 | 182 | 'delay': delay, |
claudio@69 | 183 | 'manually_cloud_free': cloudfree, } |
ioannis@54 | 184 | |
claudio@69 | 185 | logger.debug("Submitted processing parameters - System: {}, Delay: {}, Manually Cloud Free: {}".format(system_id, delay, cloudfree)) |
ioannis@54 | 186 | |
victor@7 | 187 | files = {'data': open(filename, 'rb')} |
victor@7 | 188 | |
ioannis@67 | 189 | # Add ancillary files to be uploaded |
madrouin@20 | 190 | if rs_filename is not None: |
i@38 | 191 | ancillary_file, _ = self.get_ancillary(rs_filename, 'sounding') |
i@38 | 192 | |
i@38 | 193 | if ancillary_file.already_on_scc: |
ioannis@67 | 194 | logger.warning( |
ioannis@67 | 195 | "Sounding file {0.filename} already on the SCC with id {0.id}. Ignoring it.".format(ancillary_file)) |
i@38 | 196 | else: |
i@38 | 197 | logger.debug('Adding sounding file %s' % rs_filename) |
i@38 | 198 | files['sounding_file'] = open(rs_filename, 'rb') |
madrouin@20 | 199 | |
ioannis@32 | 200 | if ov_filename is not None: |
i@38 | 201 | ancillary_file, _ = self.get_ancillary(ov_filename, 'overlap') |
i@38 | 202 | |
i@38 | 203 | if ancillary_file.already_on_scc: |
ioannis@67 | 204 | logger.warning( |
ioannis@67 | 205 | "Overlap file {0.filename} already on the SCC with id {0.id}. Ignoring it.".format(ancillary_file)) |
i@38 | 206 | else: |
i@38 | 207 | logger.debug('Adding overlap file %s' % ov_filename) |
i@38 | 208 | files['overlap_file'] = open(ov_filename, 'rb') |
i@31 | 209 | |
i@31 | 210 | if lr_filename is not None: |
i@38 | 211 | ancillary_file, _ = self.get_ancillary(lr_filename, 'lidarratio') |
i@38 | 212 | |
i@38 | 213 | if ancillary_file.already_on_scc: |
i@38 | 214 | logger.warning( |
ioannis@67 | 215 | "Lidar ratio file {0.filename} already on the SCC with id {0.id}. Ignoring it.".format( |
ioannis@67 | 216 | ancillary_file)) |
i@38 | 217 | else: |
i@38 | 218 | logger.debug('Adding lidar ratio file %s' % lr_filename) |
i@38 | 219 | files['lidar_ratio_file'] = open(lr_filename, 'rb') |
i@31 | 220 | |
ioannis@67 | 221 | # Upload the files |
i@14 | 222 | logger.info("Uploading of file %s started." % filename) |
victor@7 | 223 | |
i@14 | 224 | upload_submit = self.session.post(self.upload_url, |
victor@7 | 225 | data=upload_data, |
victor@7 | 226 | files=files, |
victor@7 | 227 | headers={'X-CSRFToken': upload_page.cookies['csrftoken'], |
moritz@29 | 228 | 'referer': self.upload_url}) |
victor@7 | 229 | |
victor@7 | 230 | if upload_submit.status_code != 200: |
i@14 | 231 | logger.warning("Connection error. Status code: %s" % upload_submit.status_code) |
victor@7 | 232 | return False |
victor@7 | 233 | |
ioannis@67 | 234 | # Check if there was a redirect to a new page. If not, something went wrong |
i@14 | 235 | if upload_submit.url == self.upload_url: |
victor@7 | 236 | measurement_id = False |
ioannis@32 | 237 | logger.error("Uploaded file(s) rejected! Try to upload manually to see the error.") |
victor@7 | 238 | else: |
ioannis@67 | 239 | # TODO: Check if this is needed. This was used when the measurement ID was not read from the input file. |
ioannis@67 | 240 | measurement_id = re.findall(regex, upload_submit.text)[0] # Get the measurement ID from the output page |
ioannis@43 | 241 | logger.info("Successfully uploaded measurement with id %s." % measurement_id) |
ioannis@67 | 242 | logger.info("You can monitor the processing progress online: {}".format( |
ioannis@67 | 243 | self.measurement_page_pattern.format(measurement_id))) |
ioannis@43 | 244 | return measurement_id |
ioannis@43 | 245 | |
ioannis@43 | 246 | @staticmethod |
ioannis@43 | 247 | def measurement_id_from_file(filename): |
ioannis@67 | 248 | """ Get the measurement id from the input file. |
ioannis@67 | 249 | |
ioannis@67 | 250 | Parameters |
ioannis@67 | 251 | ---------- |
ioannis@67 | 252 | filename : str |
ioannis@67 | 253 | File path of the input file. |
ioannis@67 | 254 | """ |
ioannis@43 | 255 | |
ioannis@43 | 256 | if not os.path.isfile(filename): |
ioannis@43 | 257 | logger.error("File {} does not exist.".format(filename)) |
ioannis@43 | 258 | sys.exit(1) |
ioannis@43 | 259 | |
ioannis@43 | 260 | with netcdf.Dataset(filename) as f: |
ioannis@43 | 261 | try: |
ioannis@43 | 262 | measurement_id = f.Measurement_ID |
ioannis@43 | 263 | except AttributeError: |
ioannis@43 | 264 | logger.error( |
ioannis@43 | 265 | "Input file {} does not contain a Measurement_ID global attribute. Wrong file format?".format( |
ioannis@43 | 266 | filename)) |
ioannis@43 | 267 | sys.exit(1) |
victor@7 | 268 | |
victor@7 | 269 | return measurement_id |
victor@7 | 270 | |
victor@7 | 271 | def download_files(self, measurement_id, subdir, download_url): |
victor@7 | 272 | """ Downloads some files from the download_url to the specified |
ioannis@67 | 273 | subdir. |
ioannis@67 | 274 | |
ioannis@67 | 275 | This is a general method used to download preprocessed file, optical |
ioannis@67 | 276 | files by other, file-specific, methods. |
victor@7 | 277 | """ |
ioannis@32 | 278 | # TODO: Make downloading more robust (e.g. in case that files do not exist on server). |
victor@7 | 279 | # Get the file |
moritz@29 | 280 | request = self.session.get(download_url, stream=True) |
moritz@29 | 281 | |
moritz@29 | 282 | if not request.ok: |
moritz@29 | 283 | raise Exception("Could not download files for measurement '%s'" % measurement_id) |
victor@7 | 284 | |
victor@7 | 285 | # Create the dir if it does not exist |
victor@7 | 286 | local_dir = os.path.join(self.output_dir, measurement_id, subdir) |
victor@7 | 287 | if not os.path.exists(local_dir): |
victor@7 | 288 | os.makedirs(local_dir) |
victor@7 | 289 | |
victor@7 | 290 | # Save the file by chunk, needed if the file is big. |
ioannis@43 | 291 | memory_file = BytesIO() |
victor@7 | 292 | |
victor@7 | 293 | for chunk in request.iter_content(chunk_size=1024): |
victor@7 | 294 | if chunk: # filter out keep-alive new chunks |
victor@7 | 295 | memory_file.write(chunk) |
victor@7 | 296 | memory_file.flush() |
victor@7 | 297 | |
victor@7 | 298 | zip_file = ZipFile(memory_file) |
victor@7 | 299 | |
victor@7 | 300 | for ziped_name in zip_file.namelist(): |
victor@7 | 301 | basename = os.path.basename(ziped_name) |
victor@7 | 302 | |
victor@7 | 303 | local_file = os.path.join(local_dir, basename) |
victor@7 | 304 | |
victor@7 | 305 | with open(local_file, 'wb') as f: |
victor@7 | 306 | f.write(zip_file.read(ziped_name)) |
victor@7 | 307 | |
claudio@71 | 308 | def download_hirelpp(self, measurement_id, is_automatic_upload=False): |
ioannis@43 | 309 | """ Download hirelpp files for the measurement id. """ |
ioannis@43 | 310 | # Construct the download url |
ioannis@43 | 311 | download_url = self.download_hirelpp_pattern.format(measurement_id) |
ioannis@43 | 312 | try: |
ioannis@50 | 313 | self.download_files(measurement_id, 'hirelpp', download_url) |
claudio@71 | 314 | if is_automatic_upload: |
claudio@71 | 315 | self.sccquery = self.sccquery.replace("#hirelpp_transfer_status#", "true") |
claudio@71 | 316 | self.sccquery = self.sccquery.replace("#hirelpp_transfer_message#", "'HIRELPP products downloaded'") |
ioannis@43 | 317 | except Exception as e: |
ioannis@43 | 318 | logger.error("Could not download HiRElPP files. Error message: {}".format(e)) |
claudio@71 | 319 | if is_automatic_upload: |
claudio@71 | 320 | self.sccquery = self.sccquery.replace("#hirelpp_transfer_status#", "false") |
claudio@71 | 321 | self.sccquery = self.sccquery.replace("#hirelpp_transfer_message#", |
claudio@71 | 322 | "'Could not download HIRELPP files. Error message: {}".format(e))+"'" |
ioannis@43 | 323 | logger.debug('Download exception:', exc_info=True) |
ioannis@43 | 324 | |
claudio@71 | 325 | def download_cloudmask(self, measurement_id, is_automatic_upload=False): |
ioannis@43 | 326 | """ Download cloudmask files for the measurement id. """ |
ioannis@43 | 327 | # Construct the download url |
ioannis@43 | 328 | download_url = self.download_cloudmask_pattern.format(measurement_id) |
ioannis@43 | 329 | try: |
ioannis@50 | 330 | self.download_files(measurement_id, 'cloudscreen', download_url) |
claudio@71 | 331 | if is_automatic_upload: |
claudio@71 | 332 | self.sccquery = self.sccquery.replace("#cloudmask_transfer_status#", "true") |
claudio@71 | 333 | self.sccquery = self.sccquery.replace("#cloudmask_transfer_message#", "'CLODUMASK products downloaded'") |
ioannis@43 | 334 | except Exception as e: |
ioannis@43 | 335 | logger.error("Could not download cloudscreen files. Error message: {}".format(e)) |
claudio@71 | 336 | if is_automatic_upload: |
claudio@71 | 337 | self.sccquery = self.sccquery.replace("#cloudmask_transfer_status#", "false") |
claudio@71 | 338 | self.sccquery = self.sccquery.replace("#cloudmask_transfer_message#", |
claudio@71 | 339 | "'Could not download CLOUDMASK files. Error message: {}".format(e))+"'" |
ioannis@43 | 340 | logger.debug('Download exception:', exc_info=True) |
ioannis@43 | 341 | |
claudio@71 | 342 | def download_elpp(self, measurement_id, is_automatic_upload=False): |
victor@7 | 343 | """ Download preprocessed files for the measurement id. """ |
victor@7 | 344 | # Construct the download url |
ioannis@43 | 345 | download_url = self.download_elpp_pattern.format(measurement_id) |
ioannis@43 | 346 | try: |
ioannis@50 | 347 | self.download_files(measurement_id, 'elpp', download_url) |
claudio@71 | 348 | if is_automatic_upload: |
claudio@71 | 349 | self.sccquery = self.sccquery.replace("#elpp_transfer_status#", "true") |
claudio@71 | 350 | self.sccquery = self.sccquery.replace("#elpp_transfer_message#", "'ELPP products downloaded'") |
ioannis@43 | 351 | except Exception as e: |
ioannis@43 | 352 | logger.error("Could not download ElPP files. Error message: {}".format(e)) |
claudio@71 | 353 | if is_automatic_upload: |
claudio@71 | 354 | self.sccquery = self.sccquery.replace("#elpp_transfer_status#", "false") |
claudio@71 | 355 | self.sccquery = self.sccquery.replace("#elpp_transfer_message#", |
claudio@71 | 356 | "'Could not download ELPP files. Error message: {}".format(e))+"'" |
ioannis@43 | 357 | logger.debug('Download exception:', exc_info=True) |
victor@7 | 358 | |
claudio@71 | 359 | def download_elda(self, measurement_id, is_automatic_upload=False): |
victor@7 | 360 | """ Download optical files for the measurement id. """ |
victor@7 | 361 | # Construct the download url |
ioannis@43 | 362 | download_url = self.download_elda_pattern.format(measurement_id) |
ioannis@43 | 363 | try: |
ioannis@50 | 364 | self.download_files(measurement_id, 'elda', download_url) |
claudio@71 | 365 | if is_automatic_upload: |
claudio@71 | 366 | self.sccquery = self.sccquery.replace("#elda_transfer_status#", "true") |
claudio@71 | 367 | self.sccquery = self.sccquery.replace("#elda_transfer_message#", "'ELDA products downloaded'") |
ioannis@43 | 368 | except Exception as e: |
ioannis@43 | 369 | logger.error("Could not download ELDA files. Error message: {}".format(e)) |
claudio@71 | 370 | if is_automatic_upload: |
claudio@71 | 371 | self.sccquery = self.sccquery.replace("#elda_transfer_status#", "false") |
claudio@71 | 372 | self.sccquery = self.sccquery.replace("#elda_transfer_message#", |
claudio@71 | 373 | "'Could not download ELDA files. Error message: {}".format(e))+"'" |
ioannis@43 | 374 | logger.debug('Download exception:', exc_info=True) |
victor@7 | 375 | |
ioannis@43 | 376 | def download_plots(self, measurement_id): |
victor@7 | 377 | """ Download profile graphs for the measurement id. """ |
victor@7 | 378 | # Construct the download url |
ioannis@45 | 379 | download_url = self.download_plots_pattern.format(measurement_id) |
ioannis@43 | 380 | try: |
ioannis@50 | 381 | self.download_files(measurement_id, 'elda_plots', download_url) |
ioannis@43 | 382 | except Exception as e: |
ioannis@43 | 383 | logger.error("Could not download ELDA plots. Error message: {}".format(e)) |
ioannis@43 | 384 | logger.debug('Download exception:', exc_info=True) |
victor@7 | 385 | |
claudio@71 | 386 | def download_elic(self, measurement_id, is_automatic_upload=False): |
ioannis@43 | 387 | """ Download ELIC files for the measurement id. """ |
ioannis@43 | 388 | # Construct the download url |
ioannis@43 | 389 | download_url = self.download_elic_pattern.format(measurement_id) |
ioannis@43 | 390 | try: |
ioannis@50 | 391 | self.download_files(measurement_id, 'elic', download_url) |
claudio@71 | 392 | if is_automatic_upload: |
claudio@71 | 393 | self.sccquery = self.sccquery.replace("#elic_transfer_status#", "true") |
claudio@71 | 394 | self.sccquery = self.sccquery.replace("#elic_transfer_message#", "'ELIC products downloaded'") |
ioannis@43 | 395 | except Exception as e: |
ioannis@43 | 396 | logger.error("Could not download ELIC files. Error message: {}".format(e)) |
claudio@71 | 397 | if is_automatic_upload: |
claudio@71 | 398 | self.sccquery = self.sccquery.replace("#elic_transfer_status#", "false") |
claudio@71 | 399 | self.sccquery = self.sccquery.replace("#elic_transfer_message#", |
claudio@71 | 400 | "'Could not download ELIC files. Error message: {}".format(e))+"'" |
ioannis@43 | 401 | logger.debug('Download exception:', exc_info=True) |
ioannis@43 | 402 | |
claudio@71 | 403 | def download_eldec(self, measurement_id, is_automatic_upload=False): |
ioannis@43 | 404 | """ Download ELDEC files for the measurement id. """ |
ioannis@43 | 405 | # Construct the download url |
ioannis@43 | 406 | download_url = self.download_elda_pattern.format(measurement_id) # ELDA patter is used for now |
ioannis@43 | 407 | try: |
ioannis@50 | 408 | self.download_files(measurement_id, 'eldec', download_url) |
claudio@71 | 409 | if is_automatic_upload: |
claudio@71 | 410 | self.sccquery = self.sccquery.replace("#eldec_transfer_status#", "true") |
claudio@71 | 411 | self.sccquery = self.sccquery.replace("#eldec_transfer_message#", "'ELDEC products downloaded'") |
ioannis@43 | 412 | except Exception as e: |
ioannis@43 | 413 | logger.error("Could not download EDELC files. Error message: {}".format(e)) |
claudio@71 | 414 | if is_automatic_upload: |
claudio@71 | 415 | self.sccquery = self.sccquery.replace("#eldec_transfer_status#", "false") |
claudio@71 | 416 | self.sccquery = self.sccquery.replace("#eldec_transfer_message#", |
claudio@71 | 417 | "'Could not download ELDEC files. Error message: {}".format(e))+"'" |
ioannis@43 | 418 | logger.debug('Download exception:', exc_info=True) |
ioannis@43 | 419 | |
ioannis@43 | 420 | def rerun_elpp(self, measurement_id, monitor=True): |
ioannis@45 | 421 | logger.debug("Started rerun_elpp procedure.") |
ioannis@45 | 422 | |
ioannis@45 | 423 | logger.debug("Getting measurement %s" % measurement_id) |
madrouin@26 | 424 | measurement, status = self.get_measurement(measurement_id) |
victor@7 | 425 | |
victor@7 | 426 | if measurement: |
ioannis@45 | 427 | logger.debug("Attempting to rerun ElPP through %s." % measurement.rerun_all_url) |
ioannis@45 | 428 | request = self.session.get(measurement.rerun_elpp_url, stream=True) |
victor@7 | 429 | |
victor@7 | 430 | if request.status_code != 200: |
i@14 | 431 | logger.error( |
i@14 | 432 | "Could not rerun processing for %s. Status code: %s" % (measurement_id, request.status_code)) |
ioannis@45 | 433 | else: |
ioannis@45 | 434 | logger.info("Rerun-elpp command submitted successfully for id {}.".format(measurement_id)) |
victor@7 | 435 | |
victor@7 | 436 | if monitor: |
victor@7 | 437 | self.monitor_processing(measurement_id) |
victor@7 | 438 | |
victor@7 | 439 | def rerun_all(self, measurement_id, monitor=True): |
victor@7 | 440 | logger.debug("Started rerun_all procedure.") |
victor@7 | 441 | |
victor@7 | 442 | logger.debug("Getting measurement %s" % measurement_id) |
madrouin@26 | 443 | measurement, status = self.get_measurement(measurement_id) |
victor@7 | 444 | |
victor@7 | 445 | if measurement: |
victor@7 | 446 | logger.debug("Attempting to rerun all processing through %s." % measurement.rerun_all_url) |
victor@7 | 447 | |
moritz@29 | 448 | request = self.session.get(measurement.rerun_all_url, stream=True) |
victor@7 | 449 | |
victor@7 | 450 | if request.status_code != 200: |
victor@7 | 451 | logger.error("Could not rerun pre processing for %s. Status code: %s" % |
victor@7 | 452 | (measurement_id, request.status_code)) |
ioannis@45 | 453 | else: |
ioannis@45 | 454 | logger.info("Rerun-all command submitted successfully for id {}.".format(measurement_id)) |
victor@7 | 455 | |
victor@7 | 456 | if monitor: |
victor@7 | 457 | self.monitor_processing(measurement_id) |
victor@7 | 458 | |
claudio@69 | 459 | def process(self, filename, system_id, monitor, force_upload, delete_related, delay=0, cloudfree=False, |
claudio@69 | 460 | rs_filename=None, lr_filename=None, ov_filename=None): |
victor@7 | 461 | """ Upload a file for processing and wait for the processing to finish. |
victor@7 | 462 | If the processing is successful, it will download all produced files. |
victor@7 | 463 | """ |
victor@7 | 464 | logger.info("--- Processing started on %s. ---" % datetime.datetime.now()) |
victor@7 | 465 | # Upload file |
ioannis@50 | 466 | logger.info("Uploading file.") |
ioannis@45 | 467 | measurement_id = self.upload_file(filename, system_id, force_upload, delete_related, |
ioannis@54 | 468 | delay=delay, |
claudio@69 | 469 | cloudfree=cloudfree, |
ioannis@32 | 470 | rs_filename=rs_filename, |
ioannis@32 | 471 | lr_filename=lr_filename, |
ioannis@32 | 472 | ov_filename=ov_filename) |
ioannis@43 | 473 | |
ioannis@54 | 474 | if monitor and (delay > 0): |
ioannis@54 | 475 | logger.warning("Will not start monitoring, since a delay was specified: {} hours.".format(delay)) |
ioannis@54 | 476 | return None |
ioannis@54 | 477 | |
ioannis@32 | 478 | if measurement_id and monitor: |
ioannis@54 | 479 | logger.info("Monitoring processing.") |
moritz@29 | 480 | return self.monitor_processing(measurement_id) |
victor@7 | 481 | |
moritz@29 | 482 | return None |
victor@7 | 483 | |
ioannis@61 | 484 | def monitor_processing(self, measurement_id, retry_max=2, time_sleep=2, exit_if_missing=True): |
victor@7 | 485 | """ Monitor the processing progress of a measurement id""" |
victor@7 | 486 | |
madrouin@24 | 487 | # try to deal with error 404 |
ioannis@61 | 488 | attempts_count = 0 |
ioannis@61 | 489 | max_attempts = retry_max + 1 |
madrouin@24 | 490 | |
ioannis@67 | 491 | # try to wait for measurement to appear in API. A user has reported that this does not happen immediately. |
madrouin@24 | 492 | measurement = None |
ioannis@50 | 493 | logger.info("Looking for measurement %s on the SCC.", measurement_id) |
ioannis@62 | 494 | |
ioannis@61 | 495 | while attempts_count < max_attempts: |
ioannis@61 | 496 | attempts_count += 1 |
madrouin@26 | 497 | measurement, status = self.get_measurement(measurement_id) |
ioannis@61 | 498 | if status != 200: |
ioannis@61 | 499 | logger.warning("Measurement not found.") |
ioannis@61 | 500 | if attempts_count < max_attempts: |
ioannis@61 | 501 | logger.warning("Waiting %ds.", time_sleep) |
ioannis@61 | 502 | time.sleep(time_sleep) |
madrouin@26 | 503 | else: |
madrouin@26 | 504 | break |
ioannis@62 | 505 | print("Measurement: {}".format(measurement)) |
madrouin@24 | 506 | |
ioannis@62 | 507 | if measurement is None: |
ioannis@61 | 508 | logger.error("Measurement %s doesn't seem to exist.", measurement_id) |
ioannis@61 | 509 | if exit_if_missing: |
ioannis@61 | 510 | sys.exit(1) |
ioannis@61 | 511 | else: |
ioannis@61 | 512 | return measurement |
madrouin@26 | 513 | |
ioannis@45 | 514 | logger.info('Measurement %s found.', measurement_id) |
ioannis@62 | 515 | while not measurement.has_finished: |
ioannis@62 | 516 | measurement.log_processing_status() |
ioannis@62 | 517 | time.sleep(10) |
ioannis@62 | 518 | measurement, status = self.get_measurement(measurement_id) |
madrouin@24 | 519 | |
ioannis@62 | 520 | logger.info("Measurement processing finished.") |
ioannis@62 | 521 | measurement.log_detailed_status() |
ioannis@53 | 522 | |
ioannis@62 | 523 | if measurement.hirelpp == 127: |
ioannis@62 | 524 | logger.info("Downloading HiRElPP files.") |
ioannis@62 | 525 | self.download_hirelpp(measurement_id) |
ioannis@62 | 526 | if measurement.cloudmask == 127: |
ioannis@62 | 527 | logger.info("Downloading cloud screening files.") |
ioannis@62 | 528 | self.download_cloudmask(measurement_id) |
ioannis@62 | 529 | if measurement.elpp == 127: |
ioannis@62 | 530 | logger.info("Downloading ElPP files.") |
ioannis@62 | 531 | self.download_elpp(measurement_id) |
ioannis@62 | 532 | if measurement.elda == 127: |
ioannis@62 | 533 | logger.info("Downloading ELDA files.") |
ioannis@62 | 534 | self.download_elda(measurement_id) |
ioannis@62 | 535 | logger.info("Downloading ELDA plots.") |
ioannis@62 | 536 | self.download_plots(measurement_id) |
ioannis@62 | 537 | if measurement.elic == 127: |
ioannis@62 | 538 | logger.info("Downloading ELIC files.") |
ioannis@62 | 539 | self.download_elic(measurement_id) |
ioannis@67 | 540 | if measurement.is_calibration and measurement.eldec == 0: |
ioannis@62 | 541 | logger.info("Downloading ELDEC files.") |
ioannis@62 | 542 | self.download_eldec(measurement_id) |
ioannis@62 | 543 | logger.info("--- Processing finished. ---") |
ioannis@43 | 544 | |
victor@7 | 545 | return measurement |
victor@7 | 546 | |
claudio@71 | 547 | def download_products(self, measurement, dir_name, is_automatic_upload=False): |
claudio@71 | 548 | """ Download all the products of a measurement id (used for E-SHAPE and automatic_upload)""" |
claudio@71 | 549 | |
claudio@68 | 550 | measurement_id = measurement.id |
claudio@71 | 551 | self.sccquery = self.sccquery.replace("#__measurements__ID#", "'"+measurement_id+"'") |
claudio@68 | 552 | base_output_dir = self.output_dir |
claudio@68 | 553 | self.output_dir = self.output_dir + dir_name + "/" |
claudio@68 | 554 | |
claudio@68 | 555 | if measurement.hirelpp == 127: |
claudio@68 | 556 | logger.info("Downloading HiRElPP files.") |
claudio@71 | 557 | self.download_hirelpp(measurement_id, is_automatic_upload) |
claudio@71 | 558 | else: |
claudio@71 | 559 | if is_automatic_upload: |
claudio@71 | 560 | self.sccquery = self.sccquery.replace("#hirelpp_transfer_status#", "NULL") |
claudio@71 | 561 | self.sccquery = self.sccquery.replace("#hirelpp_transfer_message#", "'No HIRELPP products available'") |
claudio@68 | 562 | if measurement.cloudmask == 127: |
claudio@68 | 563 | logger.info("Downloading cloud screening files.") |
claudio@71 | 564 | self.download_cloudmask(measurement_id, is_automatic_upload) |
claudio@71 | 565 | else: |
claudio@71 | 566 | if is_automatic_upload: |
claudio@71 | 567 | self.sccquery = self.sccquery.replace("#cloudmask_transfer_status#", "NULL") |
claudio@71 | 568 | self.sccquery = self.sccquery.replace("#cloudmask_transfer_message#", "'No CLOUDMASK products available'") |
claudio@68 | 569 | if measurement.elpp == 127: |
claudio@68 | 570 | logger.info("Downloading ElPP files.") |
claudio@71 | 571 | self.download_elpp(measurement_id, is_automatic_upload) |
claudio@71 | 572 | else: |
claudio@71 | 573 | if is_automatic_upload: |
claudio@71 | 574 | self.sccquery = self.sccquery.replace("#elpp_transfer_status#", "NULL") |
claudio@71 | 575 | self.sccquery = self.sccquery.replace("#elpp_transfer_message#", "'No ELPP products available'") |
claudio@68 | 576 | if measurement.elda == 127: |
claudio@68 | 577 | logger.info("Downloading ELDA files.") |
claudio@71 | 578 | self.download_elda(measurement_id, is_automatic_upload) |
claudio@71 | 579 | if not is_automatic_upload: |
claudio@71 | 580 | logger.info("Downloading ELDA plots.") |
claudio@71 | 581 | self.download_plots(measurement_id) |
claudio@71 | 582 | else: |
claudio@71 | 583 | if is_automatic_upload: |
claudio@71 | 584 | self.sccquery = self.sccquery.replace("#elda_transfer_status#", "NULL") |
claudio@71 | 585 | self.sccquery = self.sccquery.replace("#elda_transfer_message#", "'No ELDA products available'") |
claudio@68 | 586 | if measurement.elic == 127: |
claudio@68 | 587 | logger.info("Downloading ELIC files.") |
claudio@71 | 588 | self.download_elic(measurement_id, is_automatic_upload) |
claudio@71 | 589 | else: |
claudio@71 | 590 | if is_automatic_upload: |
claudio@71 | 591 | self.sccquery = self.sccquery.replace("#elic_transfer_status#", "NULL") |
claudio@71 | 592 | self.sccquery = self.sccquery.replace("#elic_transfer_message#", "'No ELIC products available'") |
claudio@71 | 593 | if measurement.is_calibration and measurement.eldec == 127: |
claudio@68 | 594 | logger.info("Downloading ELDEC files.") |
claudio@71 | 595 | self.download_eldec(measurement_id, is_automatic_upload) |
claudio@71 | 596 | elif not measurement.is_calibration: |
claudio@71 | 597 | if is_automatic_upload: |
claudio@71 | 598 | self.sccquery = self.sccquery.replace("#eldec_transfer_status#", "NULL") |
claudio@71 | 599 | self.sccquery = self.sccquery.replace("#eldec_transfer_message#", "'The measurement is not a calibration'") |
claudio@71 | 600 | elif measurement.is_calibration and measurement.eldec != 127: |
claudio@71 | 601 | if is_automatic_upload: |
claudio@71 | 602 | self.sccquery = self.sccquery.replace("#eldec_transfer_status#", "NULL") |
claudio@71 | 603 | self.sccquery = self.sccquery.replace("#eldec_transfer_message#", "'No ELDEC products available'") |
claudio@68 | 604 | logger.info("--- Processing finished. ---") |
claudio@68 | 605 | |
claudio@68 | 606 | self.output_dir = base_output_dir |
claudio@68 | 607 | return measurement |
claudio@68 | 608 | |
ioannis@43 | 609 | def get_measurement(self, measurement_id): |
ioannis@67 | 610 | """ Get a measurement information from the SCC API. |
victor@7 | 611 | |
ioannis@67 | 612 | Parameters |
ioannis@67 | 613 | ---------- |
ioannis@67 | 614 | measurement_id : str |
ioannis@67 | 615 | The measurement ID to search. |
ioannis@67 | 616 | |
ioannis@67 | 617 | Returns |
ioannis@67 | 618 | ------- |
ioannis@67 | 619 | : Measurement object or None |
ioannis@67 | 620 | If the measurement is found, a Measurement object is returned. If not, it returns None |
ioannis@67 | 621 | """ |
ioannis@67 | 622 | # TODO: Consider to homogenize with get_ancillary method (i.e. always return a Measurement object). |
ioannis@67 | 623 | |
ioannis@67 | 624 | if measurement_id is None: # TODO: Is this still required? |
victor@7 | 625 | return None |
victor@7 | 626 | |
ioannis@67 | 627 | # Access the API |
i@31 | 628 | measurement_url = self.api_measurement_pattern.format(measurement_id) |
i@31 | 629 | logger.debug("Measurement API URL: %s" % measurement_url) |
victor@7 | 630 | |
moritz@29 | 631 | response = self.session.get(measurement_url) |
victor@7 | 632 | |
ioannis@43 | 633 | response_dict = None |
ioannis@45 | 634 | |
ioannis@43 | 635 | if response.status_code == 200: |
ioannis@43 | 636 | response_dict = response.json() |
ioannis@45 | 637 | elif response.status_code == 404: |
ioannis@43 | 638 | logger.info("No measurement with id %s found on the SCC." % measurement_id) |
ioannis@45 | 639 | else: |
i@14 | 640 | logger.error('Could not access API. Status code %s.' % response.status_code) |
victor@7 | 641 | |
ioannis@55 | 642 | # TODO: Implement better handling for status 401. |
ioannis@55 | 643 | |
victor@7 | 644 | if response_dict: |
ioannis@43 | 645 | measurement = Measurement(self.base_url, response_dict) |
victor@7 | 646 | else: |
ioannis@45 | 647 | measurement = None |
victor@7 | 648 | |
ioannis@45 | 649 | return measurement, response.status_code |
victor@7 | 650 | |
ioannis@45 | 651 | def delete_measurement(self, measurement_id, delete_related): |
victor@7 | 652 | """ Deletes a measurement with the provided measurement id. The user |
ioannis@67 | 653 | should have the appropriate permissions (i.e. access to the admin site). |
madrouin@20 | 654 | |
victor@7 | 655 | The procedures is performed directly through the web interface and |
victor@7 | 656 | NOT through the API. |
victor@7 | 657 | """ |
victor@7 | 658 | # Get the measurement object |
i@31 | 659 | measurement, _ = self.get_measurement(measurement_id) |
victor@7 | 660 | |
victor@7 | 661 | # Check that it exists |
victor@7 | 662 | if measurement is None: |
victor@7 | 663 | logger.warning("Nothing to delete.") |
victor@7 | 664 | return None |
victor@7 | 665 | |
victor@7 | 666 | # Go the the page confirming the deletion |
moritz@29 | 667 | delete_url = self.delete_measurement_pattern.format(measurement_id) |
ioannis@43 | 668 | |
moritz@29 | 669 | confirm_page = self.session.get(delete_url) |
victor@7 | 670 | |
victor@7 | 671 | # Check that the page opened properly |
victor@7 | 672 | if confirm_page.status_code != 200: |
victor@7 | 673 | logger.warning("Could not open delete page. Status: {0}".format(confirm_page.status_code)) |
victor@7 | 674 | return None |
victor@7 | 675 | |
ioannis@43 | 676 | # Get the delete related value |
ioannis@43 | 677 | if delete_related: |
ioannis@43 | 678 | delete_related_option = 'delete_related' |
ioannis@43 | 679 | else: |
ioannis@43 | 680 | delete_related_option = 'not_delete_related' |
ioannis@43 | 681 | |
victor@7 | 682 | # Delete the measurement |
victor@7 | 683 | delete_page = self.session.post(delete_url, |
ioannis@43 | 684 | data={'post': 'yes', |
ioannis@43 | 685 | 'select_delete_related_measurements': delete_related_option}, |
victor@7 | 686 | headers={'X-CSRFToken': confirm_page.cookies['csrftoken'], |
victor@7 | 687 | 'referer': delete_url} |
victor@7 | 688 | ) |
i@31 | 689 | if not delete_page.ok: |
victor@7 | 690 | logger.warning("Something went wrong. Delete page status: {0}".format( |
victor@7 | 691 | delete_page.status_code)) |
victor@7 | 692 | return None |
victor@7 | 693 | |
victor@7 | 694 | logger.info("Deleted measurement {0}".format(measurement_id)) |
victor@7 | 695 | return True |
victor@7 | 696 | |
claudio@70 | 697 | def available_measurements(self, start_gte=None, stop_lte=None, is_being_processed=None, is_queued=None): |
ioannis@67 | 698 | """ Get a list of available measurement on the SCC. |
ioannis@67 | 699 | |
ioannis@67 | 700 | The methods is currently not used, could be merged with list_measurements. |
ioannis@67 | 701 | """ |
claudio@68 | 702 | |
claudio@68 | 703 | params = {} |
claudio@68 | 704 | if start_gte is not None: |
claudio@68 | 705 | params['start__gte'] = start_gte |
claudio@68 | 706 | if stop_lte is not None: |
claudio@68 | 707 | params['stop__lte'] = stop_lte |
claudio@70 | 708 | if is_being_processed is not None: |
claudio@70 | 709 | params['is_being_processed'] = is_being_processed |
claudio@70 | 710 | if is_queued is not None: |
claudio@70 | 711 | params['is_queued'] = is_queued |
claudio@70 | 712 | |
claudio@68 | 713 | |
claudio@68 | 714 | response = self.session.get(self.api_measurements_url, params=params) |
victor@7 | 715 | response_dict = response.json() |
victor@7 | 716 | |
victor@7 | 717 | if response_dict: |
victor@7 | 718 | measurement_list = response_dict['objects'] |
i@14 | 719 | measurements = [Measurement(self.base_url, measurement_dict) for measurement_dict in measurement_list] |
victor@7 | 720 | logger.info("Found %s measurements on the SCC." % len(measurements)) |
victor@7 | 721 | else: |
victor@7 | 722 | logger.warning("No response received from the SCC when asked for available measurements.") |
ioannis@43 | 723 | measurements = None |
victor@7 | 724 | |
victor@7 | 725 | return measurements |
victor@7 | 726 | |
claudio@68 | 727 | def list_measurements(self, id_exact=None, id_startswith=None, |
claudio@68 | 728 | start_exact=None, start_gte=None, start_lte=None, |
claudio@68 | 729 | stop_exact=None, stop_gte=None, stop_lte=None, |
claudio@68 | 730 | station_exact=None, station_in=None): |
ioannis@67 | 731 | """ Get the response text from the API. """ |
moritz@29 | 732 | |
ioannis@67 | 733 | # TODO: Add some error handling, e.g. as per available_measurements method |
i@31 | 734 | |
moritz@29 | 735 | # Need to set to empty string if not specified, we won't get any results |
ioannis@65 | 736 | params = {} |
i@31 | 737 | |
ioannis@65 | 738 | if id_exact is not None: |
ioannis@65 | 739 | params['id__exact'] = id_exact |
claudio@68 | 740 | elif id_startswith is not None: |
claudio@68 | 741 | params['id__startswith'] = id_startswith |
claudio@68 | 742 | |
claudio@68 | 743 | if start_exact is not None: |
claudio@68 | 744 | params['start__exact'] = start_exact |
ioannis@65 | 745 | else: |
claudio@68 | 746 | if start_gte is not None: |
claudio@68 | 747 | params['start__gte'] = start_gte |
claudio@68 | 748 | if start_lte is not None: |
claudio@68 | 749 | params['start__lte'] = start_lte |
claudio@68 | 750 | |
claudio@68 | 751 | if stop_exact is not None: |
claudio@68 | 752 | params['stop__exact'] = stop_exact |
claudio@68 | 753 | else: |
claudio@68 | 754 | if stop_gte is not None: |
claudio@68 | 755 | params['stop__gte'] = stop_gte |
claudio@68 | 756 | if stop_lte is not None: |
claudio@68 | 757 | params['stop__lte'] = stop_lte |
claudio@68 | 758 | |
claudio@68 | 759 | if station_exact is not None: |
claudio@68 | 760 | params['station__exact'] = station_exact |
claudio@68 | 761 | elif station_in is not None: |
claudio@68 | 762 | params['station__in'] = station_in |
moritz@29 | 763 | |
ioannis@65 | 764 | response_json = self.session.get(self.api_measurements_url, params=params).text |
moritz@29 | 765 | |
ioannis@65 | 766 | return response_json |
moritz@29 | 767 | |
ioannis@43 | 768 | def measurement_id_for_date(self, t1, call_sign, base_number=0): |
victor@7 | 769 | """ Give the first available measurement id on the SCC for the specific |
madrouin@20 | 770 | date. |
victor@7 | 771 | """ |
ioannis@67 | 772 | # TODO: Check if this method needs updating to handle all measurement_ID formats. |
victor@7 | 773 | date_str = t1.strftime('%Y%m%d') |
i@31 | 774 | base_id = "%s%s" % (date_str, call_sign) |
i@31 | 775 | search_url = urlparse.urljoin(self.api_base_url, 'measurements/?id__startswith=%s' % base_id) |
victor@7 | 776 | |
moritz@29 | 777 | response = self.session.get(search_url) |
victor@7 | 778 | |
victor@7 | 779 | response_dict = response.json() |
victor@7 | 780 | |
victor@7 | 781 | measurement_id = None |
victor@7 | 782 | |
victor@7 | 783 | if response_dict: |
victor@7 | 784 | measurement_list = response_dict['objects'] |
i@31 | 785 | |
i@31 | 786 | if len(measurement_list) == 100: |
i@31 | 787 | raise ValueError('No available measurement id found.') |
i@31 | 788 | |
victor@7 | 789 | existing_ids = [measurement_dict['id'] for measurement_dict in measurement_list] |
victor@7 | 790 | |
victor@7 | 791 | measurement_number = base_number |
i@31 | 792 | measurement_id = "%s%02i" % (base_id, measurement_number) |
victor@7 | 793 | |
victor@7 | 794 | while measurement_id in existing_ids: |
victor@7 | 795 | measurement_number = measurement_number + 1 |
i@31 | 796 | measurement_id = "%s%02i" % (base_id, measurement_number) |
victor@7 | 797 | |
victor@7 | 798 | return measurement_id |
victor@7 | 799 | |
i@40 | 800 | def get_ancillary(self, file_path, file_type): |
i@38 | 801 | """ |
i@38 | 802 | Try to get the ancillary file data from the SCC API. |
i@38 | 803 | |
i@38 | 804 | The result will always be an API object. If the file does not exist, the .exists property is set to False. |
i@38 | 805 | |
i@38 | 806 | Parameters |
i@38 | 807 | ---------- |
i@40 | 808 | file_path : str |
i@40 | 809 | Path of the uploaded file. |
i@38 | 810 | file_type : str |
i@38 | 811 | Type of ancillary file. One of 'sounding', 'overlap', 'lidarratio'. |
i@38 | 812 | |
i@38 | 813 | Returns |
i@38 | 814 | : AncillaryFile |
i@38 | 815 | The api object. |
i@38 | 816 | """ |
i@38 | 817 | assert file_type in ['sounding', 'overlap', 'lidarratio'] |
i@38 | 818 | |
i@40 | 819 | filename = os.path.basename(file_path) |
i@40 | 820 | |
i@38 | 821 | if file_type == 'sounding': |
i@38 | 822 | file_url = self.api_sounding_search_pattern.format(filename) |
i@38 | 823 | elif file_type == 'overlap': |
i@38 | 824 | file_url = self.api_overlap_search_pattern.format(filename) |
i@38 | 825 | else: |
i@38 | 826 | file_url = self.api_lidarratio_search_pattern.format(filename) |
i@38 | 827 | |
i@38 | 828 | response = self.session.get(file_url) |
i@38 | 829 | |
i@38 | 830 | if not response.ok: |
i@38 | 831 | logger.error('Could not access API. Status code %s.' % response.status_code) |
i@38 | 832 | return None, response.status_code |
i@38 | 833 | |
i@38 | 834 | response_dict = response.json() |
i@38 | 835 | object_list = response_dict['objects'] |
i@38 | 836 | |
i@38 | 837 | logger.debug("Ancillary file JSON: {0}".format(object_list)) |
i@38 | 838 | |
i@38 | 839 | if object_list: |
i@38 | 840 | ancillary_file = AncillaryFile(self.api_base_url, object_list[0]) # Assume only one file is returned |
i@38 | 841 | else: |
i@38 | 842 | ancillary_file = AncillaryFile(self.api_base_url, None) # Create an empty object |
i@38 | 843 | |
i@38 | 844 | return ancillary_file, response.status_code |
i@38 | 845 | |
ioannis@43 | 846 | def __enter__(self): |
ioannis@43 | 847 | return self |
victor@7 | 848 | |
ioannis@43 | 849 | def __exit__(self, *args): |
ioannis@43 | 850 | logger.debug("Closing SCC connection session.") |
ioannis@43 | 851 | self.session.close() |
ioannis@43 | 852 | |
i@31 | 853 | class PageNotAccessibleError(RuntimeError): |
i@31 | 854 | pass |
ioannis@43 | 855 | |
victor@7 | 856 | |
ioannis@17 | 857 | class ApiObject(object): |
victor@7 | 858 | """ A generic class object. """ |
victor@7 | 859 | |
i@14 | 860 | def __init__(self, base_url, dict_response): |
i@14 | 861 | self.base_url = base_url |
victor@7 | 862 | |
victor@7 | 863 | if dict_response: |
victor@7 | 864 | # Add the dictionary key value pairs as object properties |
victor@7 | 865 | for key, value in dict_response.items(): |
ioannis@17 | 866 | # logger.debug('Setting key {0} to value {1}'.format(key, value)) |
ioannis@17 | 867 | try: |
ioannis@17 | 868 | setattr(self, key, value) |
ioannis@17 | 869 | except: |
ioannis@17 | 870 | logger.warning('Could not set attribute {0} to value {1}'.format(key, value)) |
victor@7 | 871 | self.exists = True |
victor@7 | 872 | else: |
victor@7 | 873 | self.exists = False |
victor@7 | 874 | |
victor@7 | 875 | |
victor@7 | 876 | class Measurement(ApiObject): |
ioannis@43 | 877 | """ This class represents the measurement object as returned in the SCC API. |
ioannis@43 | 878 | """ |
victor@7 | 879 | |
i@14 | 880 | def __init__(self, base_url, dict_response): |
victor@7 | 881 | |
ioannis@45 | 882 | # Define expected attributes to assist debugging |
ioannis@53 | 883 | |
ioannis@53 | 884 | self.hirelpp = None |
ioannis@53 | 885 | self.hirelpp_exit_code = None |
ioannis@43 | 886 | self.cloudmask = None |
ioannis@53 | 887 | self.cloudmask_exit_code = None |
ioannis@53 | 888 | self.elpp = None |
ioannis@53 | 889 | self.elpp_exit_code = None |
ioannis@43 | 890 | self.elda = None |
ioannis@53 | 891 | self.elda_exit_code = None |
ioannis@43 | 892 | self.elic = None |
ioannis@53 | 893 | self.elic_exit_code = None |
ioannis@53 | 894 | self.eldec = None |
ioannis@53 | 895 | self.eldec_exit_code = None |
ioannis@53 | 896 | self.elquick = None |
ioannis@53 | 897 | self.elquick_exit_code = None |
ioannis@53 | 898 | |
ioannis@43 | 899 | self.id = None |
claudio@68 | 900 | self.num_id = None |
ioannis@43 | 901 | self.is_calibration = None |
ioannis@43 | 902 | self.is_running = None |
claudio@68 | 903 | self.is_queued = None |
ioannis@53 | 904 | |
ioannis@43 | 905 | self.resource_uri = None |
ioannis@43 | 906 | self.start = None |
ioannis@43 | 907 | self.stop = None |
ioannis@43 | 908 | self.system = None |
claudio@68 | 909 | self.station = None |
ioannis@43 | 910 | self.upload = None |
ioannis@43 | 911 | |
ioannis@45 | 912 | super().__init__(base_url, dict_response) |
victor@7 | 913 | |
ioannis@57 | 914 | @property |
ioannis@57 | 915 | def has_finished(self): |
ioannis@57 | 916 | """ Temporary implementation for SCC version 5.2.0, until the API returns a flag indicating if the |
ioannis@57 | 917 | processing measurement has finished. """ |
ioannis@57 | 918 | if (self.is_running is False) and (self.hirelpp != 0 or self.elpp != 0): |
ioannis@57 | 919 | return True |
ioannis@57 | 920 | else: |
ioannis@57 | 921 | return False |
ioannis@57 | 922 | |
ioannis@53 | 923 | def log_processing_status(self): |
ioannis@53 | 924 | """ Log module status. """ |
claudio@68 | 925 | logger.info("Measurement is being processed. Status: {}, {}, {}, {}, {}, {}, {}). Please wait.".format( |
ioannis@53 | 926 | self.upload, |
ioannis@53 | 927 | self.hirelpp, |
ioannis@53 | 928 | self.cloudmask, |
ioannis@53 | 929 | self.elpp, |
ioannis@53 | 930 | self.elda, |
claudio@68 | 931 | self.elic, |
claudio@68 | 932 | self.elquick)) |
ioannis@53 | 933 | |
ioannis@53 | 934 | def log_detailed_status(self): |
ioannis@53 | 935 | """ Log module exit and status codes.""" |
ioannis@53 | 936 | logger.info("Measurement exit status:".format(self.id)) |
ioannis@53 | 937 | if self.is_calibration: |
ioannis@53 | 938 | self._log_module_status('ElPP', self.elpp, self.elpp_exit_code) |
ioannis@53 | 939 | self._log_module_status('ElDEC', self.eldec, self.eldec_exit_code) |
ioannis@53 | 940 | else: |
ioannis@53 | 941 | self._log_module_status('HiRElPP', self.hirelpp, self.hirelpp_exit_code) |
ioannis@53 | 942 | self._log_module_status('CloudScreen', self.cloudmask, self.cloudmask_exit_code) |
ioannis@53 | 943 | self._log_module_status('ElPP', self.elpp, self.elpp_exit_code) |
ioannis@53 | 944 | self._log_module_status('ELDA', self.elda, self.elda_exit_code) |
ioannis@53 | 945 | self._log_module_status('ELIC', self.elic, self.elic_exit_code) |
ioannis@53 | 946 | self._log_module_status('ELQuick', self.elquick, self.elquick_exit_code) |
ioannis@53 | 947 | |
ioannis@53 | 948 | def _log_module_status(self, name, status, exit_code): |
ioannis@53 | 949 | if exit_code: |
ioannis@54 | 950 | if exit_code['exit_code'] > 0: |
ioannis@54 | 951 | logger.warning("{0} exit code: {2[exit_code]} - {2[description]}".format(name, status, exit_code)) |
ioannis@54 | 952 | else: |
ioannis@54 | 953 | logger.info("{0} exit code: {2[exit_code]} - {2[description]}".format(name, status, exit_code)) |
ioannis@53 | 954 | else: |
ioannis@53 | 955 | logger.info("{0} exit code: {2}".format(name, status, exit_code)) |
ioannis@53 | 956 | |
ioannis@43 | 957 | @property |
ioannis@43 | 958 | def rerun_elda_url(self): |
ioannis@43 | 959 | url_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/rerun-elda/') |
ioannis@43 | 960 | return url_pattern.format(self.id) |
victor@7 | 961 | |
victor@7 | 962 | @property |
ioannis@43 | 963 | def rerun_elpp_url(self): |
ioannis@43 | 964 | url_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/rerun-elpp/') |
i@14 | 965 | return url_pattern.format(self.id) |
victor@7 | 966 | |
victor@7 | 967 | @property |
victor@7 | 968 | def rerun_all_url(self): |
i@14 | 969 | ulr_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/rerun-all/') |
i@14 | 970 | return ulr_pattern.format(self.id) |
victor@7 | 971 | |
victor@7 | 972 | def __str__(self): |
ioannis@43 | 973 | return "Measurement {}".format(self.id) |
victor@7 | 974 | |
victor@7 | 975 | |
i@38 | 976 | class AncillaryFile(ApiObject): |
i@38 | 977 | """ This class represents the ancilalry file object as returned in the SCC API. |
i@38 | 978 | """ |
ioannis@67 | 979 | |
i@38 | 980 | @property |
i@38 | 981 | def already_on_scc(self): |
i@38 | 982 | if self.exists is False: |
i@38 | 983 | return False |
i@38 | 984 | |
i@38 | 985 | return not self.status == 'missing' |
i@38 | 986 | |
i@38 | 987 | def __str__(self): |
i@38 | 988 | return "%s: %s, %s" % (self.id, |
i@38 | 989 | self.filename, |
i@38 | 990 | self.status) |
i@38 | 991 | |
i@38 | 992 | |
ioannis@67 | 993 | # Methods that use the SCC class to perform specific tasks. |
ioannis@45 | 994 | def process_file(filename, system_id, settings, force_upload, delete_related, |
claudio@69 | 995 | delay=0, cloudfree=False, monitor=True, rs_filename=None, lr_filename=None, ov_filename=None): |
ioannis@32 | 996 | """ Shortcut function to process a file to the SCC. """ |
ioannis@32 | 997 | logger.info("Processing file %s, using system %s" % (filename, system_id)) |
victor@7 | 998 | |
ioannis@43 | 999 | with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc: |
ioannis@43 | 1000 | scc.login(settings['website_credentials']) |
ioannis@45 | 1001 | measurement = scc.process(filename, system_id, |
ioannis@45 | 1002 | force_upload=force_upload, |
ioannis@45 | 1003 | delete_related=delete_related, |
ioannis@54 | 1004 | delay=delay, |
claudio@69 | 1005 | cloudfree=cloudfree, |
ioannis@45 | 1006 | monitor=monitor, |
ioannis@45 | 1007 | rs_filename=rs_filename, |
ioannis@45 | 1008 | lr_filename=lr_filename, |
ioannis@45 | 1009 | ov_filename=ov_filename) |
ioannis@43 | 1010 | scc.logout() |
victor@7 | 1011 | return measurement |
victor@7 | 1012 | |
victor@7 | 1013 | |
ioannis@45 | 1014 | def delete_measurements(measurement_ids, delete_related, settings): |
moritz@29 | 1015 | """ Shortcut function to delete measurements from the SCC. """ |
ioannis@45 | 1016 | with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc: |
ioannis@45 | 1017 | scc.login(settings['website_credentials']) |
ioannis@45 | 1018 | for m_id in measurement_ids: |
ioannis@54 | 1019 | logger.info("Deleting %s." % m_id) |
ioannis@45 | 1020 | scc.delete_measurement(m_id, delete_related) |
ioannis@45 | 1021 | scc.logout() |
moritz@29 | 1022 | |
moritz@29 | 1023 | |
moritz@29 | 1024 | def rerun_all(measurement_ids, monitor, settings): |
moritz@29 | 1025 | """ Shortcut function to rerun measurements from the SCC. """ |
i@14 | 1026 | |
ioannis@43 | 1027 | with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc: |
ioannis@43 | 1028 | scc.login(settings['website_credentials']) |
ioannis@45 | 1029 | for m_id in measurement_ids: |
ioannis@54 | 1030 | logger.info("Rerunning all products for %s." % m_id) |
ioannis@45 | 1031 | scc.rerun_all(m_id, monitor) |
ioannis@43 | 1032 | scc.logout() |
victor@7 | 1033 | |
victor@7 | 1034 | |
moritz@29 | 1035 | def rerun_processing(measurement_ids, monitor, settings): |
victor@7 | 1036 | """ Shortcut function to delete a measurement from the SCC. """ |
i@14 | 1037 | |
ioannis@43 | 1038 | with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc: |
ioannis@43 | 1039 | scc.login(settings['website_credentials']) |
ioannis@45 | 1040 | for m_id in measurement_ids: |
ioannis@45 | 1041 | logger.info("Rerunning (optical) processing for %s" % m_id) |
ioannis@45 | 1042 | scc.rerun_elpp(m_id, monitor) |
ioannis@45 | 1043 | scc.logout() |
moritz@29 | 1044 | |
moritz@29 | 1045 | |
claudio@68 | 1046 | def list_measurements(settings, id_exact=None, id_startswith=None, |
claudio@68 | 1047 | start_exact=None, start_gte=None, start_lte=None, |
claudio@68 | 1048 | stop_exact=None, stop_gte=None, stop_lte=None, |
claudio@68 | 1049 | station_exact=None, station_in=None): |
moritz@29 | 1050 | """List all available measurements""" |
ioannis@45 | 1051 | with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc: |
ioannis@45 | 1052 | scc.login(settings['website_credentials']) |
ioannis@65 | 1053 | |
claudio@68 | 1054 | results_json = scc.list_measurements(id_exact=id_exact, id_startswith=id_startswith, |
claudio@68 | 1055 | start_exact=start_exact, start_gte=start_gte, start_lte=start_lte, |
claudio@68 | 1056 | stop_exact=stop_exact, stop_gte=stop_gte, stop_lte=stop_lte, |
claudio@68 | 1057 | station_exact=station_exact, station_in=station_in) |
ioannis@65 | 1058 | print(results_json) |
ioannis@65 | 1059 | |
ioannis@43 | 1060 | scc.logout() |
victor@7 | 1061 | |
victor@7 | 1062 | |
ioannis@61 | 1063 | def download_measurements(measurement_ids, max_retries, exit_if_missing, settings): |
moritz@29 | 1064 | """Download all measurements for the specified IDs""" |
ioannis@43 | 1065 | with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc: |
ioannis@43 | 1066 | scc.login(settings['website_credentials']) |
ioannis@45 | 1067 | for m_id in measurement_ids: |
ioannis@61 | 1068 | scc.monitor_processing(m_id, retry_max=max_retries, time_sleep=3, exit_if_missing=exit_if_missing) |
ioannis@61 | 1069 | |
ioannis@43 | 1070 | scc.logout() |
i@14 | 1071 | |
i@14 | 1072 | |
claudio@68 | 1073 | def eshape_downloader(settings): |
claudio@68 | 1074 | eshape_dir = settings['output_dir'] |
claudio@68 | 1075 | #directories = [[x[0] for x in os.walk(eshape_dir)]] |
claudio@68 | 1076 | all_directories = [f.name for f in os.scandir(eshape_dir) if f.is_dir()] |
claudio@68 | 1077 | folders_pattern = re.compile("^(\d{6}_{1}\d{12})+$") |
claudio@68 | 1078 | directories = [dir_name for dir_name in all_directories if folders_pattern.match(dir_name)] |
claudio@68 | 1079 | for dir_name in directories: |
claudio@68 | 1080 | start_string = dir_name.split("_")[1] |
claudio@68 | 1081 | date_time_start = datetime.datetime.strptime(start_string, '%Y%m%d%H%M') |
claudio@68 | 1082 | if (datetime.datetime.now() - datetime.timedelta(days=3)) <= date_time_start: |
claudio@68 | 1083 | start_parameter = date_time_start.strftime("%Y-%m-%dT%H:%M:%S") |
claudio@68 | 1084 | date_time_stop = datetime.datetime.now() |
claudio@68 | 1085 | stop_parameter = date_time_stop.strftime("%Y-%m-%dT%H:%M:%S") |
claudio@68 | 1086 | |
claudio@68 | 1087 | with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc: |
claudio@68 | 1088 | scc.login(settings['website_credentials']) |
claudio@68 | 1089 | |
claudio@68 | 1090 | measurements = scc.available_measurements(start_gte=start_parameter, stop_lte=stop_parameter) |
claudio@68 | 1091 | if measurements is not None: |
claudio@68 | 1092 | for meas in measurements: |
claudio@68 | 1093 | scc.download_products(meas, dir_name) |
claudio@68 | 1094 | |
claudio@68 | 1095 | scc.logout() |
claudio@68 | 1096 | |
claudio@68 | 1097 | |
claudio@70 | 1098 | def automatic_upload(settings): |
claudio@71 | 1099 | date_time_start = datetime.datetime.now() - datetime.timedelta(hours=8) |
claudio@71 | 1100 | date_time_start = date_time_start.replace(minute=0, second=0) |
claudio@70 | 1101 | start_parameter = date_time_start.strftime("%Y-%m-%dT%H:%M:%S") |
claudio@71 | 1102 | date_time_stop = date_time_start + datetime.timedelta(hours=6) |
claudio@70 | 1103 | stop_parameter = date_time_stop.strftime("%Y-%m-%dT%H:%M:%S") |
claudio@70 | 1104 | |
claudio@71 | 1105 | # Altri esempi di utilizzo della libreria |
claudio@71 | 1106 | #date_time_start = datetime.datetime.combine(date_time_start, datetime.time.min) |
claudio@71 | 1107 | #date_time_stop = date_time_start.replace(hour=6, minute=0, second=0) |
claudio@71 | 1108 | #date_time_stop = datetime.datetime.combine(date_time_stop, datetime.time.max) |
claudio@71 | 1109 | |
claudio@71 | 1110 | with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url'], settings['sccdb_host'], settings['sccdb_credentials'], settings['sccdb']) as scc: |
claudio@70 | 1111 | scc.login(settings['website_credentials']) |
claudio@70 | 1112 | |
claudio@70 | 1113 | measurements = scc.available_measurements(start_gte=start_parameter, stop_lte=stop_parameter, is_queued=False, is_being_processed=False) |
claudio@71 | 1114 | is_automatic_upload = True |
claudio@70 | 1115 | if measurements is not None: |
claudio@70 | 1116 | for meas in measurements: |
claudio@71 | 1117 | #Create SQL query to compose |
claudio@71 | 1118 | mySql_insert_query = ("INSERT INTO scctoearlinettransfer_log (__measurements__ID, creation_date, " |
claudio@71 | 1119 | "hirelpp_transfer_status, hirelpp_transfer_message, cloudmask_transfer_status, " |
claudio@71 | 1120 | "cloudmask_transfer_message, elpp_transfer_status, elpp_transfer_message, " |
claudio@71 | 1121 | "elda_transfer_status, elda_transfer_message, eldamwl_transfer_status, " |
claudio@71 | 1122 | "eldamwl_transfer_message, eldec_transfer_status, eldec_transfer_message, " |
claudio@71 | 1123 | "elic_transfer_status, elic_transfer_message, transfer_status, transfer_message) " |
claudio@71 | 1124 | "VALUES " |
claudio@71 | 1125 | "(#__measurements__ID#, #creation_date# , #hirelpp_transfer_status#, " |
claudio@71 | 1126 | "#hirelpp_transfer_message# , #cloudmask_transfer_status#, #cloudmask_transfer_message#, " |
claudio@71 | 1127 | "#elpp_transfer_status#, #elpp_transfer_message#, #elda_transfer_status#, " |
claudio@71 | 1128 | "#elda_transfer_message#, #eldamwl_transfer_status#, #eldamwl_transfer_message#, " |
claudio@71 | 1129 | "#eldec_transfer_status#, #eldec_transfer_message#, #elic_transfer_status#, " |
claudio@71 | 1130 | "#elic_transfer_message#, #transfer_status#, #transfer_message#);") |
claudio@71 | 1131 | scc.sccquery = mySql_insert_query |
claudio@71 | 1132 | scc.download_products(meas, "", is_automatic_upload) |
claudio@71 | 1133 | |
claudio@71 | 1134 | scc.sccquery = scc.sccquery.replace("#creation_date#", "'"+datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")+"'") |
claudio@71 | 1135 | scc.sccquery = scc.sccquery.replace("#eldamwl_transfer_status#", "NULL") |
claudio@71 | 1136 | scc.sccquery = scc.sccquery.replace("#eldamwl_transfer_message#", "'No ELDAMWL products available'") |
claudio@71 | 1137 | scc.sccquery = scc.sccquery.replace("#transfer_status#", "true") |
claudio@71 | 1138 | scc.sccquery = scc.sccquery.replace("#transfer_message#", "NULL") |
claudio@71 | 1139 | |
claudio@71 | 1140 | #Execute SQL query |
claudio@71 | 1141 | try: |
claudio@71 | 1142 | connection = mysql.connector.connect(host=scc.sccdb_host, |
claudio@71 | 1143 | database=scc.sccdb, |
claudio@71 | 1144 | user=scc.sccdb_credentials[0], |
claudio@71 | 1145 | password=scc.sccdb_credentials[1]) |
claudio@71 | 1146 | |
claudio@71 | 1147 | cursor = connection.cursor() |
claudio@71 | 1148 | cursor.execute(scc.sccquery) |
claudio@71 | 1149 | connection.commit() |
claudio@71 | 1150 | logger.info("{} Record inserted successfully into scctoearlinettransfer_log table".format(cursor.rowcount)) |
claudio@71 | 1151 | cursor.close() |
claudio@71 | 1152 | |
claudio@71 | 1153 | except mysql.connector.Error as error: |
claudio@71 | 1154 | logger.error("Failed to insert record into scctoearlinettransfer_log table {}".format(error)) |
claudio@71 | 1155 | |
claudio@71 | 1156 | finally: |
claudio@71 | 1157 | if connection.is_connected(): |
claudio@71 | 1158 | connection.close() |
claudio@71 | 1159 | logger.debug("MySQL connection is closed") |
claudio@70 | 1160 | |
claudio@70 | 1161 | scc.logout() |
claudio@70 | 1162 | |
claudio@70 | 1163 | |
moritz@29 | 1164 | def settings_from_path(config_file_path): |
i@14 | 1165 | """ Read the configuration file. |
i@14 | 1166 | |
i@14 | 1167 | The file should be in YAML syntax.""" |
i@14 | 1168 | |
i@14 | 1169 | if not os.path.isfile(config_file_path): |
moritz@29 | 1170 | raise argparse.ArgumentTypeError("Wrong path for configuration file (%s)" % config_file_path) |
i@14 | 1171 | |
i@14 | 1172 | with open(config_file_path) as yaml_file: |
i@14 | 1173 | try: |
i@14 | 1174 | settings = yaml.safe_load(yaml_file) |
i@14 | 1175 | logger.debug("Read settings file(%s)" % config_file_path) |
moritz@29 | 1176 | except Exception: |
moritz@29 | 1177 | raise argparse.ArgumentTypeError("Could not parse YAML file (%s)" % config_file_path) |
i@14 | 1178 | |
i@14 | 1179 | # YAML limitation: does not read tuples |
i@14 | 1180 | settings['basic_credentials'] = tuple(settings['basic_credentials']) |
i@14 | 1181 | settings['website_credentials'] = tuple(settings['website_credentials']) |
i@14 | 1182 | return settings |
i@14 | 1183 | |
i@14 | 1184 | |
moritz@29 | 1185 | # Setup for command specific parsers |
moritz@29 | 1186 | def setup_delete(parser): |
moritz@29 | 1187 | def delete_from_args(parsed): |
ioannis@45 | 1188 | delete_measurements(parsed.IDs, |
ioannis@45 | 1189 | delete_related=False, |
ioannis@45 | 1190 | settings=parsed.config) |
moritz@29 | 1191 | |
moritz@29 | 1192 | parser.add_argument("IDs", nargs="+", help="measurement IDs to delete.") |
moritz@29 | 1193 | parser.set_defaults(execute=delete_from_args) |
moritz@29 | 1194 | |
moritz@29 | 1195 | |
moritz@29 | 1196 | def setup_rerun_all(parser): |
moritz@29 | 1197 | def rerun_all_from_args(parsed): |
moritz@29 | 1198 | rerun_all(parsed.IDs, parsed.process, parsed.config) |
moritz@29 | 1199 | |
moritz@29 | 1200 | parser.add_argument("IDs", nargs="+", help="Measurement IDs to rerun.") |
moritz@29 | 1201 | parser.add_argument("-p", "--process", help="Wait for the results of the processing.", |
moritz@29 | 1202 | action="store_true") |
moritz@29 | 1203 | parser.set_defaults(execute=rerun_all_from_args) |
moritz@29 | 1204 | |
moritz@29 | 1205 | |
ioannis@45 | 1206 | def setup_rerun_elpp(parser): |
moritz@29 | 1207 | def rerun_processing_from_args(parsed): |
moritz@29 | 1208 | rerun_processing(parsed.IDs, parsed.process, parsed.config) |
moritz@29 | 1209 | |
moritz@29 | 1210 | parser.add_argument("IDs", nargs="+", help="Measurement IDs to rerun the processing on.") |
moritz@29 | 1211 | parser.add_argument("-p", "--process", help="Wait for the results of the processing.", |
moritz@29 | 1212 | action="store_true") |
moritz@29 | 1213 | parser.set_defaults(execute=rerun_processing_from_args) |
moritz@29 | 1214 | |
moritz@29 | 1215 | |
ioannis@45 | 1216 | def setup_upload_file(parser): |
ioannis@45 | 1217 | """ Upload but do not monitor processing progress. """ |
ioannis@67 | 1218 | |
ioannis@45 | 1219 | def upload_file_from_args(parsed): |
ioannis@45 | 1220 | process_file(parsed.filename, parsed.system, parsed.config, |
ioannis@54 | 1221 | delay=parsed.delay, |
claudio@69 | 1222 | cloudfree=parsed.cloudfree, |
ioannis@45 | 1223 | monitor=parsed.process, |
ioannis@45 | 1224 | force_upload=parsed.force_upload, |
ioannis@45 | 1225 | delete_related=False, # For now, use this as default |
ioannis@32 | 1226 | rs_filename=parsed.radiosounding, |
ioannis@32 | 1227 | ov_filename=parsed.overlap, |
ioannis@32 | 1228 | lr_filename=parsed.lidarratio) |
moritz@29 | 1229 | |
ioannis@54 | 1230 | def delay(arg): |
ioannis@54 | 1231 | try: |
ioannis@54 | 1232 | int_arg = int(arg) |
ioannis@54 | 1233 | except ValueError: |
ioannis@54 | 1234 | raise argparse.ArgumentTypeError("Could not convert delay argument {} to integer.".format(arg)) |
ioannis@54 | 1235 | |
ioannis@54 | 1236 | if 0 <= int_arg <= 96: |
ioannis@54 | 1237 | return int_arg |
ioannis@54 | 1238 | else: |
ioannis@54 | 1239 | raise argparse.ArgumentTypeError("Delay should be an integer between 0 and 96.") |
ioannis@54 | 1240 | |
moritz@29 | 1241 | parser.add_argument("filename", help="Measurement file name or path.") |
moritz@29 | 1242 | parser.add_argument("system", help="Processing system id.") |
ioannis@54 | 1243 | parser.add_argument("--delay", help="Delay processing by the specified number of hours (0 to 96).", |
ioannis@54 | 1244 | default=0, type=delay) |
claudio@69 | 1245 | parser.add_argument("--cloudfree", help="Manually assume this measurement as cloud free.", |
claudio@69 | 1246 | action="store_true", default=False) |
ioannis@45 | 1247 | parser.add_argument("-p", "--process", help="Wait for the processing results.", |
ioannis@45 | 1248 | action="store_true") |
ioannis@45 | 1249 | parser.add_argument("--force_upload", help="If measurement ID exists on SCC, delete before uploading.", |
ioannis@45 | 1250 | action="store_true") |
i@30 | 1251 | parser.add_argument("--radiosounding", default=None, help="Radiosounding file name or path") |
ioannis@32 | 1252 | parser.add_argument("--overlap", default=None, help="Overlap file name or path") |
ioannis@32 | 1253 | parser.add_argument("--lidarratio", default=None, help="Lidar ratio file name or path") |
i@30 | 1254 | |
moritz@29 | 1255 | parser.set_defaults(execute=upload_file_from_args) |
moritz@29 | 1256 | |
moritz@29 | 1257 | |
moritz@29 | 1258 | def setup_list_measurements(parser): |
moritz@29 | 1259 | def list_measurements_from_args(parsed): |
claudio@68 | 1260 | list_measurements(parsed.config, id_exact=parsed.id_exact, id_startswith=parsed.id_startswith, |
claudio@68 | 1261 | start_exact=parsed.start_exact, start_gte=parsed.start_gte, start_lte=parsed.start_lte, |
claudio@68 | 1262 | stop_exact=parsed.stop_exact, stop_gte=parsed.stop_gte, stop_lte=parsed.stop_lte, |
claudio@68 | 1263 | station_exact=parsed.station_exact, station_in=parsed.station_in) |
claudio@68 | 1264 | |
claudio@68 | 1265 | group = parser.add_argument_group() |
claudio@68 | 1266 | |
claudio@68 | 1267 | group_id = group.add_mutually_exclusive_group() |
claudio@68 | 1268 | group_id.add_argument("--id_exact", help="Exact measurement id.") |
claudio@68 | 1269 | group_id.add_argument("--id_startswith", help="Initial part of measurement id.") |
moritz@29 | 1270 | |
claudio@68 | 1271 | group_start = group.add_argument_group() |
claudio@68 | 1272 | group_start.add_argument("--start_exact", help="Exact start date of the measurement.") |
claudio@68 | 1273 | group_start.add_argument("--start_gte", help="Start date of the measurement after the given date.") |
claudio@68 | 1274 | group_start.add_argument("--start_lte", help="Start date of the measurement before the given date.") |
claudio@68 | 1275 | |
claudio@68 | 1276 | group_stop = group.add_argument_group() |
claudio@68 | 1277 | group_stop.add_argument("--stop_exact", help="Exact stop date of the measurement.") |
claudio@68 | 1278 | group_stop.add_argument("--stop_gte", help="Stop date of the measurement after the given date.") |
claudio@68 | 1279 | group_stop.add_argument("--stop_lte", help="Stop date of the measurement before the given date.") |
claudio@68 | 1280 | |
claudio@68 | 1281 | group_station = group.add_mutually_exclusive_group() |
claudio@68 | 1282 | group_station.add_argument("--station_exact", help="Station the performed the measurement.") |
claudio@68 | 1283 | group_station.add_argument("--station_in", help="List of stations (separated by comma) that performed the measurement.") |
moritz@29 | 1284 | |
moritz@29 | 1285 | parser.set_defaults(execute=list_measurements_from_args) |
moritz@29 | 1286 | |
moritz@29 | 1287 | |
moritz@29 | 1288 | def setup_download_measurements(parser): |
moritz@29 | 1289 | def download_measurements_from_args(parsed): |
ioannis@61 | 1290 | download_measurements(parsed.IDs, parsed.max_retries, parsed.ignore_errors, parsed.config) |
moritz@29 | 1291 | |
moritz@29 | 1292 | parser.add_argument("IDs", help="Measurement IDs that should be downloaded.", nargs="+") |
ioannis@67 | 1293 | parser.add_argument("--max_retries", help="Number of times to retry in cases of missing measurement id.", default=0, |
ioannis@67 | 1294 | type=int) |
ioannis@67 | 1295 | parser.add_argument("--ignore_errors", help="Ignore errors when downloading multiple measurements.", |
ioannis@67 | 1296 | action="store_false") |
moritz@29 | 1297 | parser.set_defaults(execute=download_measurements_from_args) |
moritz@29 | 1298 | |
moritz@29 | 1299 | |
claudio@68 | 1300 | def setup_eshape_downloader(parser): |
claudio@68 | 1301 | def run_eshape_downloader(parsed): |
claudio@68 | 1302 | eshape_downloader(parsed.config) |
claudio@68 | 1303 | |
claudio@68 | 1304 | parser.set_defaults(execute=run_eshape_downloader) |
claudio@68 | 1305 | |
claudio@68 | 1306 | |
claudio@70 | 1307 | def setup_automatic_upload(parser): |
claudio@70 | 1308 | def run_automatic_upload(parsed): |
claudio@70 | 1309 | automatic_upload(parsed.config) |
claudio@70 | 1310 | |
claudio@70 | 1311 | parser.set_defaults(execute=run_automatic_upload) |
claudio@70 | 1312 | |
claudio@70 | 1313 | |
victor@7 | 1314 | def main(): |
victor@7 | 1315 | # Define the command line arguments. |
victor@7 | 1316 | parser = argparse.ArgumentParser() |
moritz@29 | 1317 | subparsers = parser.add_subparsers() |
moritz@29 | 1318 | |
moritz@29 | 1319 | delete_parser = subparsers.add_parser("delete", help="Deletes a measurement.") |
ioannis@67 | 1320 | rerun_all_parser = subparsers.add_parser("rerun-all", |
ioannis@67 | 1321 | help="Rerun all processing steps for the provided measurement IDs.") |
ioannis@45 | 1322 | rerun_processing_parser = subparsers.add_parser("rerun-elpp", |
ioannis@45 | 1323 | help="Rerun low-resolution processing steps for the provided measurement ID.") |
ioannis@67 | 1324 | upload_file_parser = subparsers.add_parser("upload-file", |
ioannis@67 | 1325 | help="Submit a file and, optionally, download the output products.") |
ioannis@32 | 1326 | list_parser = subparsers.add_parser("list", help="List measurements registered on the SCC.") |
moritz@29 | 1327 | download_parser = subparsers.add_parser("download", help="Download selected measurements.") |
claudio@68 | 1328 | eshape_parser = subparsers.add_parser("eshape-downloader", help="Search and download relevant products for E-SHAPE.") |
claudio@70 | 1329 | automatic_upload_parser = subparsers.add_parser("automatic-upload", help="Select and download products available for the upload to EARLINET db.") |
moritz@29 | 1330 | |
moritz@29 | 1331 | setup_delete(delete_parser) |
moritz@29 | 1332 | setup_rerun_all(rerun_all_parser) |
ioannis@45 | 1333 | setup_rerun_elpp(rerun_processing_parser) |
ioannis@45 | 1334 | |
moritz@29 | 1335 | setup_upload_file(upload_file_parser) |
moritz@29 | 1336 | setup_list_measurements(list_parser) |
moritz@29 | 1337 | setup_download_measurements(download_parser) |
claudio@68 | 1338 | setup_eshape_downloader(eshape_parser) |
claudio@70 | 1339 | setup_automatic_upload(automatic_upload_parser) |
victor@7 | 1340 | |
victor@7 | 1341 | # Verbosity settings from http://stackoverflow.com/a/20663028 |
victor@7 | 1342 | parser.add_argument('-d', '--debug', help="Print debugging information.", action="store_const", |
victor@7 | 1343 | dest="loglevel", const=logging.DEBUG, default=logging.INFO, |
victor@7 | 1344 | ) |
victor@7 | 1345 | parser.add_argument('-s', '--silent', help="Show only warning and error messages.", action="store_const", |
victor@7 | 1346 | dest="loglevel", const=logging.WARNING |
victor@7 | 1347 | ) |
victor@7 | 1348 | |
ioannis@32 | 1349 | # Setup default config location |
moritz@29 | 1350 | home = os.path.expanduser("~") |
moritz@29 | 1351 | default_config_location = os.path.abspath(os.path.join(home, ".scc_access.yaml")) |
moritz@29 | 1352 | parser.add_argument("-c", "--config", help="Path to the config file.", type=settings_from_path, |
moritz@29 | 1353 | default=default_config_location) |
victor@7 | 1354 | |
victor@7 | 1355 | args = parser.parse_args() |
ioannis@43 | 1356 | |
victor@7 | 1357 | # Get the logger with the appropriate level |
victor@7 | 1358 | logging.basicConfig(format='%(levelname)s: %(message)s', level=args.loglevel) |
victor@7 | 1359 | |
moritz@29 | 1360 | # Dispatch to appropriate function |
moritz@29 | 1361 | args.execute(args) |
moritz@29 | 1362 | |
i@14 | 1363 | |
moritz@29 | 1364 | # When running through terminal |
moritz@29 | 1365 | if __name__ == '__main__': |
moritz@29 | 1366 | main() |