Spaces:
Runtime error
Runtime error
| """Streamlit UI block for Tab 2: Phyphox live sensor upload. | |
| Call from streamlit_app.py: | |
| from phyphox_app_block import render_phyphox_tab | |
| with tab2: | |
| render_phyphox_tab(ffn_model, ffn_status, cnn_model, cnn_status) | |
| """ | |
| import json | |
| import os | |
| import numpy as np | |
| import pandas as pd | |
| import streamlit as st | |
| from phyphox_pipeline import process_phyphox_files, FS, WINDOW, STEP | |
| # ── Constants ───────────────────────────────────────────────────────────────── | |
| LABEL_MAP = { | |
| 0: "WALKING", | |
| 1: "WALKING_UPSTAIRS", | |
| 2: "WALKING_DOWNSTAIRS", | |
| 3: "SITTING", | |
| 4: "STANDING", | |
| 5: "LAYING", | |
| } | |
| EXPLANATIONS = { | |
| "LAYING": "Minimal movement detected across all axes: consistent with a stationary horizontal posture.", | |
| "SITTING": "Low dynamic acceleration with stable gravity: stationary upright posture.", | |
| "STANDING": "Similar to sitting with slight postural micro-movements.", | |
| "WALKING": "Rhythmic periodic acceleration on the vertical axis: level walking at normal cadence.", | |
| "WALKING_DOWNSTAIRS": "Downward gravitational shift with higher impact peaks: descending stairs.", | |
| "WALKING_UPSTAIRS": "Elevated vertical acceleration effort: climbing stairs.", | |
| } | |
| # ── Normalisation ───────────────────────────────────────────────────────────── | |
| def _load_norm_params(norm_path: str): | |
| """Load per-feature min/max from norm_params.json (keys are string indices 0-560).""" | |
| with open(norm_path) as f: | |
| d = json.load(f) | |
| min_vals = np.array([d[str(i)]["min"] for i in range(561)], dtype=np.float32) | |
| max_vals = np.array([d[str(i)]["max"] for i in range(561)], dtype=np.float32) | |
| return min_vals, max_vals | |
| def _normalize(features: np.ndarray, min_vals: np.ndarray, max_vals: np.ndarray) -> np.ndarray: | |
| """Best-effort feature-level min-max scaling to [-1, 1]. | |
| Uses per-feature min/max observed in the UCI HAR training set. This is | |
| an approximation: the UCI pipeline normalises raw signals before feature | |
| extraction, so physical-unit features may fall outside the training range. | |
| Values are clipped before scaling to keep outputs bounded. | |
| """ | |
| rng = max_vals - min_vals | |
| rng = np.where(rng < 1e-8, 1.0, rng) # avoid div-by-zero | |
| clipped = np.clip(features, min_vals, max_vals) | |
| return 2.0 * (clipped - min_vals) / rng - 1.0 | |
| # ── Public render function ──────────────────────────────────────────────────── | |
| def render_phyphox_tab( | |
| ffn_model, ffn_status: str, | |
| cnn_model, cnn_status: str, | |
| norm_params_path: str, | |
| ) -> None: | |
| st.subheader("Upload Phyphox sensor recording") | |
| st.markdown(""" | |
| **How to record your own data:** | |
| 1. Install [Phyphox](https://phyphox.org/) on your phone | |
| 2. Open **Acceleration (with g)** and **Gyroscope**: record simultaneously | |
| 3. Hold the phone at your waist (same position as the UCI dataset) | |
| 4. Record at least 3 seconds of a single activity | |
| 5. Export both experiments as CSV and upload below | |
| """) | |
| if "phyphox_run" not in st.session_state: | |
| st.session_state.phyphox_run = 0 | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| acc_file = st.file_uploader( | |
| "Accelerometer CSV", | |
| type=["csv"], | |
| key=f"acc_upload_{st.session_state.phyphox_run}", | |
| help="Columns: Time (s), X (m/s²), Y (m/s²), Z (m/s²)", | |
| ) | |
| with col2: | |
| gyro_file = st.file_uploader( | |
| "Gyroscope CSV", | |
| type=["csv"], | |
| key=f"gyro_upload_{st.session_state.phyphox_run}", | |
| help="Columns: Time (s), X (rad/s), Y (rad/s), Z (rad/s)", | |
| ) | |
| if acc_file is not None or gyro_file is not None: | |
| if st.button("Clear / new recording"): | |
| st.session_state.phyphox_run += 1 | |
| st.rerun() | |
| if acc_file is None or gyro_file is None: | |
| st.info("Upload both files to continue.") | |
| return | |
| # ── Feature extraction ──────────────────────────────────────────────────── | |
| try: | |
| with st.spinner("Extracting 561 features from sensor data…"): | |
| features, pipeline_warnings = process_phyphox_files(acc_file, gyro_file) | |
| except ValueError as err: | |
| st.error(str(err)) | |
| return | |
| except Exception as err: | |
| st.error(f"Unexpected error during feature extraction: {err}") | |
| return | |
| for w in pipeline_warnings: | |
| st.warning(w) | |
| n_windows = len(features) | |
| duration_s = (n_windows - 1) * (STEP / FS) + (WINDOW / FS) | |
| c1, c2, c3 = st.columns(3) | |
| c1.metric("Windows extracted", n_windows) | |
| c2.metric("Approx. duration", f"{duration_s:.1f} s") | |
| c3.metric("Features per window", 561) | |
| st.caption( | |
| f"Each window = {WINDOW / FS:.2f} s at {FS} Hz · " | |
| f"50% overlap ({STEP / FS:.2f} s hop)" | |
| ) | |
| # ── Normalisation ───────────────────────────────────────────────────────── | |
| if os.path.exists(norm_params_path): | |
| min_vals, max_vals = _load_norm_params(norm_params_path) | |
| features = _normalize(features, min_vals, max_vals) | |
| st.caption( | |
| "Accelerometer converted from m/s² to g units to match UCI training data. " | |
| "Features scaled to [−1, 1] using physical-unit min/max computed from the " | |
| "UCI HAR training set raw inertial signals." | |
| ) | |
| else: | |
| st.warning( | |
| "norm_params.json not found: features are in physical units. " | |
| "Predictions will be unreliable until normalisation is applied." | |
| ) | |
| # ── Predictions ─────────────────────────────────────────────────────────── | |
| if ffn_status != "ready" and cnn_status != "ready": | |
| st.warning("Models not loaded: cannot predict yet.") | |
| return | |
| st.markdown("---") | |
| st.subheader("Model comparison") | |
| left, right = st.columns(2) | |
| def _render_model_col(col, model, status, name): | |
| with col: | |
| st.markdown(f"#### {name}") | |
| if status != "ready": | |
| st.error(f"Model not loaded: {status}") | |
| return | |
| probs_all = model.predict(features, verbose=0) # (n_windows, 6) | |
| pred_labels = [LABEL_MAP[int(np.argmax(p))] for p in probs_all] | |
| from collections import Counter | |
| vote = Counter(pred_labels).most_common(1)[0][0] | |
| avg_conf = float(np.mean(np.max(probs_all, axis=1))) * 100 | |
| st.success(f"**{vote}** · {avg_conf:.1f}% avg confidence") | |
| st.markdown(f"_{EXPLANATIONS[vote]}_") | |
| if n_windows > 1: | |
| with st.expander(f"Per-window breakdown ({n_windows} windows)"): | |
| rows = [] | |
| for i, (p, label) in enumerate(zip(probs_all, pred_labels)): | |
| t_start = i * STEP / FS | |
| rows.append({ | |
| "Window": i + 1, | |
| "Time (s)": f"{t_start:.1f}–{t_start + WINDOW/FS:.1f}", | |
| "Prediction": label, | |
| "Confidence": f"{float(np.max(p))*100:.1f}%", | |
| }) | |
| st.dataframe(pd.DataFrame(rows), use_container_width=True) | |
| mean_probs = probs_all.mean(axis=0) | |
| st.markdown("**Average confidence across all classes**") | |
| st.bar_chart(pd.DataFrame( | |
| {"Confidence (%)": [float(mean_probs[i]) * 100 for i in range(6)]}, | |
| index=[LABEL_MAP[i] for i in range(6)], | |
| )) | |
| _render_model_col(left, ffn_model, ffn_status, "Feedforward Network") | |
| _render_model_col(right, cnn_model, cnn_status, "1D Convolutional Network") | |