victor@7: import requests
victor@7: requests.packages.urllib3.disable_warnings()
victor@7:
i@14: import sys
victor@7: import urlparse
victor@7: import argparse
victor@7: import os
victor@7: import re
victor@7: import time
victor@7: import StringIO
victor@7: from zipfile import ZipFile
victor@7: import datetime
victor@7: import logging
i@14: import yaml
victor@7:
victor@7:
i@14: logger = logging.getLogger(__name__)
victor@7:
victor@7: # The regex to find the measurement id from the measurement page
victor@7: # This should be read from the uploaded file, but would require an extra NetCDF module.
victor@7: regex = "
Measurement (?P.{12}) "
victor@7:
victor@7:
victor@7: class SCC:
victor@7: """ 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):
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()
i@14: self.construct_urls()
victor@7:
i@14: def construct_urls(self):
i@14: """ Construct all URLs needed for processing. """
i@14: # Construct the absolute URLs
i@14: self.login_url = urlparse.urljoin(self.base_url, 'accounts/login/')
i@14: self.upload_url = urlparse.urljoin(self.base_url, 'data_processing/measurements/quick/')
i@14: self.download_preprocessed_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/download-preprocessed/')
i@14: self.download_optical_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/download-optical/')
i@14: self.download_graph_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/download-plots/')
i@14: self.delete_measurement_pattern = urlparse.urljoin(self.base_url, 'admin/database/measurements/{0}/delete/')
i@14: self.api_base_url = urlparse.urljoin(self.base_url, 'api/v1/')
i@14:
i@14: def login(self, credentials):
victor@7: """ Login the the website. """
victor@7: logger.debug("Attempting to login to SCC, username %s." % credentials[0])
victor@7: self.login_credentials = {'username': credentials[0],
victor@7: 'password': credentials[1]}
victor@7:
i@14: logger.debug("Accessing login page at %s." % self.login_url)
victor@7:
victor@7: # Get upload form
i@14: login_page = self.session.get(self.login_url, auth=self.auth, verify=False)
i@14:
i@14: if login_page.status_code != 200:
i@14: logger.error('Could not access login pages. Status code %s' % login_page.status_code)
i@14: sys.exit(1)
victor@7:
victor@7: logger.debug("Submiting credentials.")
madrouin@20:
victor@7: # Submit the login data
i@14: login_submit = self.session.post(self.login_url,
victor@7: data=self.login_credentials,
victor@7: headers={'X-CSRFToken': login_page.cookies['csrftoken'],
i@14: 'referer': self.login_url},
victor@7: verify=False,
victor@7: auth=self.auth)
victor@7: return login_submit
victor@7:
victor@7: def logout(self):
victor@7: pass
victor@7:
madrouin@20: def upload_file(self, filename, system_id, rs_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. """
victor@7: # Get submit page
i@14: upload_page = self.session.get(self.upload_url,
victor@7: auth=self.auth,
victor@7: verify=False)
victor@7:
victor@7: # Submit the data
victor@7: upload_data = {'system': system_id}
victor@7: files = {'data': open(filename, 'rb')}
victor@7:
madrouin@20: if rs_filename is not None:
madrouin@20: files['sounding_file'] = open(rs_filename, 'rb')
madrouin@20:
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'],
i@14: 'referer': self.upload_url,},
victor@7: verify=False,
victor@7: auth=self.auth)
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
i@14: logger.error("Uploaded file rejected! Try to upload manually to see the error.")
victor@7: else:
victor@7: measurement_id = re.findall(regex, upload_submit.text)[0]
i@14: logger.error("Successfully uploaded measurement with id %s." % measurement_id)
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: """
victor@7: # Get the file
victor@7: request = self.session.get(download_url, auth=self.auth,
victor@7: verify=False,
victor@7: stream=True)
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.
victor@7: memory_file = StringIO.StringIO()
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:
victor@7: def download_preprocessed(self, measurement_id):
victor@7: """ Download preprocessed files for the measurement id. """
victor@7: # Construct the download url
i@14: download_url = self.download_preprocessed_pattern.format(measurement_id)
victor@7: self.download_files(measurement_id, 'scc_preprocessed', download_url)
victor@7:
victor@7: def download_optical(self, measurement_id):
victor@7: """ Download optical files for the measurement id. """
victor@7: # Construct the download url
i@15: download_url = self.download_optical_pattern.format(measurement_id)
victor@7: self.download_files(measurement_id, 'scc_optical', download_url)
victor@7:
victor@7: def download_graphs(self, measurement_id):
victor@7: """ Download profile graphs for the measurement id. """
victor@7: # Construct the download url
i@14: download_url = self.download_graph_pattern.format(measurement_id)
victor@7: self.download_files(measurement_id, 'scc_plots', download_url)
victor@7:
victor@7: def rerun_processing(self, measurement_id, monitor=True):
victor@7: measurement = self.get_measurement(measurement_id)
victor@7:
victor@7: if measurement:
victor@7: request = self.session.get(measurement.rerun_processing_url, auth=self.auth,
victor@7: verify=False,
victor@7: 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))
victor@7: return
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)
victor@7: measurement = 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:
victor@7: request = self.session.get(measurement.rerun_all_url, auth=self.auth,
victor@7: verify=False,
victor@7: 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))
victor@7: return
victor@7:
victor@7: if monitor:
victor@7: self.monitor_processing(measurement_id)
victor@7:
victor@7: def process(self, filename, system_id):
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
victor@7: measurement_id = self.upload_file(filename, system_id)
victor@7:
victor@7: measurement = self.monitor_processing(measurement_id)
victor@7: return measurement
victor@7:
victor@7: def monitor_processing(self, measurement_id):
victor@7: """ Monitor the processing progress of a measurement id"""
victor@7:
victor@7: measurement = self.get_measurement(measurement_id)
victor@7: if measurement is not None:
victor@7: while measurement.is_running:
victor@7: logger.info("Measurement is being processed (status: %s, %s, %s). Please wait." % (measurement.upload,
i@14: measurement.pre_processing,
i@14: measurement.processing))
victor@7: time.sleep(10)
victor@7: measurement = self.get_measurement(measurement_id)
victor@7: logger.info("Measurement processing finished (status: %s, %s, %s)." % (measurement.upload,
i@14: measurement.pre_processing,
i@14: measurement.processing))
victor@7: if measurement.pre_processing == 127:
victor@7: logger.info("Downloading preprocessed files.")
victor@7: self.download_preprocessed(measurement_id)
victor@7: if measurement.processing == 127:
victor@7: logger.info("Downloading optical files.")
victor@7: self.download_optical(measurement_id)
victor@7: logger.info("Downloading graphs.")
victor@7: self.download_graphs(measurement_id)
victor@7: logger.info("--- Processing finished. ---")
victor@7: return measurement
victor@7:
victor@7: def get_status(self, measurement_id):
victor@7: """ Get the processing status for a measurement id through the API. """
i@14: measurement_url = urlparse.urljoin(self.api_base_url, 'measurements/?id__exact=%s' % measurement_id)
victor@7:
victor@7: response = self.session.get(measurement_url,
victor@7: auth=self.auth,
victor@7: verify=False)
victor@7:
victor@7: response_dict = response.json()
victor@7:
victor@7: if response_dict['objects']:
victor@7: measurement_list = response_dict['objects']
i@14: measurement = Measurement(self.base_url, measurement_list[0])
i@14: return measurement.upload, measurement.pre_processing, measurement.processing
victor@7: else:
victor@7: logger.error("No measurement with id %s found on the SCC." % measurement_id)
victor@7: return None
victor@7:
victor@7: def get_measurement(self, measurement_id):
i@14: measurement_url = urlparse.urljoin(self.api_base_url, 'measurements/%s/' % measurement_id)
victor@7:
victor@7: response = self.session.get(measurement_url,
victor@7: auth=self.auth,
victor@7: verify=False)
victor@7:
i@14: if response.status_code != 200:
i@14: logger.error('Could not access API. Status code %s.' % response.status_code)
i@14: sys.exit(1)
i@14:
victor@7: response_dict = response.json()
victor@7:
victor@7: if response_dict:
i@14: measurement = Measurement(self.base_url,response_dict)
victor@7: return measurement
victor@7: else:
victor@7: logger.error("No measurement with id %s found on the SCC." % measurement_id)
victor@7: return None
victor@7:
victor@7: def delete_measurement(self, measurement_id):
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
victor@7: 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
i@14: delete_url = self.delete_measurement_pattern.format(measurement.id)
victor@7:
victor@7: confirm_page = self.session.get(delete_url,
victor@7: auth=self.auth,
victor@7: verify=False)
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:
victor@7: # Delete the measurement
victor@7: delete_page = self.session.post(delete_url,
victor@7: auth=self.auth,
victor@7: verify=False,
victor@7: data={'post': 'yes'},
victor@7: headers={'X-CSRFToken': confirm_page.cookies['csrftoken'],
victor@7: 'referer': delete_url}
victor@7: )
victor@7: if delete_page.status_code != 200:
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@14: measurement_url = urlparse.urljoin(self.api_base_url, 'measurements')
victor@7: response = self.session.get(measurement_url,
victor@7: auth=self.auth,
victor@7: verify=False)
victor@7: response_dict = response.json()
victor@7:
victor@7: measurements = None
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.")
victor@7:
victor@7: return measurements
victor@7:
victor@7: def measurement_id_for_date(self, t1, call_sign='bu', base_number=0):
victor@7: """ Give the first available measurement id on the SCC for the specific
madrouin@20: date.
victor@7: """
victor@7: date_str = t1.strftime('%Y%m%d')
i@14: search_url = urlparse.urljoin(self.api_base_url, 'measurements/?id__startswith=%s' % date_str)
victor@7:
victor@7: response = self.session.get(search_url,
victor@7: auth=self.auth,
victor@7: verify=False)
victor@7:
victor@7: response_dict = response.json()
victor@7:
victor@7: measurement_id = None
victor@7:
victor@7: if response_dict:
victor@7: measurement_list = response_dict['objects']
victor@7: existing_ids = [measurement_dict['id'] for measurement_dict in measurement_list]
victor@7:
victor@7: measurement_number = base_number
victor@7: measurement_id = "%s%s%02i" % (date_str, call_sign, measurement_number)
victor@7:
victor@7: while measurement_id in existing_ids:
victor@7: measurement_number = measurement_number + 1
victor@7: measurement_id = "%s%s%02i" % (date_str, call_sign, measurement_number)
victor@7: if measurement_number == 100:
victor@7: raise ValueError('No available measurement id found.')
victor@7:
victor@7: return measurement_id
victor@7:
victor@7:
victor@7: class ApiObject:
victor@7: """ A generic class object. """
victor@7:
i@14: def __init__(self, base_url, dict_response):
i@14: self.base_url = base_url
victor@7:
victor@7: if dict_response:
victor@7: # Add the dictionary key value pairs as object properties
victor@7: for key, value in dict_response.items():
victor@7: setattr(self, key, value)
victor@7: self.exists = True
victor@7: else:
victor@7: self.exists = False
victor@7:
victor@7:
victor@7: class Measurement(ApiObject):
victor@7: """ This class represents the measurement object as returned in the SCC API.
victor@7: """
victor@7:
victor@7: @property
victor@7: def is_running(self):
victor@7: """ Returns True if the processing has not finished.
victor@7: """
victor@7: if self.upload == 0:
victor@7: return False
victor@7: if self.pre_processing == -127:
victor@7: return False
victor@7: if self.pre_processing == 127:
victor@7: if self.processing in [127, -127]:
victor@7: return False
victor@7: return True
victor@7:
victor@7: @property
victor@7: def rerun_processing_url(self):
i@14: url_pattern = urlparse.urljoin(self.base_url, 'data_processing/measurements/{0}/rerun-optical/')
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):
victor@7: return "%s: %s, %s, %s" % (self.id,
victor@7: self.upload,
victor@7: self.pre_processing,
victor@7: self.processing)
victor@7:
victor@7:
madrouin@20: def upload_file(filename, system_id, settings, rs_filename=None):
victor@7: """ Shortcut function to upload a file to the SCC. """
victor@7: logger.info("Uploading file %s, using sytem %s" % (filename, system_id))
victor@7:
i@14: scc = SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url'])
i@14: scc.login(settings['website_credentials'])
madrouin@20: measurement_id = scc.upload_file(filename, system_id, rs_filename=rs_filename)
victor@7: scc.logout()
victor@7: return measurement_id
victor@7:
victor@7:
i@14: def process_file(filename, system_id, settings):
victor@7: """ Shortcut function to process a file to the SCC. """
victor@7: logger.info("Processing file %s, using sytem %s" % (filename, system_id))
victor@7:
i@14: scc = SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url'])
i@14: scc.login(settings['website_credentials'])
victor@7: measurement = scc.process(filename, system_id)
victor@7: scc.logout()
victor@7: return measurement
victor@7:
victor@7:
i@14: def delete_measurement(measurement_id, settings):
victor@7: """ Shortcut function to delete a measurement from the SCC. """
victor@7: logger.info("Deleting %s" % measurement_id)
i@14:
i@14: scc = SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url'])
i@14: scc.login(settings['website_credentials'])
victor@7: scc.delete_measurement(measurement_id)
victor@7: scc.logout()
victor@7:
victor@7:
i@14: def rerun_all(measurement_id, monitor, settings):
victor@7: """ Shortcut function to delete a measurement from the SCC. """
victor@7: logger.info("Rerunning all products for %s" % measurement_id)
i@14:
i@14: scc = SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url'])
i@14: scc.login(settings['website_credentials'])
victor@7: scc.rerun_all(measurement_id, monitor)
victor@7: scc.logout()
victor@7:
victor@7:
i@14: def rerun_processing(measurement_id, monitor, settings):
victor@7: """ Shortcut function to delete a measurement from the SCC. """
victor@7: logger.info("Rerunning (optical) processing for %s" % measurement_id)
i@14:
i@14: scc = SCC(settings['basic_credentials'], settings['output_dir'], settings['base_url'])
i@14: scc.login(settings['website_credentials'])
victor@7: scc.rerun_processing(measurement_id, monitor)
victor@7: scc.logout()
i@14:
i@14:
i@14: def import_settings(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):
i@14: logger.error("Wrong path for configuration file (%s)" % config_file_path)
i@14: sys.exit(1)
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)
i@14: except:
i@14: logger.error("Could not parse YAML file (%s)" % config_file_path)
i@14: sys.exit(1)
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:
victor@7: def main():
victor@7: # Define the command line arguments.
victor@7: parser = argparse.ArgumentParser()
i@16: parser.add_argument("config", help="Path to configuration file")
victor@7: parser.add_argument("filename", nargs='?', help="Measurement file name or path.", default='')
victor@7: parser.add_argument("system", nargs='?', help="Processing system id.", default=0)
victor@7: parser.add_argument("-p", "--process", help="Wait for the results of the processing.",
victor@7: action="store_true")
victor@7: parser.add_argument("--delete", help="Measurement ID to delete.")
victor@7: parser.add_argument("--rerun-all", help="Measurement ID to rerun.")
i@14: parser.add_argument("--rerun-processing", help="Measurement ID to rerun processing routines.")
victor@7:
madrouin@20: # others files
madrouin@20: parser.add_argument("--radiosounding", default=None, help="Radiosounding file name or path")
madrouin@20:
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:
victor@7: args = parser.parse_args()
victor@7:
victor@7: # Get the logger with the appropriate level
victor@7: logging.basicConfig(format='%(levelname)s: %(message)s', level=args.loglevel)
victor@7:
i@14: settings = import_settings(args.config)
i@14:
i@14: # If the arguments are OK, try to log-in to SCC and upload.
victor@7: if args.delete:
victor@7: # If the delete is provided, do nothing else
i@14: delete_measurement(args.delete, settings)
victor@7: elif args.rerun_all:
i@14: rerun_all(args.rerun_all, args.process, settings)
victor@7: elif args.rerun_processing:
i@14: rerun_processing(args.rerun_processing, args.process, settings)
victor@7: else:
victor@7: if (args.filename == '') or (args.system == 0):
victor@7: parser.error('Provide a valid filename and system parameters.\nRun with -h for help.\n')
victor@7:
victor@7: if args.process:
i@14: process_file(args.filename, args.system, settings)
victor@7: else:
madrouin@20: upload_file(args.filename, args.system, settings, rs_filename=args.radiosounding)