""" Finds the multiprocessing process and gets the result.

with g_process_lock:

if id not in g_processes:

return json.dumps({'error': api_utils.invalid_session_id_err_str})

pdata = g_processes[id]

# Check to see this processes has shutdown

if not pdata['shutdown'].empty():

# This process has shutdown, tell the requester that it needs to restart it.

del g_processes[id]

result = json.dumps({'error': 'Session has timed out, you need to reload the page.', 'session_timeout': True})

return result

pdata['input'].put(SessionRequest(function_name, args))

result = pdata['output'].get()

if isinstance(result, dict):

# Unpack and process DB transactions.

if 'db_insert' in result:

print('need to do DB transaction...')

insert_db(result['db_insert']['sql'], result['db_insert']['vals'])

print('done with DB transaction, returning')

elif 'db_insert_multi' in result:

print('need to do multi-DB transaction...')

for dbi in result['db_insert_multi']:

insert_db(dbi['sql'], dbi['vals'])

print('done with multi-DB transaction, returning')

return result['json']

else:

# Already a string.

""" Inputs come in this form. """

def __init__(self, function_name, args):

self.function_name = function_name

""" Each UUID session has a worker process which holds the DataFrame for that process in its memeory.

def __init__(self, id, datapath, gdrive_file_info, input_queue,

output_queue, shutdown_queue, subsets_from_db, math_vars):

super(mp.Process, self).__init__()

self.id = id

self.path = datapath

self.gdrive_file_info = gdrive_file_info

self.input_queue = input_queue

self.output_queue = output_queue

self.shutdown_queue = shutdown_queue

self.subsets_from_db = subsets_from_db

self.loading_progress_queue = mp.Queue()

self.gdrive_progress_queue = mp.Queue()

self.data_lock = mp.Lock()

self.data = dict()

self.last_progress = {

'progress': 0,

'rows_loaded': 0,

'total_rows': 0,

'math_vars_loaded': None,

'math_vars_total': None,

'text': 'Opening file...',

}

self.last_gdrive_progress = {

'progress': 0.0,

'downloaded_bytes': 0.0,

'total_bytes': 0.0,

'in_cache': False

}

self.pending_requests = []

self.pending_download_files = {}

self.gdrive = None

self.math_vars = math_vars

self.funcs = {

'processing_progress': self.processing_progress,

'get_subsets': self.get_subsets,

'get_columns': self.get_columns,

'get_non_numeric_columns': self.get_non_numeric_columns,

'plot_json': self.plot_json,

'select_data': self.select_data,

'delete_subset': self.delete_subset,

'do_math': self.do_math,

'download_subset': self.download_subset,

'download_file': self.download_file,

'filter': self.filter,

'get_unique_strings': self.get_unique_strings,

'cloud_progress': self.gdrive_progress,

'calc_correlation': self.calc_correlation,

'bulk_import': self.bulk_import,

'levenshtein_filter': self.levenshtein_filter,

}

def run(self):

if self.gdrive_file_info is not None and plotplot_config.get_boolean_with_default('google drive', 'google_drive_connection_enabled', False):

self.gdrive = PlotplotGdrive()

self.start_gdrive_download()

else:

# Start the data loading thread.

self.start_loading_data()

while True:

if len(self.pending_requests) < 1:

# Blocking call, waiting for inputs.

try:

req = self.input_queue.get(timeout=26 * 3600)

except Empty:

print(f'Session {self.id} queue timeout, shutting down.')

break

self.pending_requests.append(req)

# Get any additional requests.

while not self.input_queue.empty():

self.pending_requests.append(self.input_queue.get())

request = self.pending_requests.pop(0)

if request.function_name not in self.funcs:

self.output_queue.put(

json.dumps({

'error':

'Requested function: ' + request.function_name +

' not in list of functions.'

}))

else:

if ENABLE_LINE_PROFILER:

lp = LineProfiler()

lp_wrapper = lp(self.funcs[request.function_name])

try:

if ENABLE_LINE_PROFILER:

result = lp_wrapper(request.args)

else:

result = self.funcs[request.function_name](request.args)

except BaseException as e:

tb = traceback.format_exc()

print(e)

print(str(tb))

result = json.dumps(dict(error=str(e) + '\n\n' + str(tb)))

if ENABLE_LINE_PROFILER:

lp.print_stats(output_unit=0.001)

self.output_queue.put(result)

self.shutdown_queue.put(True)

print(f'Session {self.id} end of thread.')

def processing_progress(self, args):

with self.data_lock:

if 'df' in self.data:

return json.dumps({

'progress': 1,

'rows_loaded': None,

'total_rows': None,

'data_id': self.id,

'done': True,

})

# The processing isn't done yet, get the progress.

if self.loading_progress_queue.empty():

progress_data = self.last_progress

else:

while not self.loading_progress_queue.empty():

progress_data = self.loading_progress_queue.get()

self.last_progress = progress_data

progress_data['data_id'] = self.id

progress_data['done'] = False

return json.dumps(progress_data)

def gdrive_progress(self, args):

if self.gdrive_progress_queue.empty():

gdrive_progress = self.last_gdrive_progress

else:

while not self.gdrive_progress_queue.empty():

gdrive_progress = self.gdrive_progress_queue.get()

self.last_gdrive_progress = gdrive_progress

return json.dumps(gdrive_progress)

def start_gdrive_download(self):

# Get the file from google drive.

output_path = api_utils.get_gdrive_path(self.gdrive_file_info['name'])

# See if the file already exists.

if os.path.exists(output_path) and pathlib.Path(

output_path).is_file() and os.path.getsize(output_path) == int(

self.gdrive_file_info['size']):

# File exists and is the same size. We'll use it.

self.gdrive_progress_queue.put({

'progress':

1,

'downloaded_bytes':

self.gdrive_file_info['size'],

'total_bytes':

self.gdrive_file_info['size'],

'in_cache':

True,

})

print('Using cached gdrive file')

self.start_loading_data()

else:

os.makedirs(api_utils.gdrive_folder, exist_ok=True)

gdrive_download_thread = GdriveDownloadThread(

self.gdrive, self.gdrive_progress_queue, self.gdrive_file_info,

api_utils.gdrive_folder, self.start_loading_data)

self.path = output_path

gdrive_download_thread.start()

def start_loading_data(self):

load_thread = LoadCsvThread(self.loading_progress_queue, self.path,

self.subsets_from_db, self.data,

self.data_lock, self.math_vars)

load_thread.start()

def get_data(self):

with self.data_lock:

if 'df' not in self.data:

return None, None, None, None, None

return self.data['df'], self.data['subsets'], self.data[

'subset_counter'], self.data['math_vars'], self.data['col_labels']

def get_subsets(self, args):

return json.dumps(self.get_subsets_no_json(args))

def get_subsets_no_json(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

subsets_out = {}

for key in subsets.keys():

subsets_out[key] = api_utils.export_subset(subsets, key)

return subsets_out

def get_columns(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

all_cols = df.columns.tolist()

numeric_cols = api_utils.get_numeric_cols(df)

non_numeric = np.setdiff1d(all_cols, numeric_cols).tolist()

#return json.dumps(api_utils.get_numeric_cols(df))

return json.dumps({

'numeric': sorted(numeric_cols, key=str.lower),

'non_numeric': sorted(non_numeric, key=str.lower),

'col_labels': col_labels

})

def get_non_numeric_columns(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

#return jsonify(df.columns.tolist())

all_cols = df.columns.tolist()

numeric_cols = api_utils.get_numeric_cols(df)

non_numeric = np.setdiff1d(all_cols, numeric_cols).tolist()

return json.dumps(non_numeric)

def future_plot_in_requests(self, args):

key = args['data']['key']

for r in self.pending_requests:

if r.function_name == 'plot_json' and r.args['data']['key'] == key:

return True

return False

def calc_correlation(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

correlations = {}

data = args['data']

subset_ids = api_utils.get_arrayp(data, 'subset_ids')

x = api_utils.get_strp(data, 'x')

y = api_utils.get_strp(data, 'y')

if x is None or y is None:

# No correlation.

out = {}

methods = ['pearson', 'kendall', 'spearman']

for m in methods:

out[m] = (0, 0)

return json.dumps(out), 200, {

'Content-Type': 'application/json; charset=utf-8'

}

for sub_id in subset_ids:

if sub_id not in subsets:

return json.dumps({'error': 'Unknown subset: ' + str(sub_id)})

all_subsets = np.zeros(len(df), dtype=bool)

for sub_id in subset_ids:

all_subsets |= subsets[sub_id]['idx']

correlations = api_utils.calculate_correlation(df[all_subsets], x, y)

jsonResult = json.dumps(correlations)

return jsonResult, 200, {

'Content-Type': 'application/json; charset=utf-8'

}

def plot_json(self, args):

plot_supports_hovering = False

longest_col_name_len = 0

if self.future_plot_in_requests(args):

print('Plot preempted!')

return json.dumps(

{'preempt': 'Plot preempted by another request.'})

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

current_user_email = args['current_user_email']

data = args['data']

print('Plot generation for ' + current_user_email + '...')

# Constants

max_rows_for_bbox = 100000

valid_percent_for_nans_cutoff = 0.90 # > 10% NaNs? Show NaN plot.

x = api_utils.get_strp(data, 'x')

y = api_utils.get_strp(data, 'y')

z = api_utils.get_strp(data, 'z')

if z == '':

z = None

xlog = api_utils.get_boolp(data, 'xlog')

ylog = api_utils.get_boolp(data, 'ylog')

subset_ids = api_utils.get_arrayp(data, 'subsets')

xmin = api_utils.get_floatp(data, 'xmin')

xmax = api_utils.get_floatp(data, 'xmax')

ymin = api_utils.get_floatp(data, 'ymin')

ymax = api_utils.get_floatp(data, 'ymax')

hist_type = api_utils.get_strp(data, 'hist_type')

nans_request = api_utils.get_strp(data, 'nans_request')

hoverlist = data['hoverlist']

xmargin = api_utils.get_floatp(data, 'xmargin')

if xmargin is None:

xmargin = 0

ymargin = api_utils.get_floatp(data, 'ymargin')

if ymargin is None:

ymargin = 0

use_bbox = True

bbox_valid = True

if xmin is None or xmax is None or ymin is None or ymax is None:

# Bounding box is invalid.

use_bbox = False

bbox_valid = False

xmin_margin = None

xmax_margin = None

ymin_margin = None

ymax_margin = None

if bbox_valid:

xmin_margin = xmin - xmargin

ymin_margin = ymin - ymargin

xmax_margin = xmax + xmargin

ymax_margin = ymax + ymargin

if len(df) < max_rows_for_bbox:

use_bbox = False

fig = go.Figure()

fig.update_layout(hovermode=False)

if subset_ids is None:

idx = pd.Series(np.ones(len(df), dtype=bool))

subset_ids = [0]

else:

for subset in subset_ids:

if subset not in subsets:

return json.dumps(

{'error': str(subset) + ' not in known subsets.'})

idxs = []

for sub_id in subset_ids:

idxs.append(subsets[sub_id]['idx'])

plot_type = data['plot_type']

is_heatmap = False

both_nan = 0

num_nan = 0

showing_nan_plots = False

if plot_type == 'histogram' and x is not None:

# Make a histogram!

fig, minmax = api_utils.generate_histogram(df, idxs, fig, x,

data['nbins'],

hist_type)

plot_type = 'histogram'

if hist_type == 'count':

y = 'Count'

elif hist_type == 'proportion':

y = 'Proportion'

else:

assert 'Unknown hist type: "' + hist_type + '"'

elif plot_type == 'rank' and y is not None:

df2 = api_utils.get_rank_df(df, y, z, hoverlist)

fig, minmax, num_nan, is_heatmap, plot_supports_hovering, longest_col_name_len = api_utils.generate_scatter(

df2, api_utils.rank_col_name, y, z, idxs, fig, use_bbox, xmin_margin, xmax_margin,

ymin_margin, ymax_margin, xlog, ylog, hoverlist=hoverlist)

x = 'Rank'

elif x is None or y is None:

scatter = go.Scattergl(x=[], y=[], mode='markers')

fig.add_trace(scatter)

if x is None:

x = 'Drop a variable'

if y is None:

y = 'Drop a variable'

minmax = [(-1, 5), (-1, 5)]

else:

min_valid_percent = 1

for this_idx in idxs:

print(df[this_idx][x])

min_valid_percent = min(

min_valid_percent,

np.sum((np.isfinite(df[this_idx][x]))

& (np.isfinite(df[this_idx][y]))) /

np.sum(this_idx))

if nans_request == 'hide' or (

nans_request == 'auto'

and min_valid_percent > valid_percent_for_nans_cutoff):

fig, minmax, num_nan, is_heatmap, plot_supports_hovering, longest_col_name_len = api_utils.generate_scatter(

df,

x,

y,

z,

idxs,

fig,

use_bbox,

xmin_margin,

xmax_margin,

ymin_margin,

ymax_margin,

xlog,

ylog,

hoverlist=hoverlist)

else:

fig, minmax, both_nan, num_nan, is_heatmap, plot_supports_hovering, longest_col_name_len = api_utils.generate_scatter_and_nans(

df,

x,

y,

z,

idxs,

fig,

use_bbox,

xmin_margin,

xmax_margin,

ymin_margin,

ymax_margin,

xlog,

ylog,

hoverlist=hoverlist)

showing_nan_plots = True

fig.update_layout(bargap=0.0)

fig.layout['dragmode'] = 'pan'

fig.update_layout( #template='plotly_white', don't set the template here, it costs 0.05 seconds

font=dict(size=16))

if plot_type != 'histogram' and showing_nan_plots == False:

fig.update_yaxes(title_text=y,

type='linear',

scaleanchor='x',

scaleratio=1,

autorange=False,

automargin=False)

fig.update_xaxes(title_text=x,

type='linear',

autorange=False,

automargin=False)

elif showing_nan_plots == False:

fig.update_yaxes(title_text=y,

type='linear',

autorange=False,

automargin=False)

fig.update_xaxes(title_text=x,

type='linear',

autorange=False,

automargin=False)

if bbox_valid:

metadata = {

'validRanges': {

'xmin': xmin_margin,

'xmax': xmax_margin,

'ymin': ymin_margin,

'ymax': ymax_margin

}

}

else:

metadata = {

'validRanges': {

'xmin': None,

'xmax': None,

'ymin': None,

'ymax': None

}

}

metadata['plot_supports_hovering'] = plot_supports_hovering

subset_export = {}

for sub_id in subset_ids:

subset_export[sub_id] = api_utils.export_subset(subsets, sub_id)

metadata['subsets'] = subset_export

metadata['plot_type'] = plot_type

metadata['is_heatmap'] = is_heatmap

metadata['both_nan'] = both_nan

metadata['num_nan'] = num_nan

metadata['hoverbox_longest_col_name_len'] = longest_col_name_len

metadata['minmax'] = {

'xmin': minmax[0][0],

'xmax': minmax[0][1],

'ymin': minmax[1][0],

'ymax': minmax[1][1],

}

if len(minmax) > 2:

metadata['minmax']['cmin'] = minmax[2][0]

metadata['minmax']['cmax'] = minmax[2][1]

# fig.add_annotation(dict(font=dict(color='black', size=15),

# x=1,

# y=1,

# showarrow=False,

# text=str(total_nans) + " NaN / Infs",

# textangle=0,

# xanchor='right',

# xref="paper",

# yref="paper"))

jsonResult = json.dumps([fig, metadata],

cls=plotly.utils.PlotlyJSONEncoder)

# total_len = 0

# for idx in idxs:

# total_len += len(df[idx])

# print('Plot generation took',

# time.time() - start, 'seconds for ', total_len, 'rows (',

# api_utils.heatmap_x, 'x', api_utils.heatmap_y, ')')

return jsonResult, 200, {

'Content-Type': 'application/json; charset=utf-8'

}

def select_data(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

data = args['data']

colx = api_utils.get_strp(data, 'colx')

coly = api_utils.get_strp(data, 'coly')

colz = api_utils.get_strp(data, 'colz')

plot_type = api_utils.get_strp(data, 'plot_type')

subset_ids = api_utils.get_arrayp(data, 'subset_ids')

include_both_nans = api_utils.get_boolp(data, 'bothNanSelected')

color_slider_values = api_utils.get_arrayp(data, 'colorSliderValuesFinal')

nan_selection = None

if 'nanSelection' in data:

nan_selection = data['nanSelection']

if plot_type == 'rank':

if coly is None or subset_ids is None:

return json.dumps({'error': 'Input data null.'})

elif colx is None or coly is None or subset_ids is None:

return json.dumps({'error': 'Input data null.'})

polygons = data['polygons']

for sub_id in subset_ids:

if sub_id not in subsets:

return json.dumps({'error': 'Unknown subset: ' + str(sub_id)})

if plot_type == 'rank':

df2 = api_utils.get_rank_df(df, coly, None, [])

colx = api_utils.rank_col_name

else:

df2 = df

in_any_poly = pd.Series(np.zeros(len(df2), dtype=bool))

if len(polygons) < 1:

# no polygons, so only do a third var check

assert colz is not None and color_slider_values is not None and len(color_slider_values) == 2, 'No polygons and no 3rd variable data.'

in_any_poly = pd.Series(np.ones(len(df2), dtype=bool))

else:

for poly in polygons:

# Create a closed polygon.

polygon = []

for i in range(0, len(poly)):

polygon.append((poly[i][0], poly[i][1]))

# Close the polygon.

polygon.append(polygon[0])

# Compute which points are inside the polygon.

in_poly = point_in_polygon.pd_inside_polygon(

df2, colx, coly, polygon)

in_any_poly = (in_any_poly | in_poly)

if nan_selection is not None:

# Add indicies that are selected from the NaN graphs

top_xmin = nan_selection[0]

top_xmax = nan_selection[1]

right_ymin = nan_selection[2]

right_ymax = nan_selection[3]

idx = np.isfinite(df2[colx]) & ~np.isfinite(df2[coly]) & (

df2[colx] > top_xmin) & (df2[colx] < top_xmax)

in_any_poly = (in_any_poly | idx)

idx = np.isfinite(df2[coly]) & ~np.isfinite(df2[colx]) & (

df2[coly] > right_ymin) & (df2[coly] < right_ymax)

in_any_poly = (in_any_poly | idx)

if include_both_nans:

idx = ~np.isfinite(df2[colx]) & ~np.isfinite(df2[coly])

in_any_poly = (in_any_poly | idx)

if colz is not None and len(colz) > 0 and color_slider_values is not None and len(color_slider_values) == 2 and color_slider_values[0] is not None and color_slider_values[1] is not None:

minval = min(color_slider_values[0], color_slider_values[1])

maxval = max(color_slider_values[0], color_slider_values[1])

print(f'min {np.min(df2[colz])} max {np.max(df2[colz])}')

print(f'before {np.sum(in_any_poly)}')

in_any_poly = in_any_poly & (df2[colz] >= minval) & (df2[colz] <= maxval)

print(f'after {np.sum(in_any_poly)}')

# Create a new column index with only these data.

all_subsets = np.zeros(len(df2), dtype=bool)

for sub_id in subset_ids:

all_subsets |= subsets[sub_id]['idx']

in_subset = (in_any_poly & all_subsets)

new_subset_id, db_insert, count = self.add_subset(in_subset, subset_counter)

subsets_out = {}

for key in subsets.keys():

subsets_out[key] = api_utils.export_subset(subsets, key)

json_out = json.dumps([new_subset_id, subsets_out])

return ({'db_insert': db_insert, 'json': json_out})

def add_subset(self, idx, new_subset_num):

count = int(np.sum(idx))

with self.data_lock:

self.data['subsets'][new_subset_num] = {

'idx': idx,

'count': count,

}

self.data['subset_counter'] += 1

# Add subset to the database.

sql = 'INSERT INTO subsets (session, id_in_session, name, pd_idx) VALUES (?, ?, ?, ?)'

vals = [self.id, new_subset_num, '-', idx.to_numpy()] # subset names are only tracked on the frontend now.

db_insert = {'sql': sql, 'vals': vals}

return new_subset_num, db_insert, count

def delete_subset(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

subset_id = args['subset_id']

if subset_id == 0:

# Don't allow deletion of the (all) subset.

return json.dumps({'error': 'Cannot delete (all) subset.'})

with self.data_lock:

del subsets[subset_id]

# Update the database.

sql = 'DELETE from subsets WHERE session = ? AND id_in_session = ?'

vals = [self.id, subset_id]

return {

'json': self.get_subsets({}),

'db_insert': {

'sql': sql,

'vals': vals

}

}

def do_math(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

data = args['data']

# subset_id = get_intp(data, 'subset_id')

# if subset_id is None:

# idx = pd.Series(np.ones(len(df), dtype=bool))

# subset_id = 0

# elif subset_id not in g_subsets:

# return('Error: ' + str(subset_id) + ' not in known subsets.')

# else:

# idx = g_subsets[subset_id]['idx']

# Get math data.

raw_expr = api_utils.get_strp(data, 'expr')

math_out = api_utils.do_math_helper(df, math_vars, raw_expr)

if 'error' in math_out:

return json.dumps({'error': math_out['error']})

name_expression = math_out['name']

new_col = math_out['new_col']

math_vars = math_out['math_vars']

with self.data_lock:

df[name_expression] = new_col

self.data['df'] = df

self.data['math_vars'] = math_vars

# Update the math database

# Each variable is a json

new_math_var_str = json.dumps(math_vars)

sql = 'UPDATE sessions SET math_vars = ? WHERE id = ?'

vals = [new_math_var_str, self.id]

return {

'json':

json.dumps({

'col_labels': col_labels,

'cols': api_utils.get_numeric_cols(df),

'new_var': name_expression

}),

'db_insert': {

'sql': sql,

'vals': vals

}

}

def download_subset(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

data = args['data']

subset_id = data['subset_id']

subset_names = None

if 'subsets' in data:

subset_names = data['subsets']

jupyter_filename = api_utils.get_strp(data, 'jupyterFilename')

if subset_id is None:

return json.dumps({'error': 'subset_id is invalid.'})

if subset_id not in subsets:

return json.dumps({'error': 'subset_id not found.'})

# Build an array of the columns that the user is requesting.

include = []

for col, val in data['nonNumericCols'].items():

if val:

include.append(col)

for col, val in data['numericCols'].items():

if val:

include.append(col)

# Order the columns like they are in the file.

output = []

for col in df.columns:

if col in include:

output.append(col)

if jupyter_filename is None or not plotplot_config.get_boolean_with_default('jupyter notebook export', 'jupyter_notebook_export_enabled', False):

csv = df[subsets[subset_id]['idx']][output].to_csv(index=False)

file_id = str(uuid.uuid4())

self.pending_download_files[file_id] = csv

return json.dumps({'file_id': file_id})

else:

jupyter_filename = secure_filename(jupyter_filename)

jupyter_filepath = os.path.join(plotplot_config.get_plotplot_config()['jupyter notebook export']['jupyter_notebook_export_path'], jupyter_filename)

df_out = df

if subset_id == 0:

# add a column for each subset

new_cols = {}

for sub_id in subsets:

col_name = f"plotplot_{subset_names[str(sub_id)]['name']}"

i = 1

new_col_name = col_name

# Loop until the new column name is not in df.columns

while new_col_name in df.columns:

new_col_name = f"{col_name}_{i}"

i += 1

new_cols[new_col_name] = subsets[sub_id]['idx']

output += [new_col_name]

df_out = df.assign(**new_cols)

# write the file to the path

df_out[subsets[subset_id]['idx']][output].to_csv(jupyter_filepath, index=False)

return json.dumps({'file_written': jupyter_filepath})

def download_file(self, args):

file_id = args['file_id']

if file_id not in self.pending_download_files:

return json.dumps({

'error':

'File not found, you may need to call download_subset first.'

})

return self.pending_download_files.pop(file_id)

def filter(self, args):

df, subsets, subset_counter, math_vars, col_labels = self.get_data()

if df is None:

return json.dumps({'error': api_utils.data_not_loaded_str})

data = args['data']

subset_id = data['subset_id']

filter_var = data['filter_var']

filters = data['filters']

add_subset = data['add_subset']

exact_match = data['exact_match']

add_multiple_subsets = api_utils.get_boolp(data, 'add_multiple_subsets')

use_contains = api_utils.get_boolp(data, 'use_contains')

# Can't set exact match and use_contains.

if exact_match and use_contains:

return json.dumps({'error': 'Cannot set exact_match = true and use_contains = true.'})

if subset_id is None:

return json.dumps({'error': 'subset_id is invalid.'})

if subset_id not in subsets:

return json.dumps({'error': 'subset_id not found.'})

# Ensure that the filter variable exists

if filter_var is None:

return json.dumps({'error': 'filter_var is invalid.'})

if filter_var not in df.columns:

return json.dumps(

{'error': 'fitler_var: ' + filter_var + ' not found'})

if filters is None:

return json.dumps({'error': 'filters is invalid.'})

if len(filters) < 1:

return json.dumps({'error': 'must supply at least one filter.'})

if len(filters) > 99:

return json.dumps({'error': 'too many filters'})

idx = subsets[subset_id]['idx']

filter_results = []

filter_idxs = []

for filt in filters:

f = filt['filter']

key = filt['key']

if f == '':

filter_results.append({'filter': f, 'rows': None, 'key': key})

continue

if exact_match:

f_idx = df[idx][filter_var].astype(str).str.upper() == f.upper()

else:

# Use a simplified regex language.

# * ---> .* "match anything"

# ? ---> . "match single character"

# [abc] ---> no change

reg = api_utils.translate_filter_to_regex(f)

if use_contains:

f_idx = df[idx][filter_var].astype(str).str.contains(reg,

na=False,

regex=True,

case=False)

else:

f_idx = df[idx][filter_var].astype(str).str.fullmatch(reg,

na=False,

case=False)

filter_results.append({

'filter': f,

'rows': int(np.sum(f_idx)),

'key': key

})

filter_idxs.append(f_idx)

out = {}

out_to_json = {}

all_idx = pd.Series(np.zeros(len(df), dtype=bool))

for i in range(len(filter_idxs)):

filter_idxs[i] = all_idx | filter_idxs[i]

if add_multiple_subsets:

return self.add_multiple_subsets(args, subsets, subset_counter, filter_var, filter_results, filter_idxs)

for this_idx in filter_idxs:

all_idx = all_idx | this_idx

total_rows = int(np.sum(all_idx))

if add_subset:

new_subset_id, db_insert, count = self.add_subset(all_idx,

subset_counter)

out['db_insert'] = db_insert

out_to_json['new_subset_id'] = new_subset_id

out_to_json['subsets'] = self.get_subsets_no_json({})

subsets_out = {}

for key in subsets.keys():

subsets_out[key] = api_utils.export_subset(subsets, key)

out_to_json['filter_results'] = filter_results

out_to_json['total_rows'] = total_rows

out['json'] = json.dumps(out_to_json)

return out