"""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 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.", } @st.cache_resource 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 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") with st.expander("How to record and export your data - step by step", expanded=True): st.markdown(""" **Step 1 - Install Phyphox** Download the free [Phyphox](https://phyphox.org/) app from the Play Store (Android) or App Store (iOS). **Step 2 - Create a new experiment** - Open the app and tap the **+** icon in the top-right corner - Select **Add simple experiment** - In the active sensors list, add **Accelerometer** and **Gyroscope** - Set the **sensor rate to 50** (Hz) for both - Give the experiment a title (e.g. "Walking") then tap **Proceed** **Step 3 - Configure a timed run** - Tap the **three-dot menu** in the top-right corner and select **Timed run** - Set a **recording duration** (15-20 seconds recommended) - Optionally set a **start delay** (e.g. 5 seconds) so you have time to get into position before recording begins **Step 4 - Position the device** Place your phone in your **trouser pocket** or attach it at your **waist**. This mirrors how the original UCI dataset was collected and gives the most accurate results. **Step 5 - Record one activity** - Press the **play button** to start - Perform a **single activity continuously** - do not stop and restart mid-recording - Stay in steady motion for the full duration before stopping **Step 6 - Export the data** - After the recording ends, tap the **three-dot menu** and select **Export data** - Choose **CSV (comma separated values)** and confirm - You will receive a **.zip file** - unzip it to find separate CSV files for the **Accelerometer** and **Gyroscope** **Step 7 - Upload below** Upload each CSV file into its corresponding field below, then the pipeline will extract features and classify your activity. > ⚠️ **Important:** Each export is a fresh experiment. Do not reuse an old experiment - it will contain data from previous recordings stitched together, which will confuse the classifier. """) 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 try: import io as _io _raw = acc_file.read() acc_file.seek(0) _df = pd.read_csv(_io.StringIO(_raw.decode("utf-8") if isinstance(_raw, bytes) else _raw)) _df.columns = [c.strip('"').strip() for c in _df.columns] _num_cols = [c for c in _df.columns if c != "Time (s)"] _mag = float((_df[_num_cols].apply(pd.to_numeric, errors="coerce") ** 2).sum(axis=1).mean() ** 0.5) if _mag < 3.0: st.error( f"Mean accelerometer magnitude is only {_mag:.2f} m/s² — this looks like " "'Acceleration **without** g'. Please re-record using **'Acceleration (with g)'** " "so gravity is included in the signal." ) return else: st.caption(f"Signal check: mean magnitude = {_mag:.2f} m/s² (gravity present)") # Detect Phyphox pause/resume sessions: large gaps in the time column # mean different activities were stitched together in one CSV. _t = pd.to_numeric(_df["Time (s)"], errors="coerce").dropna().values _gaps = np.diff(_t) _expected_dt = np.median(_gaps) _session_breaks = int(np.sum(_gaps > _expected_dt * 20)) if _session_breaks > 0: st.warning( f"**{_session_breaks} session break(s) detected** in this recording. " "Phyphox accumulates data across pause/resume cycles — your CSV contains " f"{_session_breaks + 1} separate recordings stitched together. " "Only windows from a single activity will predict correctly. " "To fix: tap the **trash icon** in Phyphox to clear data before each new recording." ) except Exception: pass 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)" ) 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." ) 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 # Skip first and last window for the final vote: these are typically # contaminated by recording start/stop transients (person not yet # in full motion, or the gravity filter still warming up). core = probs_all[1:-1] if n_windows > 3 else probs_all core_labels = [LABEL_MAP[int(np.argmax(p))] for p in core] vote = Counter(core_labels).most_common(1)[0][0] avg_conf = float(np.mean(np.max(core, 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 is_edge = (i == 0 or i == n_windows - 1) and n_windows > 3 rows.append({ "Window": i + 1, "Time (s)": f"{t_start:.1f}–{t_start + WINDOW/FS:.1f}", "Prediction": label + (" *" if is_edge else ""), "Confidence": f"{float(np.max(p))*100:.1f}%", }) st.dataframe(pd.DataFrame(rows), use_container_width=True) if n_windows > 3: st.caption("* Edge windows excluded from overall vote (recording start/stop transient).") mean_probs = core.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")