import streamlit as st import pandas as pd import math import os import tempfile import xlrd import re from openpyxl import load_workbook from openpyxl.utils.cell import coordinate_to_tuple # ============================================================ # PAGE CONFIG # ============================================================ st.set_page_config( page_title="HTRI Excel Extractor", layout="wide" ) st.title("HTRI Excel Extractor + ML Dataset Builder") st.markdown(""" Upload multiple HTRI Excel files (.xls / .xlsx) The app will: - Extract fixed-cell engineering data - Normalize units - Generate ML-ready dataset - Export combined Excel file """) # ============================================================ # FILE UPLOAD # ============================================================ uploaded_files = st.file_uploader( "Upload HTRI Excel Files", type=["xls", "xlsx"], accept_multiple_files=True ) run = st.button("🚀 Extract Data") # ============================================================ # SAFE FLOAT # ============================================================ def safe_float(v): try: if v is None: return math.nan s = str(v).strip() if s in ["", "-", "None", "N/A"]: return math.nan return float(s.replace(",", "")) except: return math.nan # ============================================================ # NORMALIZE UNIT # ============================================================ def norm(u): if u is None: return "" return ( str(u) .lower() .replace(" ", "") .strip() ) # ============================================================ # FLOW VALUE FIX # ALWAYS PICK VALUE OUTSIDE BRACKET # Example: # 1000 (454) # -> 1000 # ============================================================ def extract_main_number(v): if v is None: return math.nan s = str(v).strip() match = re.match(r"^\s*([-+]?\d*\.?\d+(?:[Ee][-+]?\d+)?)", s) if match: return safe_float(match.group(1)) return safe_float(v) # ============================================================ # UNIT CONVERSIONS # ============================================================ def flow(v, u): if pd.isna(v): return math.nan if norm(u) in ["kg/h", "kg/hr"]: return round(v * 2.20462, 3) return round(v, 3) def heat(v, u): if pd.isna(v): return math.nan u = norm(u) if u == "kw": return round(v * 3412.14, 3) if u == "w": return round(v * 3.41214, 3) if u == "mmbtu/hr": return round(v * 1000000, 3) return round(v, 3) def lmtd(v, u): if pd.isna(v): return math.nan if norm(u) == "c": return round((v * 1.8) + 32, 3) return round(v, 3) def mm_to_in(v, u): if pd.isna(v): return math.nan if norm(u) == "mm": return round(v / 25.4, 3) return round(v, 3) def mm_to_ft(v, u): if pd.isna(v): return math.nan u = norm(u) if u == "mm": return round(v / 304.8, 3) if u == "m": return round(v * 3.28084, 3) return round(v, 3) def pressure_drop(v, u): if pd.isna(v): return math.nan if norm(u) == "kpa": return round(v * 0.145038, 3) return round(v, 3) def velocity(v, u): if pd.isna(v): return math.nan if norm(u) == "m/s": return round(v * 3.28084, 3) return round(v, 3) # ============================================================ # TUBE QUANTITY LOGIC # IF "U" EXISTS -> MULTIPLY BY 2 # ============================================================ def process_tube_qty(v): if v is None: return math.nan s = str(v).strip() nums = re.findall(r"[\d.]+", s) if not nums: return math.nan num = float(nums[0]) if "u" in s.lower(): return num * 2 return num # ============================================================ # SAVE TEMP FILE # ============================================================ def save_temp(uploaded_file): suffix = os.path.splitext(uploaded_file.name)[1] tmp = tempfile.NamedTemporaryFile( delete=False, suffix=suffix ) tmp.write(uploaded_file.read()) tmp.close() return tmp.name # ============================================================ # LOAD TEMA SHEET # ============================================================ def load_sheet(path): ext = path.lower().split(".")[-1] # ======================================================== # XLSX # ======================================================== if ext == "xlsx": wb = load_workbook( path, data_only=True, read_only=True ) for ws in wb.worksheets: if "tema" in ws.title.lower(): return ws, "openpyxl" # ======================================================== # XLS # ======================================================== elif ext == "xls": wb = xlrd.open_workbook(path) for name in wb.sheet_names(): if "tema" in name.lower(): return wb.sheet_by_name(name), "xlrd" return None, None # ============================================================ # UNIVERSAL CELL READER # ============================================================ def get(ws, cell, engine): try: if engine == "openpyxl": return ws[cell].value else: r, c = coordinate_to_tuple(cell) return ws.cell_value(r - 1, c - 1) except: return None # ============================================================ # PROCESS SINGLE FILE # ============================================================ def process(path, name): ws, engine = load_sheet(path) if ws is None: return None, f"{name}: No TEMA sheet found" try: # ==================================================== # FLOWS # ==================================================== flow_u = get(ws, "M14", engine) shell_flow_raw = get(ws, "T14", engine) tube_flow_raw = get(ws, "AR14", engine) shell_flow = flow(extract_main_number(shell_flow_raw), flow_u) tube_flow = flow(extract_main_number(tube_flow_raw), flow_u) # ==================================================== # OUTPUT RECORD # ==================================================== result = { "File_Name": name, # ================================================= # INPUT FEATURES # ================================================= "Shell_Flow_lb_hr": shell_flow, "Tube_Flow_lb_hr": tube_flow, "Heat_Duty_Btu_hr": heat( safe_float(get(ws, "M32", engine)), get(ws, "T32", engine) ), "LMTD_F": lmtd( safe_float(get(ws, "BB32", engine)), get(ws, "BH32", engine) ), "Shell_Passes": safe_float( get(ws, "T38", engine) ), "Tube_Passes": safe_float( get(ws, "AF38", engine) ), "Tube_Pitch_in": mm_to_in( safe_float(get(ws, "BG43", engine)), get(ws, "BL43", engine) ), "Tube_Layout_Angle": safe_float( get(ws, "BM44", engine) ), "Shell_DP_psi": pressure_drop( safe_float(get(ws, "AF30", engine)), get(ws, "M30", engine) ), "Tube_DP_psi": pressure_drop( safe_float(get(ws, "BD30", engine)), get(ws, "M30", engine) ), "Shell_Velocity_ft_s": velocity( safe_float(get(ws, "AB29", engine)), get(ws, "M29", engine) ), "Tube_Velocity_ft_s": velocity( safe_float(get(ws, "AZ29", engine)), get(ws, "M29", engine) ), # ================================================= # OUTPUT FEATURES # ================================================= "Shell_OD_in": mm_to_in( safe_float(get(ws, "AC45", engine)), get(ws, "AH45", engine) ), "Tube_Length_ft": mm_to_ft( safe_float(get(ws, "AR43", engine)), get(ws, "AW43", engine) ), "Tube_OD_in": mm_to_in( safe_float(get(ws, "N43", engine)), get(ws, "R43", engine) ), "Tube_Quantity": process_tube_qty( get(ws, "F43", engine) ), } return result, None except Exception as e: return None, f"{name}: {str(e)}" # ============================================================ # RUN EXTRACTION # ============================================================ if run: if not uploaded_files: st.warning("Please upload files first.") st.stop() results = [] errors = [] progress = st.progress(0) status = st.empty() total = len(uploaded_files) for i, file in enumerate(uploaded_files): status.text(f"Processing {i+1}/{total}: {file.name}") path = save_temp(file) data, err = process(path, file.name) if data: results.append(data) if err: errors.append(err) progress.progress((i + 1) / total) # ======================================================== # DATAFRAME # ======================================================== df = pd.DataFrame(results) st.success("Extraction Completed") st.subheader("Extracted Dataset") st.dataframe(df, use_container_width=True) # ======================================================== # ML FEATURE COLUMNS (for reference / downstream use) # ======================================================== ML_INPUTS = [ "Shell_Flow_lb_hr", "Tube_Flow_lb_hr", "Heat_Duty_Btu_hr", "LMTD_F", "Shell_Passes", "Tube_Passes", "Tube_Pitch_in", "Tube_Layout_Angle", "Shell_DP_psi", "Tube_DP_psi", "Shell_Velocity_ft_s", "Tube_Velocity_ft_s", ] ML_OUTPUTS = [ "Shell_OD_in", "Tube_Length_ft", "Tube_OD_in", "Tube_Quantity", ] available_inputs = [c for c in ML_INPUTS if c in df.columns] available_outputs = [c for c in ML_OUTPUTS if c in df.columns] if available_inputs and available_outputs: st.subheader("ML Feature Summary") col1, col2 = st.columns(2) with col1: st.markdown("**Input Features**") st.dataframe( df[available_inputs].describe().T, use_container_width=True ) with col2: st.markdown("**Output Features**") st.dataframe( df[available_outputs].describe().T, use_container_width=True ) # ======================================================== # DOWNLOAD EXCEL # ======================================================== output_file = "htri_ml_dataset.xlsx" df.to_excel(output_file, index=False) with open(output_file, "rb") as f: st.download_button( label="📥 Download Excel Dataset", data=f, file_name=output_file, mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ) # ======================================================== # ERRORS # ======================================================== if errors: st.subheader("Errors") for e in errors: st.error(e)