Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if verbose:
print('dataset_description.imas_version is missing: assuming IMAS version %s' % imas_version)
else:
print('%s IMAS version detected' % imas_version)
except Exception:
raise
except ImportError:
if imas_version is None:
imas_version = os.environ.get('IMAS_VERSION', omas_rcparams['default_imas_version'])
if not omas_rcparams['allow_fake_imas_fallback']:
raise
filename = os.sep.join([omas_rcparams['fake_imas_dir'], '%s_%s_%d_%d_v%s.pkl' % (user, machine, pulse, run, imas_versions.get(imas_version, imas_version))])
printe('Overloaded load_omas_imas: %s' % filename)
from . import load_omas_pkl
ods = load_omas_pkl(filename, consistency_check=False)
else:
try:
# see what paths have data
# NOTE: this is where the IDS.get operation occurs
fetch_paths, joined_fetch_paths = infer_fetch_paths(ids, occurrence=occurrence, paths=paths, time=time,
imas_version=imas_version, verbose=verbose)
# build omas data structure
ods = ODS(imas_version=imas_version, consistency_check=False)
for k, path in enumerate(fetch_paths):
if path[-1].endswith('_error_upper') or path[-1].endswith('_error_lower') or path[-1].endswith('_error_index'):
continue
if verbose and (k % int(numpy.ceil(len(fetch_paths) / 10)) == 0 or k == len(fetch_paths) - 1):
print('Loading {0:3.1f}%'.format(100 * float(k) / (len(fetch_paths) - 1)))
# get data from IDS
if pulse is None or run is None:
raise Exception('`pulse` and `run` must be specified')
printd('Loading from ITM (user:%s machine:%s pulse:%d run:%d, itm_version:%s)' % (user, machine, pulse, run, itm_version), topic='itm')
try:
cpo = itm_open(user=user, machine=machine, pulse=pulse, run=run, new=False, itm_version=itm_version)
except ImportError:
if not omas_rcparams['allow_fake_itm_fallback']:
raise
filename = os.sep.join([omas_rcparams['fake_itm_dir'], '%s_%s_%d_%d_v%s.pkl' % (user, machine, pulse, run, itm_versions.get(itm_version, itm_version))])
printe('Overloaded load_omas_itm: %s' % filename)
from . import load_omas_pkl
ods = load_omas_pkl(filename)
else:
try:
# if paths is None then figure out what CPO are available and get ready to retrieve everything
if paths is None:
paths = [[structure] for structure in list_structures(itm_version=itm_version)]
joined_paths = None # joined_paths==None means take everything, no need to filter out what is available based on what was requested
else:
joined_paths = map(l2i, paths)
# fetch relevant CPOs and find available signals
fetch_paths = []
for path in paths:
ds = path[0]
path = path[1:]