Bulk Scan Updates

microsetta_admin/server.py

@@ -3,6 +3,7 @@
 import secrets
 from datetime import datetime
 import io
+import random
 
 from jwt import PyJWTError
 from werkzeug.exceptions import BadRequest
@@ -774,6 +775,220 @@ def scan():
                                       recorded_type)
 
 
+def _get_color_code():
+    # val = '0123456789ABCDEF'
+    # code = ["#"+''.join([random.choice(val) for i in range(6)])]
+    codes = ["#C0C0C0", "#808080", "#800000", "#FF0000", "#800080", "#FF00FF",
+             "#008000", "#00FF00", "#FFd700", "#000080", "#0000FF", "#00FFFF",
+             "#FFE4C4", "#7FFFD4", "#DEB887", "#FF7F50", "#6495ED", "#FF8C00",
+             "#FF1493", "#00BFFF", "#1E90FF", "#DAA520", "#4B0082", "#F0E68C",
+             "#90EE90", "#778899", "#FFA500", "#BC8F8F", "#D8BFD8", "#DCDCDC"]
+
+    return codes[random.randrange(0, len(codes)-1)]
+
+
+def _get_legends(criteria):
+
+    dict = {}
+    if criteria not in session:
+        status, fields = APIRequest.get(
+            '/api/admin/barcode_query_fields')
+
+        if status == 200:
+            for obj in fields:
+                if obj["label"] == criteria:
+                    for val in obj["values"]:
+                        color_code = None
+                        if criteria == "Project":
+                            if not ('-' in obj["values"][val]):
+                                color_code = _get_color_code()
+                        elif "Sample" in criteria:
+                            color_code = _get_color_code()
+
+                        if color_code is not None:
+                            while color_code in dict.values():
+                                color_code = _get_color_code()
+
+                            if criteria == "Sample Status":
+                                tmp = obj["values"][val].replace(' ', '-')
+                                tmp = tmp.lower()
+                                dict[tmp] = color_code
+                            else:
+                                dict[obj["values"][val]] = color_code
+
+            if len(dict) > 0:
+                dict['None'] = "#000000"
+                session[criteria] = dict
+    else:
+        dict = session.get(criteria)
+
+    return dict
+
+
+def _compose_table(legends, type):
+    output_dict = {}
+    scanned_data = session.get('scan_data')
+    current_row = ""
+    current_cols = []
+
+    for rec in scanned_data:
+        status, data = APIRequest.get(
+            '/api/admin/search/samples/%s' % rec[5]
+        )
+
+        val = ""
+        proj = data["sample"]["sample_projects"][0]
+        status = data["sample"]["_latest_scan_status"]
+        src = data["sample"]["site"]
+
+        if type == "Sample Site":
+            val = data["sample"]["site"]
+        elif type == "Sample Status":
+            val = data["sample"]["_latest_scan_status"]
+        elif type == "Project":
+            val = data["sample"]["sample_projects"][0]
+
+        if current_row == "":
+            current_row = rec[4]
+            if val is None or val == "":
+                tmp = legends['None']
+                current_cols.append([rec[5], tmp, proj, status, src])
+            else:
+                current_cols.append([rec[5], legends[val], proj, status, src])
+        elif current_row == rec[4]:
+            if val is None:
+                tmp = legends['None']
+                current_cols.append([rec[5], tmp, proj, status, src])
+            else:
+                current_cols.append([rec[5], legends[val], proj, status, src])
+        else:
+            output_dict[current_row] = current_cols
+            current_cols = []
+            current_row = rec[4]
+            if val is None:
+                tmp = legends['None']
+                current_cols.append([rec[5], tmp, proj, status, src])
+            else:
+                current_cols.append([rec[5], legends[val], proj, status, src])
+
+    output_dict[current_row] = current_cols
+    return output_dict
+
+
+def _visualize_scans():
+    filename = session.get('scan_file', "")
+    criteria = request.form['sort_criteria']
+    session['recent_criteria'] = criteria
+    legends = _get_legends(criteria)
+
+    return render_template('bulk_scan.html',
+                           **build_login_variables(),
+                           stage="Visualization",
+                           filename=filename,
+                           data=criteria,
+                           legends=legends,
+                           table=_compose_table(legends, criteria),
+                           status_options=STATUS_OPTIONS
+                           )
+
+
+def _get_barcode_rack(barcode):
+    data = session.get('scan_data')
+    obj = {}
+
+    for rec in data:
+        if rec[5] == barcode:
+            obj["rack_id"] = rec[6]
+            obj["location_col"] = str(int(rec[3]))
+            obj["location_row"] = rec[4]
+            return obj
+
+    return obj
+
+
+def _post_bulk_scan_add():
+    send_email = request.form.get('send_email', False)
+    session[SEND_EMAIL_CHECKBOX_DEFAULT_NAME] = send_email
+
+    sample_barcode = request.form['sample_barcode']
+    technician_notes = request.form['technician_notes']
+    sample_status = request.form['sample_status']
+
+    # Do the actual update
+    status, response = APIRequest.post(
+        '/api/admin/scan/%s' % sample_barcode,
+        json={
+            "sample_status": sample_status,
+            "technician_notes": technician_notes
+        }
+    )
+
+    if status == 201:
+        data = _get_barcode_rack(sample_barcode)
+        if bool(data):
+            status, response = APIRequest.post(
+                '/api/admin/rack/%s/add' % sample_barcode,
+                json=data
+            )
+
+    filename = session.get('scan_file', "")
+    criteria = session.get('recent_criteria', "Project")
+    legends = _get_legends(criteria)
+    return render_template('bulk_scan.html',
+                           **build_login_variables(),
+                           stage="Visualization",
+                           filename=filename,
+                           data=criteria,
+                           legends=legends,
+                           table=_compose_table(legends, criteria),
+                           status_options=STATUS_OPTIONS
+                           )
+
+
+def _post_bulk_scan():
+    obj_file = request.files['file_picker']
+
+    if obj_file is not None:
+        filename = obj_file.filename
+        ds = pd.read_csv(obj_file, dtype={'TubeCode': str})
+        file_data = ds.values.tolist()
+        session['scan_data'] = file_data
+        session['scan_file'] = filename
+        return render_template('bulk_scan.html',
+                               **build_login_variables(),
+                               stage="Visualization",
+                               filename=filename,
+                               data="",
+                               legends={},
+                               table={},
+                               status_options={}
+                               )
+
+
+@app.route('/scan-bulk', methods=['GET', 'POST'])
+def bulk_scan():
+
+    if request.method == 'GET':
+        return render_template('bulk_scan.html',
+                               **build_login_variables(),
+                               stage="Scanning",
+                               filename="",
+                               data="",
+                               legends={},
+                               table={},
+                               status_options={}
+                               )
+
+    if request.method == 'POST':
+        stage = request.form["stage"]
+        if stage == "Scanning":
+            return _post_bulk_scan()
+        elif stage == "Visualization":
+            return _visualize_scans()
+        elif stage == "Sample_Scan":
+            return _post_bulk_scan_add()
+
+
 @app.route('/metadata_pulldown', methods=['GET', 'POST'])
 def metadata_pulldown():
     allow_missing = request.form.get('allow_missing_samples', False)