diff --git a/hapiclient/__init__.py b/hapiclient/__init__.py index f480fbd..8188fac 100644 --- a/hapiclient/__init__.py +++ b/hapiclient/__init__.py @@ -22,8 +22,3 @@ import warnings warnings.filterwarnings("ignore", message=".*urllib3.*OpenSSL.*") -if sys.version_info[0] < 3: - # Python 2.7 - reload(sys) - sys.setdefaultencoding('utf8') - diff --git a/hapiclient/cache.py b/hapiclient/cache.py index 2ab35e7..a8df13c 100644 --- a/hapiclient/cache.py +++ b/hapiclient/cache.py @@ -32,20 +32,19 @@ def cachedir(*args): return os.path.join(args[0], server2dirname(args[1])) -def request2path(*args): - # request2path(server, dataset, parameters, start, stop) - # request2path(server, dataset, parameters, start, stop, cachedir) +def request2path(server, dataset=None, parameters=None, start=None, stop=None, cache_dir=None, endpoint=None): + import os import re import platform - if len(args) == 5: - # Use default if cachedir not given. + if cache_dir is None: + # Use default if cache_dir not given. cachedirectory = cachedir() else: - cachedirectory = args[5] + cachedirectory = cache_dir - args = list(args) + args = [server, dataset, parameters, start, stop] # Replace forbidden characters in directory and filename # Replacements assume that there will be no name collisions, @@ -68,28 +67,50 @@ def request2path(*args): ) for element in reps: - args[1] = re.sub(element[0], element[1], args[1]) - args[2] = re.sub(element[0], element[1], args[2]) + if dataset is not None: + dataset = re.sub(element[0], element[1], dataset) + if parameters is not None: + parameters = re.sub(element[0], element[1], parameters) else: - args[1] = re.sub('/','@forwardslash@',args[1]) - args[2] = re.sub('/','@forwardslash@',args[2]) + if dataset is not None: + dataset = re.sub('/','@forwardslash@', dataset) + if parameters is not None: + parameters = re.sub('/','@forwardslash@', parameters) # To shorten filenames. - args[3] = re.sub(r'-|:|\.|Z', '', args[3]) - args[4] = re.sub(r'-|:|\.|Z', '', args[4]) + if start is not None: + start = re.sub(r'-|:|\.|Z', '', start) + if stop is not None: + stop = re.sub(r'-|:|\.|Z', '', stop) # URL subdirectory urldirectory = server2dirname(args[0]) - fname = '%s_%s_%s_%s' % (args[1], args[2], args[3], args[4]) - return os.path.join(cachedirectory, urldirectory, fname) + if not dataset and not endpoint: + raise ValueError('Either dataset or endpoint must be specified.') + + if endpoint is None: + endpoint = '' + + if dataset is None: + fname = endpoint + else: + fname = dataset + if parameters is not None: + fname += '_' + parameters + if start is not None: + fname += '_' + start + if stop is not None: + fname += '_' + stop + + return os.path.join(cachedirectory, urldirectory, endpoint, fname) -def meta_cache_paths(SERVER, DATASET, cachedir): +def meta_cache_paths(server, dataset, endpoint, cache_dir): """Return dict with metadata cache directory and file names.""" - fname_root = request2path(SERVER, DATASET, '', '', '', cachedir) + fname_root = request2path(server, dataset, cache_dir=cache_dir, endpoint=endpoint) return { 'json': fname_root + '.json', @@ -97,7 +118,7 @@ def meta_cache_paths(SERVER, DATASET, cachedir): } -def meta_cache_read(SERVER, DATASET, opts): +def meta_cache_read(server, dataset, endpoint, opts): """Read metadata from PKL cache. Returns meta dict or None.""" import os @@ -106,10 +127,13 @@ def meta_cache_read(SERVER, DATASET, opts): from hapiclient.log import log if not opts["usecache"]: - log('Not checking metadata cache because usecache is False.') + if endpoint in ['', 'info']: + log(f'Not checking metadata cache for /info?dataset={dataset} response because usecache is False.') + else: + log(f'Not checking metadata cache for /{endpoint} response because usecache is False.') return None - fnamepkl = meta_cache_paths(SERVER, DATASET, opts['cachedir'])['pkl'] + fnamepkl = meta_cache_paths(server, dataset, endpoint, opts['cachedir'])['pkl'] if os.path.isfile(fnamepkl): log('Reading %s' % os.path.basename(fnamepkl)) with open(fnamepkl, 'rb') as f: @@ -121,7 +145,7 @@ def meta_cache_read(SERVER, DATASET, opts): return None -def meta_cache_write(meta, SERVER, DATASET, opts): +def meta_cache_write(meta, server, dataset, endpoint, opts): """Write metadata to JSON and PKL cache files.""" import os @@ -132,7 +156,7 @@ def meta_cache_write(meta, SERVER, DATASET, opts): if not opts["cache"]: return - paths = meta_cache_paths(SERVER, DATASET, opts['cachedir']) + paths = meta_cache_paths(server, dataset, endpoint, opts['cachedir']) fnamejson, fnamepkl = paths['json'], paths['pkl'] log('Writing %s ' % os.path.basename(fnamejson)) @@ -142,10 +166,10 @@ def meta_cache_write(meta, SERVER, DATASET, opts): write_atomic(fnamepkl, meta) -def data_cache_paths(SERVER, DATASET, PARAMETERS, START, STOP, cachedir): +def data_cache_paths(server, dataset, parameters, start, stop, cache_dir): """Return dict with data cache file names.""" - fname_root = request2path(SERVER, DATASET, PARAMETERS, START, STOP, cachedir) + fname_root = request2path(server, dataset, parameters, start, stop, cache_dir, 'data') return { 'csv': fname_root + '.csv', @@ -155,7 +179,7 @@ def data_cache_paths(SERVER, DATASET, PARAMETERS, START, STOP, cachedir): } -def data_cache_read_metax(SERVER, DATASET, PARAMETERS, START, STOP, opts): +def data_cache_read_metax(server, dataset, parameters, start, stop, opts): """Read extended request metadata from PKL cache. Returns meta dict or None.""" import os @@ -167,7 +191,7 @@ def data_cache_read_metax(SERVER, DATASET, PARAMETERS, START, STOP, opts): log('Not checking subsetted metadata cache because usecache is False.') return None - fnamepklx = data_cache_paths(SERVER, DATASET, PARAMETERS, START, STOP, opts['cachedir'])['pkl'] + fnamepklx = data_cache_paths(server, dataset, parameters, start, stop, opts['cachedir'])['pkl'] if os.path.isfile(fnamepklx): log('Reading subsetted metadata cache %s' % os.path.basename(fnamepklx)) with open(fnamepklx, 'rb') as f: @@ -178,7 +202,7 @@ def data_cache_read_metax(SERVER, DATASET, PARAMETERS, START, STOP, opts): return None -def data_cache_read_npy(SERVER, DATASET, PARAMETERS, START, STOP, opts): +def data_cache_read_npy(server, dataset, parameters, start, stop, opts): """Read cached numpy data array. Returns None if not cached.""" import os @@ -189,7 +213,7 @@ def data_cache_read_npy(SERVER, DATASET, PARAMETERS, START, STOP, opts): if not opts["usecache"]: return None - fnamenpy = data_cache_paths(SERVER, DATASET, PARAMETERS, START, STOP, opts['cachedir'])['npy'] + fnamenpy = data_cache_paths(server, dataset, parameters, start, stop, opts['cachedir'])['npy'] if not os.path.isfile(fnamenpy): return None @@ -201,7 +225,7 @@ def data_cache_read_npy(SERVER, DATASET, PARAMETERS, START, STOP, opts): return data -def data_cache_write(data_result, meta, SERVER, DATASET, PARAMETERS, START, STOP, opts): +def data_cache_write(data_result, meta, server, dataset, parameters, start, stop, opts): """Write data array and extended metadata to cache files. Also updates meta with file-related x_ fields before writing. @@ -212,10 +236,10 @@ def data_cache_write(data_result, meta, SERVER, DATASET, PARAMETERS, START, STOP from hapiclient.log import log from hapiclient.util import write_atomic - data_paths = data_cache_paths(SERVER, DATASET, PARAMETERS, START, STOP, opts['cachedir']) + data_paths = data_cache_paths(server, dataset, parameters, start, stop, opts['cachedir']) fnamecsv, fnamebin, fnamenpy, fnamepklx = data_paths['csv'], data_paths['bin'], data_paths['npy'], data_paths['pkl'] - meta_paths = meta_cache_paths(SERVER, DATASET, opts['cachedir']) + meta_paths = meta_cache_paths(server, dataset, 'info', opts['cachedir']) fnamejson, fnamepkl = meta_paths['json'], meta_paths['pkl'] meta.update({"x_metaFileParsed": fnamepkl}) diff --git a/hapiclient/capabilities.py b/hapiclient/capabilities.py index bcf710a..782fd9e 100644 --- a/hapiclient/capabilities.py +++ b/hapiclient/capabilities.py @@ -1,6 +1,6 @@ -def capabilities(SERVER): - """Return the capabilities of a HAPI server. +def capabilities(SERVER, opts): + """Return the /capabilities response from a HAPI server. Args: SERVER (str): The base URL of the HAPI server. @@ -8,42 +8,15 @@ def capabilities(SERVER): Returns: dict: A dictionary containing the capabilities of the server. """ - from hapiclient.util import urlopen - caps = urlopen(SERVER + '/capabilities', parse_json=True) + import hapiclient as hc - return caps - - -def get_format(SERVER, format): - """Return the transport format to use, accounting for server capabilities. - - If the requested format is not supported by the server, falls back to 'csv'. - """ + caps = hc.cache.meta_cache_read(SERVER, None, 'capabilities', opts) + if caps is not None: + return caps - from hapiclient.util import error + caps = hc.util.urlopen(SERVER + '/capabilities', parse_json=True) - cformats = ['csv', 'binary'] # client formats - if format not in cformats: - msg = 'This client does not handle streaming format "%s". Available options: %s' - error(msg % (format, ', '.join(cformats))) + hc.cache.meta_cache_write(caps, SERVER, None, 'capabilities', opts) - if format != 'csv': - caps = capabilities(SERVER) - if "outputFormats" not in caps: - return 'csv' - - formats = caps.get("outputFormats", []) # Server formats - if len(formats) == 0: - return 'csv' - - if format not in formats: - #from hapiclient.util import warning - #msg = 'Requested streaming format "%s" not available from %s. Will use "csv". Available options: %s' - #warning(msg % (format, SERVER, ', '.join(formats))) - format = 'csv' - - if 'binary' not in formats: - format = 'csv' - - return format + return caps diff --git a/hapiclient/catalog.py b/hapiclient/catalog.py index 3f2bb4d..c3c96f9 100644 --- a/hapiclient/catalog.py +++ b/hapiclient/catalog.py @@ -1,10 +1,14 @@ -from hapiclient.log import log -from hapiclient.util import urlopen +def catalog(SERVER, opts): + import hapiclient as hc + + cat = hc.cache.meta_cache_read(SERVER, None, 'catalog', opts) + if cat is not None: + return cat -def catalog(SERVER): - # TODO: Cache url = SERVER + '/catalog' - meta = urlopen(url, parse_json=True) + cat = hc.util.urlopen(url, parse_json=True) + + hc.cache.meta_cache_write(cat, SERVER, None, 'catalog', opts) - return meta + return cat diff --git a/hapiclient/data.py b/hapiclient/data.py index a69ca87..74213df 100644 --- a/hapiclient/data.py +++ b/hapiclient/data.py @@ -1,7 +1,8 @@ -def data(SERVER, DATASET, PARAMETERS, START, STOP, opts): +def data(server, dataset, parameters, start, stop, opts): import os import time + from datetime import datetime from hapiclient.log import log @@ -9,32 +10,32 @@ def data(SERVER, DATASET, PARAMETERS, START, STOP, opts): from hapiclient.cache import cachedir, data_cache_read_metax, data_cache_read_npy, data_cache_write from hapiclient.get import get_binary, get_csv from hapiclient.info import info - from hapiclient.capabilities import get_format + from hapiclient.capabilities import capabilities - unicode_check(DATASET, PARAMETERS) - PARAMETERS = fix_parameters(PARAMETERS) + unicode_check(dataset, parameters) + parameters = fix_parameters(parameters) - urld = cachedir(opts["cachedir"], SERVER) + urld = cachedir(opts["cachedir"], server) if opts['usecache'] or opts['cache']: log('cache subdirectory = %s' % urld) - if STOP is None: + if stop is None: log('STOP was given as None. Getting stopDate for dataset.') - meta = info(SERVER, DATASET, None, opts) - STOP = meta['stopDate'] - log(f'Using STOP = {STOP}') + meta = info(server, dataset, None, opts) + stop = meta['stopDate'] + log(f'Using STOP = {stop}') tic_totalTime = time.time() - meta = data_cache_read_metax(SERVER, DATASET, PARAMETERS, START, STOP, opts) + meta = data_cache_read_metax(server, dataset, parameters, start, stop, opts) metaFromCache = meta is not None if not metaFromCache: - meta = info(SERVER, DATASET, None, opts) + meta = info(server, dataset, None, opts) # Add information to metadata so we can figure out the request needed # to generate it. Will also be used for labeling plots by hapiplot(). - meta.update({"x_server": SERVER}) - meta.update({"x_dataset": DATASET}) + meta.update({"x_server": server}) + meta.update({"x_dataset": dataset}) if opts["cache"]: if not os.path.exists(urld): @@ -44,22 +45,23 @@ def data(SERVER, DATASET, PARAMETERS, START, STOP, opts): opts['dt_chunk'] = _dt_chunk_infer(meta, opts) if opts['n_chunks'] is not None or opts['dt_chunk'] is not None: - chunk_result = _get_chunks(SERVER, DATASET, PARAMETERS, START, STOP, opts, tic_totalTime) + chunk_result = _get_chunks(server, dataset, parameters, start, stop, opts, tic_totalTime) if chunk_result is not None: return chunk_result if not metaFromCache: - meta = subset_meta(meta, PARAMETERS) + meta = subset_meta(meta, parameters) tic = time.time() - data_cached = data_cache_read_npy(SERVER, DATASET, PARAMETERS, START, STOP, opts) + data_cached = data_cache_read_npy(server, dataset, parameters, start, stop, opts) if data_cached is not None: meta['x_totalTime'] = time.time() - tic_totalTime meta['x_readTime'] = tic - tic_totalTime meta['x_downloadTime'] = 0 return data_cached, meta - opts['format'] = get_format(SERVER, opts['format']) + caps = capabilities(server, opts) + opts['format'] = _get_format(caps, opts) # length attribute required for all parameters when serving binary but # is only required for time parameter when serving CSV. This catches @@ -73,29 +75,68 @@ def data(SERVER, DATASET, PARAMETERS, START, STOP, opts): # Read the data. toc0 is time to download to file or into buffer; # toc is time to parse. if opts['format'] == 'binary': - data_result, toc0, toc = get_binary(meta, SERVER, DATASET, PARAMETERS, START, STOP, opts) + data_result, toc0, toc = get_binary(meta, server, dataset, parameters, start, stop, opts) else: - data_result, toc0, toc = get_csv(meta, SERVER, DATASET, PARAMETERS, START, STOP, opts) + data_result, toc0, toc = get_csv(meta, server, dataset, parameters, start, stop, opts) # Extra metadata associated with request will be saved in # a pkl file with same base name as npy data file. - meta.update({"x_server": SERVER}) - meta.update({"x_dataset": DATASET}) - meta.update({"x_parameters": PARAMETERS}) - meta.update({"x_time.min": START}) - meta.update({"x_time.max": STOP}) + meta.update({"x_server": server}) + meta.update({"x_dataset": dataset}) + meta.update({"x_parameters": parameters}) + meta.update({"x_time.min": start}) + meta.update({"x_time.max": stop}) meta.update({"x_requestDate": datetime.now().isoformat()[0:19]}) + meta.update({"x_capabilities": caps}) meta.update({"x_cacheDir": urld}) meta.update({"x_downloadTime": toc0}) meta.update({"x_readTime": toc}) - data_cache_write(data_result, meta, SERVER, DATASET, PARAMETERS, START, STOP, opts) + data_cache_write(data_result, meta, server, dataset, parameters, start, stop, opts) meta['x_totalTime'] = time.time() - tic_totalTime return data_result, meta +def _get_format(caps, opts): + """Return the streaming format to use, accounting for server capabilities. + + If the requested format is not supported by the server, falls back to 'csv'. + """ + + from hapiclient.util import error + + format = opts['format'] + + cformats = ['csv', 'binary'] # client formats + if format not in cformats: + msg = 'This client does not handle streaming format "%s". Available options: %s' + error(msg % (format, ', '.join(cformats))) + + if format == 'csv': + # All servers must support CSV, so no need to check capabilities. + return format + + if "outputFormats" not in caps: + return 'csv' + + formats = caps.get("outputFormats", []) # Server formats + if len(formats) == 0: + return 'csv' + + if format not in formats: + #from hapiclient.util import warning + #msg = 'Requested streaming format "%s" not available from %s. Will use "csv". Available options: %s' + #warning(msg % (format, SERVER, ', '.join(formats))) + format = 'csv' + + if 'binary' not in formats: + format = 'csv' + + return format + + def _dt_chunk_infer(meta, opts): import isodate @@ -129,7 +170,7 @@ def _dt_chunk_infer(meta, opts): return 'P1Y' -def _get_chunks(SERVER, DATASET, PARAMETERS, START, STOP, opts, tic_totalTime): +def _get_chunks(server, dataset, parameters, start, stop, opts, tic_totalTime): import sys import time @@ -144,8 +185,8 @@ def _get_chunks(SERVER, DATASET, PARAMETERS, START, STOP, opts, tic_totalTime): def padz(value): return value if 'Z' in value else value + 'Z' - pSTART = hapitime2datetime(padz(START))[0] - pSTOP = hapitime2datetime(padz(STOP))[0] + pSTART = hapitime2datetime(padz(start))[0] + pSTOP = hapitime2datetime(padz(stop))[0] if opts['dt_chunk']: pDELTA = isodate.parse_duration(opts['dt_chunk']) @@ -160,7 +201,7 @@ def padz(value): if (pSTOP - pSTART) < half: opts['n_chunks'] = None opts['dt_chunk'] = None - return data(SERVER, DATASET, PARAMETERS, START, STOP, opts) + return data(server, dataset, parameters, start, stop, opts) if opts['dt_chunk'] == 'P1Y': pSTART = datetime(pSTART.year, 1, 1) @@ -202,29 +243,22 @@ def padz(value): if opts.get('logging'): verbose = 100 - def nhapi(SERVER, DATASET, PARAMETERS, pSTART, pDELTA, i, **opts): - START = pSTART + (i * pDELTA) - START = str(START.date())+'T'+str(START.time()) + def nhapi(server, dataset, parameters, pSTART, pDELTA, i, **opts): + start = pSTART + (i * pDELTA) + start = str(start.date())+'T'+str(start.time()) - STOP = pSTART + ((i + 1) * pDELTA) - STOP = str(STOP.date()) + 'T' + str(STOP.time()) + stop = pSTART + ((i + 1) * pDELTA) + stop = str(stop.date()) + 'T' + str(stop.time()) - data_chunk, meta = data( - SERVER, - DATASET, - PARAMETERS, - START, - STOP, - opts - ) + data_chunk, meta = data(server, dataset, parameters, start, stop, opts) return data_chunk, meta resD, resM = zip( *Parallel(n_jobs=opts['n_parallel'], verbose=verbose, backend=backend)( delayed(nhapi)( - SERVER, - DATASET, - PARAMETERS, + server, + dataset, + parameters, pSTART, pDELTA, i, @@ -236,18 +270,13 @@ def nhapi(SERVER, DATASET, PARAMETERS, pSTART, pDELTA, i, **opts): resD = list(resD) tic_trimTime = time.time() - if sys.version_info < (3, ): - START = hapitime_reformat(str(resD[0]['Time'][0]), START) - resD[0] = resD[0][resD[0]['Time'] >= START] - STOP = hapitime_reformat(str(resD[-1]['Time'][0]), STOP) - resD[-1] = resD[-1][resD[-1]['Time'] < STOP] - else: - START = hapitime_reformat(resD[0]['Time'][0].decode('UTF-8'), START) - resD[0] = resD[0][resD[0]['Time'] >= bytes(START, 'UTF-8')] + start = hapitime_reformat(resD[0]['Time'][0].decode('UTF-8'), start) + resD[0] = resD[0][resD[0]['Time'] >= bytes(start, 'UTF-8')] + + stop = hapitime_reformat(resD[-1]['Time'][0].decode('UTF-8'), stop) + resD[-1] = resD[-1][resD[-1]['Time'] < bytes(stop, 'UTF-8')] - STOP = hapitime_reformat(resD[-1]['Time'][0].decode('UTF-8'), STOP) - resD[-1] = resD[-1][resD[-1]['Time'] < bytes(STOP, 'UTF-8')] trimTime = time.time() - tic_trimTime tic_catTime = time.time() diff --git a/hapiclient/hapi.py b/hapiclient/hapi.py index c8a8136..259ce34 100644 --- a/hapiclient/hapi.py +++ b/hapiclient/hapi.py @@ -240,7 +240,7 @@ def hapi(*args, **kwargs): # hapi(SERVER) if nin == 1: - return catalog(SERVER) + return catalog(SERVER, opts) # hapi(SERVER, DATASET) if nin == 2: diff --git a/hapiclient/hapitime.py b/hapiclient/hapitime.py index d97199f..2250648 100644 --- a/hapiclient/hapitime.py +++ b/hapiclient/hapitime.py @@ -10,6 +10,7 @@ from hapiclient.log import log from hapiclient.util import error + def hapitime_reformat(form_to_match, given_form): """Reformat a given HAPI time to match format of another HAPI time. diff --git a/hapiclient/info.py b/hapiclient/info.py index 45356d6..54f1138 100644 --- a/hapiclient/info.py +++ b/hapiclient/info.py @@ -1,27 +1,25 @@ -def info(SERVER, DATASET, PARAMETERS, opts): +def info(server, dataset, parameters, opts): - from hapiclient.util import urlopen, subset_meta, unicode_check, fix_parameters, query_name - from hapiclient.cache import meta_cache_read, meta_cache_write - from hapiclient.catalog import catalog + import hapiclient as hc - unicode_check(DATASET, PARAMETERS) - PARAMETERS = fix_parameters(PARAMETERS) + hc.util.unicode_check(dataset, parameters) + parameters = hc.util.fix_parameters(parameters) - meta = meta_cache_read(SERVER, DATASET, opts) + meta = hc.cache.meta_cache_read(server, dataset, 'info', opts) if meta is not None: return meta - cat = catalog(SERVER) - url = SERVER + '/info?' + query_name(cat, 'dataset') + '=' + DATASET - meta = urlopen(url, parse_json=True) + cat = hc.catalog.catalog(server, opts) + url = server + '/info?' + hc.util.query_name(cat, 'dataset') + '=' + dataset + meta = hc.util.urlopen(url, parse_json=True) - meta_cache_write(meta, SERVER, DATASET, opts) + hc.cache.meta_cache_write(meta, server, dataset, 'info', opts) - meta.update({"x_server": SERVER}) - meta.update({"x_dataset": DATASET}) + meta.update({"x_server": server}) + meta.update({"x_dataset": dataset}) - if PARAMETERS is not None: - subset_meta(meta, PARAMETERS) + if parameters is not None: + hc.util.subset_meta(meta, parameters) return meta else: return meta diff --git a/hapiclient/util.py b/hapiclient/util.py index 13d91fc..869ae72 100644 --- a/hapiclient/util.py +++ b/hapiclient/util.py @@ -11,7 +11,7 @@ def setopts(defaults, given): # Override defaults for key, value in given.items(): - if type(given[key]) == dict: + if isinstance(given[key], dict): setopts(defaults[key], given[key]) continue if key in defaults: @@ -26,10 +26,11 @@ def jsonparse(res, url): """Try/catch of json.loads() function with short error message.""" from json import loads + try: return loads(res.read().decode('utf-8')) - except: - error('Could not parse JSON from %s' % url) + except Exception as e: + error('Could not parse JSON from %s: %s' % (url, str(e))) def pythonshell():