" # {12, 15} to handle both old- and new-style measurement ids.
victor@7:
victor@7:
victor@7: class SCC:
madrouin@24: """A simple class that will attempt to upload a file on the SCC server.
i@14:
victor@7: The uploading is done by simulating a normal browser session. In the current
madrouin@20: version no check is performed, and no feedback is given if the upload
madrouin@20: was successful. If everything is setup correctly, it will work.
victor@7: """
victor@7:
i@14: def __init__(self, auth, output_dir, base_url):
moritz@29:
victor@7: self.auth = auth
victor@7: self.output_dir = output_dir
i@14: self.base_url = base_url
victor@7: self.session = requests.Session()
moritz@29: self.session.auth = auth
moritz@29: self.session.verify = False
victor@7:
i@14: self.login_url = urlparse.urljoin(self.base_url, 'accounts/login/')
i@31: self.logout_url = urlparse.urljoin(self.base_url, 'accounts/logout/')
i@31: self.list_measurements_url = urlparse.urljoin(self.base_url, 'data_processing/measurements/')
i@31:
i@14: self.upload_url = urlparse.urljoin(self.base_url, 'data_processing/measurements/quick/')
ioannis@54: self.measurement_page_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/')
ioannis@43: self.download_hirelpp_pattern = urlparse.urljoin(self.base_url,
ioannis@34: 'data_processing/measurements/{0}/download-hirelpp/')
ioannis@43: self.download_cloudmask_pattern = urlparse.urljoin(self.base_url,
ioannis@34: 'data_processing/measurements/{0}/download-cloudmask/')
ioannis@34:
ioannis@43: self.download_elpp_pattern = urlparse.urljoin(self.base_url,
moritz@29: 'data_processing/measurements/{0}/download-preprocessed/')
ioannis@43: self.download_elda_pattern = urlparse.urljoin(self.base_url,
moritz@29: 'data_processing/measurements/{0}/download-optical/')
ioannis@45: self.download_plots_pattern = urlparse.urljoin(self.base_url,
moritz@29: 'data_processing/measurements/{0}/download-plots/')
ioannis@43: self.download_elic_pattern = urlparse.urljoin(self.base_url,
ioannis@34: 'data_processing/measurements/{0}/download-elic/')
i@14: self.delete_measurement_pattern = urlparse.urljoin(self.base_url, 'admin/database/measurements/{0}/delete/')
i@31:
i@14: self.api_base_url = urlparse.urljoin(self.base_url, 'api/v1/')
i@31: self.api_measurement_pattern = urlparse.urljoin(self.api_base_url, 'measurements/{0}/')
i@31: self.api_measurements_url = urlparse.urljoin(self.api_base_url, 'measurements')
i@38: self.api_sounding_search_pattern = urlparse.urljoin(self.api_base_url, 'sounding_files/?filename={0}')
i@38: self.api_lidarratio_search_pattern = urlparse.urljoin(self.api_base_url, 'lidarratio_files/?filename={0}')
i@38: self.api_overlap_search_pattern = urlparse.urljoin(self.api_base_url, 'overlap_files/?filename={0}')
ioannis@43:
i@14: def login(self, credentials):
i@31: """ Login to SCC. """
victor@7: logger.debug("Attempting to login to SCC, username %s." % credentials[0])
moritz@29: login_credentials = {'username': credentials[0],
moritz@29: 'password': credentials[1]}
victor@7:
i@14: logger.debug("Accessing login page at %s." % self.login_url)
victor@7:
victor@7: # Get upload form
moritz@29: login_page = self.session.get(self.login_url)
i@14:
i@31: if not login_page.ok:
i@31: raise self.PageNotAccessibleError('Could not access login pages. Status code %s' % login_page.status_code)
victor@7:
i@31: logger.debug("Submitting credentials.")
victor@7: # Submit the login data
i@14: login_submit = self.session.post(self.login_url,
moritz@29: data=login_credentials,
victor@7: headers={'X-CSRFToken': login_page.cookies['csrftoken'],
moritz@29: 'referer': self.login_url})
victor@7: return login_submit
victor@7:
victor@7: def logout(self):
i@31: """ Logout from SCC """
i@31: return self.session.get(self.logout_url, stream=True)
victor@7:
ioannis@54: def upload_file(self, filename, system_id, force_upload, delete_related, delay=0, rs_filename=None, ov_filename=None, lr_filename=None):
madrouin@20: """ Upload a filename for processing with a specific system. If the
victor@7: upload is successful, it returns the measurement id. """
ioannis@43: measurement_id = self.measurement_id_from_file(filename)
ioannis@43:
ioannis@43: logger.debug('Checking if a measurement with the same id already exists on the SCC server.')
ioannis@50: existing_measurement, _ = self.get_measurement(measurement_id)
ioannis@43:
ioannis@43: if existing_measurement:
ioannis@43: if force_upload:
ioannis@43: logger.info(
ioannis@43: "Measurement with id {} already exists on the SCC. Trying to delete it...".format(measurement_id))
ioannis@43: self.delete_measurement(measurement_id, delete_related)
ioannis@43: else:
ioannis@43: logger.error(
ioannis@43: "Measurement with id {} already exists on the SCC. Use --force_upload flag to overwrite it.".format(
ioannis@43: measurement_id))
ioannis@43: sys.exit(1)
ioannis@43:
victor@7: # Get submit page
moritz@29: upload_page = self.session.get(self.upload_url)
victor@7:
victor@7: # Submit the data
ioannis@54: upload_data = {'system': system_id,
ioannis@54: 'delay': delay}
ioannis@54:
ioannis@54: logger.debug("Submitted processing parameters - System: {}, Delay: {}".format(system_id, delay))
ioannis@54:
victor@7: files = {'data': open(filename, 'rb')}
victor@7:
madrouin@20: if rs_filename is not None:
i@38: ancillary_file, _ = self.get_ancillary(rs_filename, 'sounding')
i@38:
i@38: if ancillary_file.already_on_scc:
i@38: logger.warning("Sounding file {0.filename} already on the SCC with id {0.id}. Ignoring it.".format(ancillary_file))
i@38: else:
i@38: logger.debug('Adding sounding file %s' % rs_filename)
i@38: files['sounding_file'] = open(rs_filename, 'rb')
madrouin@20:
ioannis@32: if ov_filename is not None:
i@38: ancillary_file, _ = self.get_ancillary(ov_filename, 'overlap')
i@38:
i@38: if ancillary_file.already_on_scc:
i@38: logger.warning("Overlap file {0.filename} already on the SCC with id {0.id}. Ignoring it.".format(ancillary_file))
i@38: else:
i@38: logger.debug('Adding overlap file %s' % ov_filename)
i@38: files['overlap_file'] = open(ov_filename, 'rb')
i@31:
i@31: if lr_filename is not None:
i@38: ancillary_file, _ = self.get_ancillary(lr_filename, 'lidarratio')
i@38:
i@38: if ancillary_file.already_on_scc:
i@38: logger.warning(
i@38: "Lidar ratio file {0.filename} already on the SCC with id {0.id}. Ignoring it.".format(ancillary_file))
i@38: else:
i@38: logger.debug('Adding lidar ratio file %s' % lr_filename)
i@38: files['lidar_ratio_file'] = open(lr_filename, 'rb')
i@31:
i@14: logger.info("Uploading of file %s started." % filename)
victor@7:
i@14: upload_submit = self.session.post(self.upload_url,
victor@7: data=upload_data,
victor@7: files=files,
victor@7: headers={'X-CSRFToken': upload_page.cookies['csrftoken'],
moritz@29: 'referer': self.upload_url})
victor@7:
victor@7: if upload_submit.status_code != 200:
i@14: logger.warning("Connection error. Status code: %s" % upload_submit.status_code)
victor@7: return False
victor@7:
victor@7: # Check if there was a redirect to a new page.
i@14: if upload_submit.url == self.upload_url:
victor@7: measurement_id = False
ioannis@32: logger.error("Uploaded file(s) rejected! Try to upload manually to see the error.")
victor@7: else:
victor@7: measurement_id = re.findall(regex, upload_submit.text)[0]
ioannis@43: logger.info("Successfully uploaded measurement with id %s." % measurement_id)
ioannis@54: logger.info("You can monitor the processing progress online: {}".format(self.measurement_page_pattern.format(measurement_id)))
ioannis@43: return measurement_id
ioannis@43:
ioannis@43: @staticmethod
ioannis@43: def measurement_id_from_file(filename):
ioannis@43: """ Get the measurement id from the input file. """
ioannis@43:
ioannis@43: if not os.path.isfile(filename):
ioannis@43: logger.error("File {} does not exist.".format(filename))
ioannis@43: sys.exit(1)
ioannis@43:
ioannis@43: with netcdf.Dataset(filename) as f:
ioannis@43: try:
ioannis@43: measurement_id = f.Measurement_ID
ioannis@43: except AttributeError:
ioannis@43: logger.error(
ioannis@43: "Input file {} does not contain a Measurement_ID global attribute. Wrong file format?".format(
ioannis@43: filename))
ioannis@43: sys.exit(1)
victor@7:
victor@7: return measurement_id
victor@7:
victor@7: def download_files(self, measurement_id, subdir, download_url):
victor@7: """ Downloads some files from the download_url to the specified
madrouin@20: subdir. This method is used to download preprocessed file, optical
victor@7: files etc.
victor@7: """
ioannis@32: # TODO: Make downloading more robust (e.g. in case that files do not exist on server).
victor@7: # Get the file
moritz@29: request = self.session.get(download_url, stream=True)
moritz@29:
moritz@29: if not request.ok:
moritz@29: raise Exception("Could not download files for measurement '%s'" % measurement_id)
victor@7:
victor@7: # Create the dir if it does not exist
victor@7: local_dir = os.path.join(self.output_dir, measurement_id, subdir)
victor@7: if not os.path.exists(local_dir):
victor@7: os.makedirs(local_dir)
victor@7:
victor@7: # Save the file by chunk, needed if the file is big.
ioannis@43: memory_file = BytesIO()
victor@7:
victor@7: for chunk in request.iter_content(chunk_size=1024):
victor@7: if chunk: # filter out keep-alive new chunks
victor@7: memory_file.write(chunk)
victor@7: memory_file.flush()
victor@7:
victor@7: zip_file = ZipFile(memory_file)
victor@7:
victor@7: for ziped_name in zip_file.namelist():
victor@7: basename = os.path.basename(ziped_name)
victor@7:
victor@7: local_file = os.path.join(local_dir, basename)
victor@7:
victor@7: with open(local_file, 'wb') as f:
victor@7: f.write(zip_file.read(ziped_name))
victor@7:
ioannis@43: def download_hirelpp(self, measurement_id):
ioannis@43: """ Download hirelpp files for the measurement id. """
ioannis@43: # Construct the download url
ioannis@43: download_url = self.download_hirelpp_pattern.format(measurement_id)
ioannis@43: try:
ioannis@50: self.download_files(measurement_id, 'hirelpp', download_url)
ioannis@43: except Exception as e:
ioannis@43: logger.error("Could not download HiRElPP files. Error message: {}".format(e))
ioannis@43: logger.debug('Download exception:', exc_info=True)
ioannis@43:
ioannis@43: def download_cloudmask(self, measurement_id):
ioannis@43: """ Download cloudmask files for the measurement id. """
ioannis@43: # Construct the download url
ioannis@43: download_url = self.download_cloudmask_pattern.format(measurement_id)
ioannis@43: try:
ioannis@50: self.download_files(measurement_id, 'cloudscreen', download_url)
ioannis@43: except Exception as e:
ioannis@43: logger.error("Could not download cloudscreen files. Error message: {}".format(e))
ioannis@43: logger.debug('Download exception:', exc_info=True)
ioannis@43:
ioannis@43: def download_elpp(self, measurement_id):
victor@7: """ Download preprocessed files for the measurement id. """
victor@7: # Construct the download url
ioannis@43: download_url = self.download_elpp_pattern.format(measurement_id)
ioannis@43: try:
ioannis@50: self.download_files(measurement_id, 'elpp', download_url)
ioannis@43: except Exception as e:
ioannis@43: logger.error("Could not download ElPP files. Error message: {}".format(e))
ioannis@43: logger.debug('Download exception:', exc_info=True)
victor@7:
ioannis@43: def download_elda(self, measurement_id):
victor@7: """ Download optical files for the measurement id. """
victor@7: # Construct the download url
ioannis@43: download_url = self.download_elda_pattern.format(measurement_id)
ioannis@43: try:
ioannis@50: self.download_files(measurement_id, 'elda', download_url)
ioannis@43: except Exception as e:
ioannis@43: logger.error("Could not download ELDA files. Error message: {}".format(e))
ioannis@43: logger.debug('Download exception:', exc_info=True)
victor@7:
ioannis@43: def download_plots(self, measurement_id):
victor@7: """ Download profile graphs for the measurement id. """
victor@7: # Construct the download url
ioannis@45: download_url = self.download_plots_pattern.format(measurement_id)
ioannis@43: try:
ioannis@50: self.download_files(measurement_id, 'elda_plots', download_url)
ioannis@43: except Exception as e:
ioannis@43: logger.error("Could not download ELDA plots. Error message: {}".format(e))
ioannis@43: logger.debug('Download exception:', exc_info=True)
victor@7:
ioannis@43: def download_elic(self, measurement_id):
ioannis@43: """ Download ELIC files for the measurement id. """
ioannis@43: # Construct the download url
ioannis@43: download_url = self.download_elic_pattern.format(measurement_id)
ioannis@43: try:
ioannis@50: self.download_files(measurement_id, 'elic', download_url)
ioannis@43: except Exception as e:
ioannis@43: logger.error("Could not download ELIC files. Error message: {}".format(e))
ioannis@43: logger.debug('Download exception:', exc_info=True)
ioannis@43:
ioannis@43: def download_eldec(self, measurement_id):
ioannis@43: """ Download ELDEC files for the measurement id. """
ioannis@43: # Construct the download url
ioannis@43: download_url = self.download_elda_pattern.format(measurement_id) # ELDA patter is used for now
ioannis@43: try:
ioannis@50: self.download_files(measurement_id, 'eldec', download_url)
ioannis@43: except Exception as e:
ioannis@43: logger.error("Could not download EDELC files. Error message: {}".format(e))
ioannis@43: logger.debug('Download exception:', exc_info=True)
ioannis@43:
ioannis@43: def rerun_elpp(self, measurement_id, monitor=True):
ioannis@45: logger.debug("Started rerun_elpp procedure.")
ioannis@45:
ioannis@45: logger.debug("Getting measurement %s" % measurement_id)
madrouin@26: measurement, status = self.get_measurement(measurement_id)
victor@7:
victor@7: if measurement:
ioannis@45: logger.debug("Attempting to rerun ElPP through %s." % measurement.rerun_all_url)
ioannis@45: request = self.session.get(measurement.rerun_elpp_url, stream=True)
victor@7:
victor@7: if request.status_code != 200:
i@14: logger.error(
i@14: "Could not rerun processing for %s. Status code: %s" % (measurement_id, request.status_code))
ioannis@45: else:
ioannis@45: logger.info("Rerun-elpp command submitted successfully for id {}.".format(measurement_id))
victor@7:
victor@7: if monitor:
victor@7: self.monitor_processing(measurement_id)
victor@7:
victor@7: def rerun_all(self, measurement_id, monitor=True):
victor@7: logger.debug("Started rerun_all procedure.")
victor@7:
victor@7: logger.debug("Getting measurement %s" % measurement_id)
madrouin@26: measurement, status = self.get_measurement(measurement_id)
victor@7:
victor@7: if measurement:
victor@7: logger.debug("Attempting to rerun all processing through %s." % measurement.rerun_all_url)
victor@7:
moritz@29: request = self.session.get(measurement.rerun_all_url, stream=True)
victor@7:
victor@7: if request.status_code != 200:
victor@7: logger.error("Could not rerun pre processing for %s. Status code: %s" %
victor@7: (measurement_id, request.status_code))
ioannis@45: else:
ioannis@45: logger.info("Rerun-all command submitted successfully for id {}.".format(measurement_id))
victor@7:
victor@7: if monitor:
victor@7: self.monitor_processing(measurement_id)
victor@7:
ioannis@54: def process(self, filename, system_id, monitor, force_upload, delete_related, delay=0, rs_filename=None, lr_filename=None, ov_filename=None):
victor@7: """ Upload a file for processing and wait for the processing to finish.
victor@7: If the processing is successful, it will download all produced files.
victor@7: """
victor@7: logger.info("--- Processing started on %s. ---" % datetime.datetime.now())
victor@7: # Upload file
ioannis@50: logger.info("Uploading file.")
ioannis@45: measurement_id = self.upload_file(filename, system_id, force_upload, delete_related,
ioannis@54: delay=delay,
ioannis@32: rs_filename=rs_filename,
ioannis@32: lr_filename=lr_filename,
ioannis@32: ov_filename=ov_filename)
ioannis@43:
ioannis@54: if monitor and (delay > 0):
ioannis@54: logger.warning("Will not start monitoring, since a delay was specified: {} hours.".format(delay))
ioannis@54: return None
ioannis@54:
ioannis@32: if measurement_id and monitor:
ioannis@54: logger.info("Monitoring processing.")
moritz@29: return self.monitor_processing(measurement_id)
victor@7:
moritz@29: return None
victor@7:
victor@7: def monitor_processing(self, measurement_id):
victor@7: """ Monitor the processing progress of a measurement id"""
victor@7:
madrouin@24: # try to deal with error 404
madrouin@24: error_count = 0
ioannis@53: error_max = 3
ioannis@53: time_sleep = 3
madrouin@24:
madrouin@25: # try to wait for measurement to appear in API
madrouin@24: measurement = None
ioannis@50: logger.info("Looking for measurement %s on the SCC.", measurement_id)
madrouin@26: while error_count < error_max:
madrouin@26: time.sleep(time_sleep)
madrouin@26: measurement, status = self.get_measurement(measurement_id)
madrouin@26: if status != 200 and error_count < error_max:
ioannis@32: logger.error("Measurement not found. waiting %ds", time_sleep)
madrouin@24: error_count += 1
madrouin@26: else:
madrouin@26: break
madrouin@24:
madrouin@24: if error_count == error_max:
ioannis@32: logger.critical("Measurement %s doesn't seem to exist", measurement_id)
madrouin@26: sys.exit(1)
madrouin@26:
ioannis@45: logger.info('Measurement %s found.', measurement_id)
madrouin@24:
victor@7: if measurement is not None:
victor@7: while measurement.is_running:
ioannis@53: measurement.log_processing_status()
victor@7: time.sleep(10)
madrouin@26: measurement, status = self.get_measurement(measurement_id)
madrouin@24:
ioannis@34: logger.info("Measurement processing finished.")
ioannis@53: measurement.log_detailed_status()
ioannis@53:
ioannis@43: if measurement.hirelpp == 127:
ioannis@43: logger.info("Downloading HiRElPP files.")
ioannis@43: self.download_hirelpp(measurement_id)
ioannis@43: if measurement.cloudmask == 127:
ioannis@43: logger.info("Downloading cloud screening files.")
ioannis@43: self.download_cloudmask(measurement_id)
ioannis@43: if measurement.elpp == 127:
ioannis@45: logger.info("Downloading ElPP files.")
ioannis@43: self.download_elpp(measurement_id)
ioannis@43: if measurement.elda == 127:
ioannis@43: logger.info("Downloading ELDA files.")
ioannis@43: self.download_elda(measurement_id)
ioannis@50: logger.info("Downloading ELDA plots.")
ioannis@43: self.download_plots(measurement_id)
ioannis@43: if measurement.elic == 127:
ioannis@43: logger.info("Downloading ELIC files.")
ioannis@43: self.download_elic(measurement_id)
ioannis@53: if measurement.is_calibration and measurement.eldec==0:
ioannis@43: logger.info("Downloading ELDEC files.")
ioannis@43: self.download_eldec(measurement_id)
victor@7: logger.info("--- Processing finished. ---")
ioannis@43:
victor@7: return measurement
victor@7:
ioannis@43: def get_measurement(self, measurement_id):
victor@7:
ioannis@45: if measurement_id is None: # Is this still required?
victor@7: return None
victor@7:
i@31: measurement_url = self.api_measurement_pattern.format(measurement_id)
i@31: logger.debug("Measurement API URL: %s" % measurement_url)
victor@7:
moritz@29: response = self.session.get(measurement_url)
victor@7:
ioannis@43: response_dict = None
ioannis@45:
ioannis@43: if response.status_code == 200:
ioannis@43: response_dict = response.json()
ioannis@45: elif response.status_code == 404:
ioannis@43: logger.info("No measurement with id %s found on the SCC." % measurement_id)
ioannis@45: else:
i@14: logger.error('Could not access API. Status code %s.' % response.status_code)
victor@7:
victor@7: if response_dict:
ioannis@43: measurement = Measurement(self.base_url, response_dict)
victor@7: else:
ioannis@45: measurement = None
victor@7:
ioannis@45: return measurement, response.status_code
victor@7:
ioannis@45: def delete_measurement(self, measurement_id, delete_related):
victor@7: """ Deletes a measurement with the provided measurement id. The user
madrouin@20: should have the appropriate permissions.
madrouin@20:
victor@7: The procedures is performed directly through the web interface and
victor@7: NOT through the API.
victor@7: """
victor@7: # Get the measurement object
i@31: measurement, _ = self.get_measurement(measurement_id)
victor@7:
victor@7: # Check that it exists
victor@7: if measurement is None:
victor@7: logger.warning("Nothing to delete.")
victor@7: return None
victor@7:
victor@7: # Go the the page confirming the deletion
moritz@29: delete_url = self.delete_measurement_pattern.format(measurement_id)
ioannis@43:
moritz@29: confirm_page = self.session.get(delete_url)
victor@7:
victor@7: # Check that the page opened properly
victor@7: if confirm_page.status_code != 200:
victor@7: logger.warning("Could not open delete page. Status: {0}".format(confirm_page.status_code))
victor@7: return None
victor@7:
ioannis@43: # Get the delete related value
ioannis@43: if delete_related:
ioannis@43: delete_related_option = 'delete_related'
ioannis@43: else:
ioannis@43: delete_related_option = 'not_delete_related'
ioannis@43:
victor@7: # Delete the measurement
victor@7: delete_page = self.session.post(delete_url,
ioannis@43: data={'post': 'yes',
ioannis@43: 'select_delete_related_measurements': delete_related_option},
victor@7: headers={'X-CSRFToken': confirm_page.cookies['csrftoken'],
victor@7: 'referer': delete_url}
victor@7: )
i@31: if not delete_page.ok:
victor@7: logger.warning("Something went wrong. Delete page status: {0}".format(
victor@7: delete_page.status_code))
victor@7: return None
victor@7:
victor@7: logger.info("Deleted measurement {0}".format(measurement_id))
victor@7: return True
victor@7:
victor@7: def available_measurements(self):
victor@7: """ Get a list of available measurement on the SCC. """
i@31: response = self.session.get(self.api_measurements_url)
victor@7: response_dict = response.json()
victor@7:
victor@7: if response_dict:
victor@7: measurement_list = response_dict['objects']
i@14: measurements = [Measurement(self.base_url, measurement_dict) for measurement_dict in measurement_list]
victor@7: logger.info("Found %s measurements on the SCC." % len(measurements))
victor@7: else:
victor@7: logger.warning("No response received from the SCC when asked for available measurements.")
ioannis@43: measurements = None
victor@7:
victor@7: return measurements
victor@7:
moritz@29: def list_measurements(self, station=None, system=None, start=None, stop=None, upload_status=None,
moritz@29: processing_status=None, optical_processing=None):
moritz@29:
i@31: # TODO: Change this to work through the API
i@31:
moritz@29: # Need to set to empty string if not specified, we won't get any results
moritz@29: params = {
moritz@29: "station": station if station is not None else "",
moritz@29: "system": system if system is not None else "",
moritz@29: "stop": stop if stop is not None else "",
moritz@29: "start": start if start is not None else "",
moritz@29: "upload_status": upload_status if upload_status is not None else "",
moritz@29: "preprocessing_status": processing_status if processing_status is not None else "",
moritz@29: "optical_processing_status": optical_processing if optical_processing is not None else ""
moritz@29: }
i@31:
i@31: response_txt = self.session.get(self.list_measurements_url, params=params).text
moritz@29: tbl_rgx = re.compile(r'', re.DOTALL)
moritz@29: entry_rgx = re.compile(r'(.*?)
', re.DOTALL)
moritz@29: measurement_rgx = re.compile(
moritz@29: r'.*?]*>(\w+).*? | .*? | ([\w-]+ [\w:]+) | .* 0:
ioannis@54: logger.warning("{0} exit code: {2[exit_code]} - {2[description]}".format(name, status, exit_code))
ioannis@54: else:
ioannis@54: logger.info("{0} exit code: {2[exit_code]} - {2[description]}".format(name, status, exit_code))
ioannis@53: else:
ioannis@53: logger.info("{0} exit code: {2}".format(name, status, exit_code))
ioannis@53:
ioannis@43: @property
ioannis@43: def rerun_elda_url(self):
ioannis@43: url_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/rerun-elda/')
ioannis@43: return url_pattern.format(self.id)
victor@7:
victor@7: @property
ioannis@43: def rerun_elpp_url(self):
ioannis@43: url_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/rerun-elpp/')
i@14: return url_pattern.format(self.id)
victor@7:
victor@7: @property
victor@7: def rerun_all_url(self):
i@14: ulr_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/rerun-all/')
i@14: return ulr_pattern.format(self.id)
victor@7:
victor@7: def __str__(self):
ioannis@43: return "Measurement {}".format(self.id)
victor@7:
victor@7:
i@38: class AncillaryFile(ApiObject):
i@38: """ This class represents the ancilalry file object as returned in the SCC API.
i@38: """
i@38: @property
i@38: def already_on_scc(self):
i@38: if self.exists is False:
i@38: return False
i@38:
i@38: return not self.status == 'missing'
i@38:
i@38: def __str__(self):
i@38: return "%s: %s, %s" % (self.id,
i@38: self.filename,
i@38: self.status)
i@38:
i@38:
ioannis@45: def process_file(filename, system_id, settings, force_upload, delete_related,
ioannis@54: delay=0, monitor=True, rs_filename=None, lr_filename=None, ov_filename=None):
ioannis@32: """ Shortcut function to process a file to the SCC. """
ioannis@32: logger.info("Processing file %s, using system %s" % (filename, system_id))
victor@7:
ioannis@43: with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc:
ioannis@43: scc.login(settings['website_credentials'])
ioannis@45: measurement = scc.process(filename, system_id,
ioannis@45: force_upload=force_upload,
ioannis@45: delete_related=delete_related,
ioannis@54: delay=delay,
ioannis@45: monitor=monitor,
ioannis@45: rs_filename=rs_filename,
ioannis@45: lr_filename=lr_filename,
ioannis@45: ov_filename=ov_filename)
ioannis@43: scc.logout()
victor@7: return measurement
victor@7:
victor@7:
ioannis@45: def delete_measurements(measurement_ids, delete_related, settings):
moritz@29: """ Shortcut function to delete measurements from the SCC. """
ioannis@45: with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc:
ioannis@45: scc.login(settings['website_credentials'])
ioannis@45: for m_id in measurement_ids:
ioannis@54: logger.info("Deleting %s." % m_id)
ioannis@45: scc.delete_measurement(m_id, delete_related)
ioannis@45: scc.logout()
moritz@29:
moritz@29:
moritz@29: def rerun_all(measurement_ids, monitor, settings):
moritz@29: """ Shortcut function to rerun measurements from the SCC. """
i@14:
ioannis@43: with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc:
ioannis@43: scc.login(settings['website_credentials'])
ioannis@45: for m_id in measurement_ids:
ioannis@54: logger.info("Rerunning all products for %s." % m_id)
ioannis@45: scc.rerun_all(m_id, monitor)
ioannis@43: scc.logout()
victor@7:
victor@7:
moritz@29: def rerun_processing(measurement_ids, monitor, settings):
victor@7: """ Shortcut function to delete a measurement from the SCC. """
i@14:
ioannis@43: with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc:
ioannis@43: scc.login(settings['website_credentials'])
ioannis@45: for m_id in measurement_ids:
ioannis@45: logger.info("Rerunning (optical) processing for %s" % m_id)
ioannis@45: scc.rerun_elpp(m_id, monitor)
ioannis@45: scc.logout()
moritz@29:
moritz@29:
moritz@29: def list_measurements(settings, station=None, system=None, start=None, stop=None, upload_status=None,
moritz@29: preprocessing_status=None,
moritz@29: optical_processing=None):
moritz@29: """List all available measurements"""
ioannis@45: with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc:
ioannis@45: scc.login(settings['website_credentials'])
ioannis@45: ret = scc.list_measurements(station=station, system=system, start=start, stop=stop, upload_status=upload_status,
ioannis@45: processing_status=preprocessing_status, optical_processing=optical_processing)
ioannis@45: for entry in ret:
ioannis@45: print("%s" % entry.id)
ioannis@43: scc.logout()
victor@7:
victor@7:
moritz@29: def download_measurements(measurement_ids, download_preproc, download_optical, download_graph, settings):
moritz@29: """Download all measurements for the specified IDs"""
ioannis@43: with SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url']) as scc:
ioannis@43: scc.login(settings['website_credentials'])
ioannis@45: for m_id in measurement_ids:
ioannis@45: if download_preproc:
ioannis@45: logger.info("Downloading preprocessed files for '%s'" % m_id)
ioannis@45: scc.download_elpp(m_id)
ioannis@45: logger.info("Complete")
ioannis@45: if download_optical:
ioannis@45: logger.info("Downloading optical files for '%s'" % m_id)
ioannis@45: scc.download_elda(m_id)
ioannis@45: logger.info("Complete")
ioannis@45: if download_graph:
ioannis@45: logger.info("Downloading profile graph files for '%s'" % m_id)
ioannis@45: scc.download_plots(m_id)
ioannis@45: logger.info("Complete")
ioannis@43: scc.logout()
i@14:
i@14:
moritz@29: def settings_from_path(config_file_path):
i@14: """ Read the configuration file.
i@14:
i@14: The file should be in YAML syntax."""
i@14:
i@14: if not os.path.isfile(config_file_path):
moritz@29: raise argparse.ArgumentTypeError("Wrong path for configuration file (%s)" % config_file_path)
i@14:
i@14: with open(config_file_path) as yaml_file:
i@14: try:
i@14: settings = yaml.safe_load(yaml_file)
i@14: logger.debug("Read settings file(%s)" % config_file_path)
moritz@29: except Exception:
moritz@29: raise argparse.ArgumentTypeError("Could not parse YAML file (%s)" % config_file_path)
i@14:
i@14: # YAML limitation: does not read tuples
i@14: settings['basic_credentials'] = tuple(settings['basic_credentials'])
i@14: settings['website_credentials'] = tuple(settings['website_credentials'])
i@14: return settings
i@14:
i@14:
moritz@29: # Setup for command specific parsers
moritz@29: def setup_delete(parser):
moritz@29: def delete_from_args(parsed):
ioannis@45: delete_measurements(parsed.IDs,
ioannis@45: delete_related=False,
ioannis@45: settings=parsed.config)
moritz@29:
moritz@29: parser.add_argument("IDs", nargs="+", help="measurement IDs to delete.")
moritz@29: parser.set_defaults(execute=delete_from_args)
moritz@29:
moritz@29:
moritz@29: def setup_rerun_all(parser):
moritz@29: def rerun_all_from_args(parsed):
moritz@29: rerun_all(parsed.IDs, parsed.process, parsed.config)
moritz@29:
moritz@29: parser.add_argument("IDs", nargs="+", help="Measurement IDs to rerun.")
moritz@29: parser.add_argument("-p", "--process", help="Wait for the results of the processing.",
moritz@29: action="store_true")
moritz@29: parser.set_defaults(execute=rerun_all_from_args)
moritz@29:
moritz@29:
ioannis@45: def setup_rerun_elpp(parser):
moritz@29: def rerun_processing_from_args(parsed):
moritz@29: rerun_processing(parsed.IDs, parsed.process, parsed.config)
moritz@29:
moritz@29: parser.add_argument("IDs", nargs="+", help="Measurement IDs to rerun the processing on.")
moritz@29: parser.add_argument("-p", "--process", help="Wait for the results of the processing.",
moritz@29: action="store_true")
moritz@29: parser.set_defaults(execute=rerun_processing_from_args)
moritz@29:
moritz@29:
ioannis@45: def setup_upload_file(parser):
ioannis@45: """ Upload but do not monitor processing progress. """
ioannis@45: def upload_file_from_args(parsed):
ioannis@45: process_file(parsed.filename, parsed.system, parsed.config,
ioannis@54: delay=parsed.delay,
ioannis@45: monitor=parsed.process,
ioannis@45: force_upload=parsed.force_upload,
ioannis@45: delete_related=False, # For now, use this as default
ioannis@32: rs_filename=parsed.radiosounding,
ioannis@32: ov_filename=parsed.overlap,
ioannis@32: lr_filename=parsed.lidarratio)
moritz@29:
ioannis@54: def delay(arg):
ioannis@54: try:
ioannis@54: int_arg = int(arg)
ioannis@54: except ValueError:
ioannis@54: raise argparse.ArgumentTypeError("Could not convert delay argument {} to integer.".format(arg))
ioannis@54:
ioannis@54: if 0 <= int_arg <= 96:
ioannis@54: return int_arg
ioannis@54: else:
ioannis@54: raise argparse.ArgumentTypeError("Delay should be an integer between 0 and 96.")
ioannis@54:
moritz@29: parser.add_argument("filename", help="Measurement file name or path.")
moritz@29: parser.add_argument("system", help="Processing system id.")
ioannis@54: parser.add_argument("--delay", help="Delay processing by the specified number of hours (0 to 96).",
ioannis@54: default=0, type=delay)
ioannis@45: parser.add_argument("-p", "--process", help="Wait for the processing results.",
ioannis@45: action="store_true")
ioannis@45: parser.add_argument("--force_upload", help="If measurement ID exists on SCC, delete before uploading.",
ioannis@45: action="store_true")
i@30: parser.add_argument("--radiosounding", default=None, help="Radiosounding file name or path")
ioannis@32: parser.add_argument("--overlap", default=None, help="Overlap file name or path")
ioannis@32: parser.add_argument("--lidarratio", default=None, help="Lidar ratio file name or path")
i@30:
moritz@29: parser.set_defaults(execute=upload_file_from_args)
moritz@29:
moritz@29:
moritz@29: def setup_list_measurements(parser):
moritz@29: def list_measurements_from_args(parsed):
ioannis@45: # TODO: Fix this
ioannis@45: logger.warning("This method needs to be updated. Cross-chceck any results.")
ioannis@45:
moritz@29: list_measurements(parsed.config, station=parsed.station, system=parsed.system, start=parsed.start,
moritz@29: stop=parsed.stop,
moritz@29: upload_status=parsed.upload_status, preprocessing_status=parsed.preprocessing_status,
moritz@29: optical_processing=parsed.optical_processing_status)
moritz@29:
moritz@29: def status(arg):
moritz@29: if -127 <= int(arg) <= 127:
moritz@29: return arg
moritz@29: else:
moritz@29: raise argparse.ArgumentTypeError("Status must be between -127 and 127")
moritz@29:
moritz@29: def date(arg):
moritz@29: if re.match(r'\d{4}-\d{2}-\d{2}', arg):
moritz@29: return arg
moritz@29: else:
moritz@29: raise argparse.ArgumentTypeError("Date must be in format 'YYYY-MM-DD'")
moritz@29:
moritz@29: parser.add_argument("--station", help="Filter for only the selected station")
moritz@29: parser.add_argument("--system", help="Filter for only the selected station")
moritz@29: parser.add_argument("--start", help="Filter for only the selected station", type=date)
moritz@29: parser.add_argument("--stop", help="Filter for only the selected station", type=date)
moritz@29: parser.add_argument("--upload-status", help="Filter for only the selected station", type=status)
moritz@29: parser.add_argument("--preprocessing-status", help="Filter for only the selected station", type=status)
moritz@29: parser.add_argument("--optical-processing-status", help="Filter for only the selected station", type=status)
moritz@29: parser.set_defaults(execute=list_measurements_from_args)
moritz@29:
moritz@29:
moritz@29: def setup_download_measurements(parser):
moritz@29: def download_measurements_from_args(parsed):
ioannis@45: # TODO: Fix this
ioannis@45: logger.warning("This method needs to be updated. Cross-chceck any results.")
ioannis@45:
ioannis@45: preproc = parsed.download_elpp
ioannis@45: optical = parsed.download_elda
moritz@29: graphs = parsed.download_profile_graphs
moritz@29: if not preproc and not graphs:
moritz@29: optical = True
moritz@29: download_measurements(parsed.IDs, preproc, optical, graphs, parsed.config)
moritz@29:
moritz@29: parser.add_argument("IDs", help="Measurement IDs that should be downloaded.", nargs="+")
moritz@29: parser.add_argument("--download-preprocessed", action="store_true", help="Download preprocessed files.")
moritz@29: parser.add_argument("--download-optical", action="store_true",
moritz@29: help="Download optical files (default if no other download is used).")
moritz@29: parser.add_argument("--download-profile-graphs", action="store_true", help="Download profile graph files.")
moritz@29: parser.set_defaults(execute=download_measurements_from_args)
moritz@29:
moritz@29:
victor@7: def main():
victor@7: # Define the command line arguments.
victor@7: parser = argparse.ArgumentParser()
moritz@29: subparsers = parser.add_subparsers()
moritz@29:
moritz@29: delete_parser = subparsers.add_parser("delete", help="Deletes a measurement.")
ioannis@45: rerun_all_parser = subparsers.add_parser("rerun-all", help="Rerun all processing steps for the provided measurement IDs.")
ioannis@45: rerun_processing_parser = subparsers.add_parser("rerun-elpp",
ioannis@45: help="Rerun low-resolution processing steps for the provided measurement ID.")
ioannis@45: upload_file_parser = subparsers.add_parser("upload-file", help="Submit a file and, optionally, download the output products.")
ioannis@32: list_parser = subparsers.add_parser("list", help="List measurements registered on the SCC.")
moritz@29: download_parser = subparsers.add_parser("download", help="Download selected measurements.")
moritz@29:
moritz@29: setup_delete(delete_parser)
moritz@29: setup_rerun_all(rerun_all_parser)
ioannis@45: setup_rerun_elpp(rerun_processing_parser)
ioannis@45:
moritz@29: setup_upload_file(upload_file_parser)
moritz@29: setup_list_measurements(list_parser)
moritz@29: setup_download_measurements(download_parser)
victor@7:
victor@7: # Verbosity settings from http://stackoverflow.com/a/20663028
victor@7: parser.add_argument('-d', '--debug', help="Print debugging information.", action="store_const",
victor@7: dest="loglevel", const=logging.DEBUG, default=logging.INFO,
victor@7: )
victor@7: parser.add_argument('-s', '--silent', help="Show only warning and error messages.", action="store_const",
victor@7: dest="loglevel", const=logging.WARNING
victor@7: )
victor@7:
ioannis@32: # Setup default config location
moritz@29: home = os.path.expanduser("~")
moritz@29: default_config_location = os.path.abspath(os.path.join(home, ".scc_access.yaml"))
moritz@29: parser.add_argument("-c", "--config", help="Path to the config file.", type=settings_from_path,
moritz@29: default=default_config_location)
victor@7:
victor@7: args = parser.parse_args()
ioannis@43:
victor@7: # Get the logger with the appropriate level
victor@7: logging.basicConfig(format='%(levelname)s: %(message)s', level=args.loglevel)
victor@7:
moritz@29: # Dispatch to appropriate function
moritz@29: args.execute(args)
moritz@29:
i@14:
moritz@29: # When running through terminal
moritz@29: if __name__ == '__main__':
moritz@29: main()
|