import functions_framework from firebase_functions import https_fn # import firebase_admin from firebase_admin import initialize_app, firestore import pandas as pd import numpy as np import datetime import sys import platform import os import io from flask import make_response, send_file from datetime import datetime import re import xlsxwriter import math import matplotlib.pyplot as plt from scipy.stats import linregress initialize_app() @functions_framework.http def performance(request): """HTTP Cloud Function. Args: request (flask.Request): The request object. Returns: The response text, or any set of values that can be turned into a Response object using `make_response` . """ if request.method == 'OPTIONS': headers = { 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Methods': 'GET, POST', # 'Access-Control-Allow-Headers': '*', # 'Access-Control-Max-Age': '3600' } return ('', 204, headers) headers = { 'Access-Control-Allow-Origin': '*' } request_json = request.get_json(silent=True) request_args = request.args if request_json and 'start_date' in request_json and 'end_date' in request_json: start_date = request_json['start_date'] end_date = request_json['end_date'] elif request_args and 'start_date' in request_args and 'end_date' in request_args: start_date = request_args['start_date'] end_date = request_args['end_date'] else: start_date = "2023-12-04" end_date = "2023-12-05" if 'file' not in request.files: print('No file part') else: file = request.files['file'] if file.filename == '': print('No selected file') # Save the file to a temporary location temp_filepath = '/tmp/temp_file.csv' file.save(temp_filepath) # Load the file into a Pandas DataFrame try: df_reference_device_raw = pd.read_csv(temp_filepath) # Adjust the read method based on your file type (e.g., read_excel for Excel files) # Now you can work with the DataFrame (e.g., perform analysis or display it) print(df_reference_device_raw.head()) print('File uploaded and loaded into DataFrame successfully') except Exception as e: print(f'Error loading the file: {str(e)}') finally: # Remove the temporary file os.remove(temp_filepath) path_delim = "/" db = firestore.client() test_collection = db.collection("testData") data = [] test_docs = test_collection.stream() for test_doc in test_docs: test_data = test_doc.to_dict() data.append(test_data) df = pd.DataFrame(data) df = df[(df['testTime'] > start_date) & (df['testTime'] <= end_date)] df = df.sort_values(by=['testTime'], ascending=False) # df = df.drop(['resultData', "reportUploadTime", "userImageURL", "testType", "birthYear", "testStatus", "reportPath", "createdBy", "csvPath", "result", "mobileId", "resultRatio", "localFlag", "led2", "led1"], axis=1) df = df[["_id", "classificationResult", "prdClassification", "predictedDenovixRatio", "calculatedRatio", "deviceRatio", "kitSerial", "abs1", "led1Average", "led1Buffer", "led1Sample", "abs2", "led2Average", "led2Buffer", "led2Sample", "abs3", "led3Average", "led3Buffer", "led3Sample", "abs4", "led4Average", "led4Buffer", "led4Sample", "deviceId", "deviceSerialNumber", "name", "testTime"]] df.rename(columns={'deviceSerialNumber': "loginId"}, inplace = True) # df = df.reindex(sorted(df.columns), axis=1) df = df.drop_duplicates() curdir = os.getcwd() path_delim = '/' output_filename = '' precision_threshold = 0.009 accuracy_threshold = 0.02 denovix_reference_values = { "Tartrazine": { "10": 0.187191676, "30": 0.187191676, "50": 0.187447015, "60": 0.187191676, "75": 0.294739334, "90": 0.187191676, "100": 0.410113264, "120": 0.187191676, "125": 0.512526022, "150": 0.622883833, "175": 0.721095085, "180": 0.187191676, "200": 0.824065454, "210": 0.187191676, "225": 0.931385567, "250": 1.002224884, "275": 1.091996882, "300": 1.16558307, "350": 1.16558307, }, "KMnO4": { "10": 0.187191676, "100": 0.187191676, "145": 0.187191676, "175": 0.187191676, "250": 0.077846397, "290": 0.359288533, "350": 0.111884606, "435": 0.187191676, "450": 0.138031952, "550": 0.175111257, "580": 0.628039375, "650": 0.208337398, "720": 0.187191676, "750": 0.238773222, "850": 0.267574903, "950": 0.299568449, "1050": 0.329271864, "1150": 0.365827844, "1160": 0.388015579, "1250": 0.388015579, "1350": 0.417304075, "1450": 0.447634285, "1550": 0.478220085, }, "HB": { "2": 0.187191676, "4": 0.187191676, "5": 0.187191676, "6": 0.187191676, "8": 0.077846397, }, } df_reference_device = pd.DataFrame(denovix_reference_values).T df_reference_device.index.name = 'Solution' df_reference_device.columns.name = 'Wavelength' if not df.empty: conditions = [df['name'].str.contains('Tar', case=False, na=False), df['name'].str.contains('KM', case=False, na=False), df['name'].str.contains('HB', case=False, na=False)] choices = ['Tartrazine', 'KMnO4', "HB"] df['solution'] = np.select(conditions, choices, default=None) def extract_numbers(s): if "HB" in s: match = re.search(r'-(\d+)$', s) else: match = re.search(r'-(\d+)', s) or re.search(r'(\d+)', s) if match: return int(match.group(1)) else: return None df["concentration"] = df['name'].apply(extract_numbers) df['absorbance'] = df.apply(lambda row: row['led1Average'] if row['solution'] == 'Tartrazine' else row['led3Average'], axis=1) df_precision_acc = df[["deviceId", "solution", "concentration", "led1Average", "led3Average", "absorbance"]].groupby(["deviceId", "solution", "concentration"]).describe()["absorbance"][["count", "min", "max", "mean"]] df_precision_acc['precision'] = df_precision_acc['max'] - df_precision_acc['min'] df_precision_acc['precision_result'] = ['Fail' if diff > precision_threshold else 'Pass' for diff in df_precision_acc['precision']] df_precision_acc.reset_index(inplace=True) df_precision_acc.set_index(["deviceId", "solution", "concentration"], inplace=True) device_precision_results = {} for index, group_df in df_precision_acc.groupby(level=[0, 1, 2]): soln = index[1] concen = index[2] result_values = group_df['precision_result'].values if index[0] not in device_precision_results: device_precision_results[index[0]] = "Pass" else: result = 'Fail' if 'Fail' in result_values else 'Pass' if result == 'Fail': device_precision_results[index[0]] = 'Fail' df_device_precicion_results = pd.DataFrame(list(device_precision_results.items()), columns=['Device', 'Result']) df_precision_acc = df_precision_acc.assign(accuracy='', accuracy_result='') device_accuracy_results = {} for idx, row in df_precision_acc.iterrows(): device_id = row.name[0] solution = row.name[1] concentration = str(int(row.name[2])) try: reference_value = denovix_reference_values[solution][concentration] accuracy = math.fabs(row['mean'] - reference_value) df_precision_acc.at[idx, 'accuracy'] = accuracy accuracy_result = 'Pass' if accuracy < accuracy_threshold else 'Fail' df_precision_acc.at[idx, 'accuracy_result'] = accuracy_result if device_id not in device_accuracy_results: device_accuracy_results[device_id] = accuracy_result else: if accuracy_result == 'Fail': device_accuracy_results[device_id] = 'Fail' except KeyError: df_precision_acc.at[idx, 'accuracy'] = np.nan df_precision_acc.at[idx, 'accuracy_result'] = 'Fail' # Assume 'Fail' for missing data df_device_accuracy_results = pd.DataFrame(list(device_accuracy_results.items()), columns=['Device', 'Result']) device_results = {} df_device_results = pd.merge(df_device_precicion_results, df_device_accuracy_results, on='Device', how='outer', suffixes=('_precision', '_accuracy')) df_device_results['Result'] = np.where((df_device_results['Result_precision'] == 'Fail') | (df_device_results['Result_accuracy'] == 'Fail'), 'Fail', 'Pass') output_filename = f'/tmp/accuracy_{datetime.today().strftime("%d_%m_%Y_%H_%M")}.xlsx' output_path = os.getcwd() + path_delim + output_filename print('output_path', output_path) writer = pd.ExcelWriter(output_filename, engine = 'xlsxwriter') df_device_results.to_excel(writer, sheet_name="device_results") df_precision_acc.to_excel(writer, sheet_name="concentration") df.to_excel(writer, sheet_name="in") df_reference_device.to_excel(writer, sheet_name="reference_device") workbook = writer.book worksheet_device_results = writer.sheets['device_results'] red_format = workbook.add_format({'bg_color': '#FFC7CE', 'font_color': '#9C0006'}) green_format = workbook.add_format({'bg_color': '#C6EFCE', 'font_color': '#006100'}) worksheet_device_results.conditional_format('E2:E{}'.format(df_device_results.shape[0] + 1), {'type': 'text', 'criteria': 'containing', 'value': 'Fail', 'format': red_format}) worksheet_device_results.conditional_format('E2:E{}'.format(df_device_results.shape[0] + 1), {'type': 'text', 'criteria': 'containing', 'value': 'Pass', 'format': green_format}) wks1 = workbook.add_worksheet('abs_plot') wks1.write(0,0,'Abs Plot') # Write the header wks1.write(0, 0, 'Solution') wks1.write(0, 1, 'Slope') wks1.write(0, 2, 'Intercept') wks1.write(0, 3, 'R^2') row = 1 fig, ax = plt.subplots() filtered_df = df[df['solution'].isin(['KMnO4', 'Tartrazine', 'HB'])] for solution in filtered_df['solution'].unique(): if pd.notna(solution): solution_data = df[df['solution'] == solution] concentration_means = solution_data.groupby('concentration')['absorbance'].mean() # Fit a linear regression model coefficients = np.polyfit(concentration_means.index, concentration_means.values, 1) # Use linregress to get additional statistics including R-squared slope, intercept, r_value, p_value, std_err = linregress(concentration_means.index, concentration_means.values) print(f'For {solution}: Slope={slope:.4f}, Intercept={intercept:.4f}, R^2={r_value**2:.4f}') ax.plot(concentration_means.index, concentration_means.values, marker='o', linestyle='-', label=f'Mean Absorbance for {solution}') ax.plot(concentration_means.index, np.polyval(coefficients, concentration_means.index), linestyle='--', label=f'Linear Fit for {solution}') annotation_text = f'Slope: {slope:.4f}\nIntercept: {intercept:.4f}\nR^2: {r_value**2:.4f}' ax.text(concentration_means.index[-1] + 5, np.polyval(coefficients, concentration_means.index[-1]), annotation_text, fontsize=10, verticalalignment='center') wks1.write(row, 0, solution) wks1.write(row, 1, str(slope)) wks1.write(row, 2, str(intercept)) wks1.write(row, 3, str(r_value**2)) row += 1 ax.set_title('Mean Absorbance for Each Solution') ax.set_xlabel('Concentration') ax.set_ylabel('Mean Absorbance') ax.legend() # plt.show() imgdata=io.BytesIO() fig.savefig(imgdata, format='png') wks1.insert_image(6,0, '', {'image_data': imgdata}) ### debug each row # # Merge df and df_device_precicion_results on deviceId # df_merged = pd.merge(df, df_device_precicion_results, left_on='deviceId', right_on='Device', how='left') # # Drop the duplicate "Device" column # df_merged.drop(columns=['Device'], inplace=True) # # Save the merged DataFrame to the Excel file # df_merged.to_excel(writer, sheet_name="merged_results") writer.close() else: output_filename = f'/tmp/empty_excel_{datetime.today().strftime("%d_%m_%Y_%H_%M")}.xlsx' pd.DataFrame().to_excel(output_filename, engine='xlsxwriter', index=False) # firebase_admin.delete_app(firebase_admin.get_app()) with open(output_filename,'rb') as f: file_data = io.BytesIO(f.read()) # application/vnd.ms-excel response = make_response(send_file(file_data, download_name= output_filename, mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"), 200, headers) return response