initial qubit-tools

qubittool/__main__.py

@@ -0,0 +1,69 @@
+import sys, os
+import PySimpleGUI as sg
+
+from tool import read_collect_run_doc_name_strids
+from tool import read_qubit_doc_by_accession
+from tool import read_qubit_doc_by_order, write_run_doc
+
+def make_window():
+    layout = [[sg.Text('Generate RunDoc Output worksheet with Qubit run data.')],
+          [sg.Text('Run Doc:   '),
+            sg.Combo(sg.user_settings_get_entry('-filenames1-', []), default_value=sg.user_settings_get_entry('-last filename-', ''), size=(50, 1), key='-FILENAME-'),
+           sg.FileBrowse()],
+           [sg.Text('Qubit Data:'),
+            sg.Combo(sg.user_settings_get_entry('-filenames2-', []), default_value=sg.user_settings_get_entry('-last qbtfilename-', ''), size=(50, 1), key='-QFILENAME-'),
+           sg.FileBrowse()],
+          [sg.Button('Go'), sg.Button('Exit')]]
+
+    return sg.Window('Qubit Tool Window', layout)
+
+def collect_state_from_serial_accession(values):
+    keys = {}
+    stringpath = values['-FILENAME-']
+    keys = read_collect_run_doc_name_strids(min_col=1, path=stringpath)
+    keys.update(read_collect_run_doc_name_strids(min_col=9, path=stringpath))
+    stringpath = values['-QFILENAME-']
+    read_qubit_doc_by_accession(keys, stringpath)
+    stringpath = values['-FILENAME-']
+    write_run_doc(keys, stringpath)
+
+def collect_state_from_quibit_order(values):
+    keys = {}
+    stringpath = values['-FILENAME-']
+    keys = read_collect_run_doc_name_strids(min_col=1, path=stringpath)
+    keys.update(read_collect_run_doc_name_strids(min_col=9, path=stringpath))
+    stringpath = values['-QFILENAME-']
+    read_qubit_doc_by_order(keys, stringpath)
+    stringpath = values['-FILENAME-']
+    write_run_doc(keys, stringpath)
+
+def main():
+    window = make_window()
+    while True:
+        event, values = window.read()
+
+        if event == sg.WIN_CLOSED or event == 'Exit':
+            break
+        if event == 'Go':
+            sg.user_settings_set_entry('-filenames1-', list(set(sg.user_settings_get_entry('-filenames1-', []) + [values['-FILENAME-'], ])))
+            sg.user_settings_set_entry('-last filename-', values['-FILENAME-'])
+            window['-FILENAME-'].update(values=list(set(sg.user_settings_get_entry('-filenames1-', []))))
+
+            sg.user_settings_set_entry('-filenames2-', list(set(sg.user_settings_get_entry('-filenames2-', []) + [values['-QFILENAME-'], ])))
+            sg.user_settings_set_entry('-last qbtfilename-', values['-QFILENAME-'])
+            window['-QFILENAME-'].update(values=list(set(sg.user_settings_get_entry('-filenames2-', []))))
+            
+            try:
+                collect_state_from_quibit_order(values)
+                window.Close()
+            except Exception as e:
+                sg.popup_error_with_traceback(f'Error check RunDoc info:', e)
+
+        elif event == 'Clear':
+            sg.user_settings_set_entry('-filenames-', [])
+            sg.user_settings_set_entry('-last filename-', '')
+            window['-FILENAME-'].update(values=[], value='')
+
+if __name__ == '__main__':
+    main()
+    sys.exit()

qubittool/rowmodel.py

@@ -0,0 +1,65 @@
+
+class RowModel:
+    def __init__(self, coordinate:str, accessionid:str):
+        self.coordinate = coordinate
+        self.accessionid = accessionid
+        self.customerid = None
+        self.plasmavolml = float(0.0)
+        self.qubitrunid = None
+        self.qubitassay = None
+        self.runvalngml = None
+        self.testdate = None
+        self.tubeconc = None
+        self.sampleconc	= None
+
+
+    def get_Id(self):
+        return self.accessionid
+
+    def set_plasmavolml(self, volml:str):
+        self.plasmavolml = volml
+
+    def get_plasmavolml(self):
+        return self.plasmavolml 
+
+    def set_qubitrunid(self, runid:str):
+        self.qubitrunid = runid
+
+    def get_qubitrunid(self):
+        return self.qubitrunid
+
+    def set_runvalngml(self, volml:str):
+        self.runvalngml = volml
+
+    def get_runvalngml(self):
+        return self.runvalngml
+    
+    def set_customerid(self, custid:str):
+        self.customerid = custid
+
+    def get_customerid(self):
+        return self.customerid
+    
+    def set_qubitassay(self, assay:str):
+        self.qubitassay = assay
+
+    def get_qubitassay(self):
+        return self.qubitassay
+
+    def set_testdate(self, datestr:str):
+        self.testdate = datestr
+
+    def get_testdate(self):
+        return self.testdate
+
+    def set_tubeconc(self, conc:str):
+        self.tubeconc = conc
+    
+    def get_tubeconc(self):
+        return self.tubeconc
+
+    def set_sampleconc(self, conc:str):
+        self.sampleconc = conc
+    
+    def get_sampleconc(self):
+        return self.sampleconc

qubittool/tool.py

@@ -0,0 +1,163 @@
+import sys, os
+import pandas as pd
+import csv
+import openpyxl
+from openpyxl import Workbook, load_workbook
+from openpyxl.worksheet.table import Table, TableStyleInfo
+from openpyxl.worksheet.dimensions import ColumnDimension
+from openpyxl.worksheet.worksheet import Worksheet
+from openpyxl.utils import cell
+from openpyxl.styles import PatternFill, Border, Side, Alignment
+
+from rowmodel import RowModel
+
+def read_qubit_doc_by_accession(keys:dict, path:str):
+    # collect values_by_sample_names string ids #
+    with open(path, 'r', newline='', encoding='utf-8') as csvfile:
+        reader = csv.DictReader(csvfile)        
+        for row in reader:
+            try:
+                if keys[ row["Test Name"] ]: 
+                    rmv:RowModel = keys[ row["Test Name"] ]
+                    rmv.set_testdate(row["Test Date"])
+                    #rmv.set_qubitrunid(row["Run ID"])
+                    rmv.set_runvalngml(row["Green RFU"])
+                    rmv.set_sampleconc(row["Original sample conc."])
+                    rmv.set_tubeconc(row["Qubit tube conc."])
+                    print("{}:{}".format(rmv.get_qubitrunid(), rmv.get_Id()))
+            except KeyError:
+                pass
+    return
+
+def read_qubit_doc_by_order(keys:dict, path:str):
+    # collect values_by_sample_names string ids #
+    with open(path, 'r', newline='', encoding='utf-8') as csvfile:
+        reader = csv.DictReader(csvfile)        
+        rowlist:list = list()
+        for row in reader:
+            rowlist.append(row)
+        if len( rowlist ) < 1:
+            return
+        indx:int = len( rowlist ) - 1
+        for key in keys:
+            try:
+                row = rowlist[indx]
+                if row is None:
+                    return
+                indx = indx - 1
+                if row["Test Name"]: 
+                    rmv:RowModel = keys[key]
+                    rmv.set_testdate(row["Test Date"])
+                    #rmv.set_qubitrunid(row["Run ID"])
+                    rmv.set_runvalngml(row["Green RFU"])
+                    rmv.set_sampleconc(row["Original sample conc."])
+                    rmv.set_tubeconc(row["Qubit tube conc."])
+                    print("{}:{}".format(rmv.get_qubitrunid(), rmv.get_Id()))
+            except KeyError:
+                pass
+            except IndexError:
+                return
+    return
+
+def write_run_doc(dm:dict, path:str):
+    # write excel run_doc section with values_from_qubit by string ids #
+    sheetdata:list = []
+    wb = openpyxl.load_workbook(path, read_only=False)
+    sheets = wb.sheetnames
+    ws:Worksheet = None
+    for sheetname in sheets:
+        if sheetname == "Output":
+            ws = wb.get_sheet_by_name(sheetname)
+            break
+    if ws == None: 
+        ws = wb.create_sheet(title="Output")
+        ws.title = "Output"
+        wb.worksheets.append(ws)
+    
+    ws.append( ["No.", "SampleID", "Sysmex Inostics ID (Accession ID)","Secondary ID (Customer ID)","Plasma Vol. [mL]","Qubit run ID","ng/µL","GE/µL"] )
+    ws.column_dimensions["C"].width = 19.23
+    ws.column_dimensions["D"].width = 18.23
+    ws.column_dimensions["E"].width = 15.23
+    ws.column_dimensions["F"].width = 14.23
+    ws.column_dimensions["G"].width = 18.23
+    ix:int = 0
+    for r in dm:
+        model:RowModel = dm[r]
+        ix=ix+1
+        ws.append([ix, "", model.get_Id(), model.get_customerid(), model.get_plasmavolml(), model.get_qubitrunid(), 
+        model.get_sampleconc(), convert_to_GE(model.get_sampleconc())])
+
+    grayFill = PatternFill(start_color='D0CECE', end_color='D0CECE',
+                       fill_type='solid')
+
+    for columns in ws.iter_cols(min_col=1, max_col=2, min_row=None, max_row=None):
+        for cell in columns:
+            cell.fill = grayFill
+
+    wb.save(path)
+    wb.close()
+    return
+
+def convert_to_numeric(vlu, stndv:float=0.0):
+    try:
+        return float(vlu)
+    except Exception as e:
+        return decimal.Decimal(stndv)
+
+def convert_to_GE(vlu, stndfv:float=0.0033):
+    try:
+        return float(vlu) / stndfv
+    except Exception as e:
+        return "undefined"
+
+def read_collect_run_doc_name_strids(min_col:int=1, path:str=None):
+    wb = openpyxl.load_workbook(path, read_only=True)
+    ws = wb.active
+    if ws.title != "Documentation":
+       raise Exception("Run Doc not found, set Active worksheet")
+
+    ProtocolRunID = ws.cell(3,3).value
+    InosticsID = None
+    CustomerID = None
+    PlasmaID = None
+    InosticsIDKeys = {}
+    for row in ws.iter_rows(max_row=30, min_col=min_col):
+        for cell in row:
+            if InosticsID == None and cell.value and isinstance(cell.value, str) and "Inostics ID" in cell.value:
+                InosticsID = cell.column
+                continue
+            
+            if CustomerID == None and cell.value and isinstance(cell.value, str) and "Relevant Customer ID" in cell.value:
+                CustomerID = cell.column
+                continue
+
+            if PlasmaID == None and cell.value and isinstance(cell.value, str) and "Plasma Vol" in cell.value:
+                PlasmaID = cell.column
+                continue
+            
+            if cell.value and InosticsID == cell.column:
+                InosticsIDKeys[cell.value] = RowModel(cell.coordinate, cell.value)
+                rmv:RowModel = InosticsIDKeys[cell.value]
+                if ProtocolRunID != None :
+                    rmv.set_qubitrunid(ProtocolRunID)
+                continue
+
+            if cell.value and CustomerID == cell.column:
+                print(F"{cell.coordinate}:{cell.row}x{cell.column}={cell.value}") 
+                try:
+                    if InosticsIDKeys[row[InosticsID-1].value] != None: 
+                        rmv:RowModel = InosticsIDKeys[row[InosticsID-1].value]
+                        rmv.set_customerid( cell.value )
+                except:
+                    pass
+
+            if cell.value and PlasmaID == cell.column:
+                print(F"{cell.coordinate}:{cell.row}x{cell.column}={cell.value}") 
+                try:
+                    if InosticsIDKeys[row[InosticsID-1].value] != None: 
+                        rmv:RowModel = InosticsIDKeys[row[InosticsID-1].value]
+                        rmv.set_plasmavolml( convert_to_numeric(cell.value) )
+                except:
+                    pass
+    wb.close()
+    return InosticsIDKeys