Blame for src/phyphox_app_block.py

"""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 ─────────────────────────────────────────────────────────────

@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


# ── 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")