feat: FASE 5b/6.1+6.2 - SPC Backend + Dashboard Metrologist (Plotly.js)
Aggiunge servizio SPC con calcoli Cp/Cpk/Pp/Ppk, carta di controllo (UCL/LCL), istogramma con curva normale. Router FastAPI con 5 endpoint statistics, blueprint Flask con proxy AJAX, dashboard interattiva Alpine.js + Plotly.js con filtri per ricetta/subtask/date, riepilogo pass/fail, gauge Cpk e i18n IT/EN completo. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Adriano
@@ -15,6 +15,7 @@ from routers.tasks import router as tasks_router
|
||||
from routers.measurements import router as measurements_router
|
||||
from routers.files import router as files_router
|
||||
from routers.settings import router as settings_router
|
||||
from routers.statistics import router as statistics_router
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
@@ -58,6 +59,7 @@ app.include_router(tasks_router)
|
||||
app.include_router(measurements_router)
|
||||
app.include_router(files_router)
|
||||
app.include_router(settings_router)
|
||||
app.include_router(statistics_router)
|
||||
|
||||
|
||||
@app.get("/api/health")
|
||||
|
||||
@@ -0,0 +1,236 @@
|
||||
"""Statistics router - SPC endpoints for Metrologist dashboard."""
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from sqlalchemy import and_, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from database import get_db
|
||||
from middleware.api_key import require_metrologist
|
||||
from models.measurement import Measurement
|
||||
from models.recipe import RecipeVersion
|
||||
from models.task import RecipeSubtask, RecipeTask
|
||||
from models.user import User
|
||||
from schemas.statistics import (
|
||||
CapabilityData,
|
||||
ControlChartData,
|
||||
HistogramData,
|
||||
SummaryData,
|
||||
)
|
||||
from services.spc_service import (
|
||||
compute_capability,
|
||||
compute_control_chart,
|
||||
compute_histogram,
|
||||
compute_summary,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api/statistics", tags=["statistics"])
|
||||
|
||||
|
||||
async def _query_measurements(
|
||||
db: AsyncSession,
|
||||
recipe_id: int,
|
||||
version_id: int | None = None,
|
||||
subtask_id: int | None = None,
|
||||
date_from: datetime | None = None,
|
||||
date_to: datetime | None = None,
|
||||
operator_id: int | None = None,
|
||||
lot_number: str | None = None,
|
||||
serial_number: str | None = None,
|
||||
) -> list[Measurement]:
|
||||
"""Query measurements with common filters.
|
||||
|
||||
Returns list of Measurement ORM objects ordered by measured_at.
|
||||
"""
|
||||
filters = []
|
||||
|
||||
# recipe_id filter via RecipeVersion subquery
|
||||
if version_id is not None:
|
||||
filters.append(Measurement.version_id == version_id)
|
||||
else:
|
||||
version_ids = select(RecipeVersion.id).where(
|
||||
RecipeVersion.recipe_id == recipe_id
|
||||
)
|
||||
filters.append(Measurement.version_id.in_(version_ids))
|
||||
|
||||
if subtask_id is not None:
|
||||
filters.append(Measurement.subtask_id == subtask_id)
|
||||
if date_from is not None:
|
||||
filters.append(Measurement.measured_at >= date_from)
|
||||
if date_to is not None:
|
||||
filters.append(Measurement.measured_at <= date_to)
|
||||
if operator_id is not None:
|
||||
filters.append(Measurement.measured_by == operator_id)
|
||||
if lot_number is not None:
|
||||
filters.append(Measurement.lot_number == lot_number)
|
||||
if serial_number is not None:
|
||||
filters.append(Measurement.serial_number == serial_number)
|
||||
|
||||
query = (
|
||||
select(Measurement)
|
||||
.where(and_(*filters) if filters else True)
|
||||
.order_by(Measurement.measured_at.asc())
|
||||
)
|
||||
result = await db.execute(query)
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
async def _get_subtask_tolerances(
|
||||
db: AsyncSession,
|
||||
subtask_id: int,
|
||||
) -> dict:
|
||||
"""Get tolerance limits for a specific subtask."""
|
||||
result = await db.execute(
|
||||
select(RecipeSubtask).where(RecipeSubtask.id == subtask_id)
|
||||
)
|
||||
subtask = result.scalar_one_or_none()
|
||||
if subtask is None:
|
||||
return {"utl": None, "uwl": None, "lwl": None, "ltl": None, "nominal": None}
|
||||
return {
|
||||
"utl": float(subtask.utl) if subtask.utl is not None else None,
|
||||
"uwl": float(subtask.uwl) if subtask.uwl is not None else None,
|
||||
"lwl": float(subtask.lwl) if subtask.lwl is not None else None,
|
||||
"ltl": float(subtask.ltl) if subtask.ltl is not None else None,
|
||||
"nominal": float(subtask.nominal) if subtask.nominal is not None else None,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/summary", response_model=SummaryData)
|
||||
async def get_summary(
|
||||
recipe_id: int = Query(...),
|
||||
version_id: int | None = Query(None),
|
||||
subtask_id: int | None = Query(None),
|
||||
date_from: datetime | None = Query(None),
|
||||
date_to: datetime | None = Query(None),
|
||||
operator_id: int | None = Query(None),
|
||||
lot_number: str | None = Query(None),
|
||||
serial_number: str | None = Query(None),
|
||||
user: User = Depends(require_metrologist),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
) -> SummaryData:
|
||||
"""Get pass/fail/warning summary for filtered measurements."""
|
||||
measurements = await _query_measurements(
|
||||
db, recipe_id, version_id, subtask_id,
|
||||
date_from, date_to, operator_id, lot_number, serial_number,
|
||||
)
|
||||
pass_fail_values = [m.pass_fail for m in measurements]
|
||||
return compute_summary(pass_fail_values)
|
||||
|
||||
|
||||
@router.get("/capability", response_model=CapabilityData)
|
||||
async def get_capability(
|
||||
recipe_id: int = Query(...),
|
||||
subtask_id: int = Query(...),
|
||||
version_id: int | None = Query(None),
|
||||
date_from: datetime | None = Query(None),
|
||||
date_to: datetime | None = Query(None),
|
||||
operator_id: int | None = Query(None),
|
||||
lot_number: str | None = Query(None),
|
||||
serial_number: str | None = Query(None),
|
||||
user: User = Depends(require_metrologist),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
) -> CapabilityData:
|
||||
"""Get capability indices (Cp/Cpk/Pp/Ppk) for a specific subtask."""
|
||||
measurements = await _query_measurements(
|
||||
db, recipe_id, version_id, subtask_id,
|
||||
date_from, date_to, operator_id, lot_number, serial_number,
|
||||
)
|
||||
values = [float(m.value) for m in measurements]
|
||||
tol = await _get_subtask_tolerances(db, subtask_id)
|
||||
return compute_capability(values, tol["utl"], tol["ltl"], tol["nominal"])
|
||||
|
||||
|
||||
@router.get("/control-chart", response_model=ControlChartData)
|
||||
async def get_control_chart(
|
||||
recipe_id: int = Query(...),
|
||||
subtask_id: int = Query(...),
|
||||
version_id: int | None = Query(None),
|
||||
date_from: datetime | None = Query(None),
|
||||
date_to: datetime | None = Query(None),
|
||||
operator_id: int | None = Query(None),
|
||||
lot_number: str | None = Query(None),
|
||||
serial_number: str | None = Query(None),
|
||||
user: User = Depends(require_metrologist),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
) -> ControlChartData:
|
||||
"""Get control chart data with UCL/LCL and OOC detection."""
|
||||
measurements = await _query_measurements(
|
||||
db, recipe_id, version_id, subtask_id,
|
||||
date_from, date_to, operator_id, lot_number, serial_number,
|
||||
)
|
||||
values = [float(m.value) for m in measurements]
|
||||
timestamps = [m.measured_at for m in measurements]
|
||||
tol = await _get_subtask_tolerances(db, subtask_id)
|
||||
return compute_control_chart(
|
||||
values, timestamps,
|
||||
tol["utl"], tol["uwl"], tol["lwl"], tol["ltl"], tol["nominal"],
|
||||
)
|
||||
|
||||
|
||||
@router.get("/histogram", response_model=HistogramData)
|
||||
async def get_histogram(
|
||||
recipe_id: int = Query(...),
|
||||
subtask_id: int = Query(...),
|
||||
version_id: int | None = Query(None),
|
||||
date_from: datetime | None = Query(None),
|
||||
date_to: datetime | None = Query(None),
|
||||
operator_id: int | None = Query(None),
|
||||
lot_number: str | None = Query(None),
|
||||
serial_number: str | None = Query(None),
|
||||
n_bins: int = Query(20, ge=5, le=100),
|
||||
user: User = Depends(require_metrologist),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
) -> HistogramData:
|
||||
"""Get histogram data with normal curve overlay."""
|
||||
measurements = await _query_measurements(
|
||||
db, recipe_id, version_id, subtask_id,
|
||||
date_from, date_to, operator_id, lot_number, serial_number,
|
||||
)
|
||||
values = [float(m.value) for m in measurements]
|
||||
return compute_histogram(values, n_bins)
|
||||
|
||||
|
||||
@router.get("/subtasks")
|
||||
async def get_recipe_subtasks(
|
||||
recipe_id: int = Query(...),
|
||||
version_id: int | None = Query(None),
|
||||
user: User = Depends(require_metrologist),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
) -> list[dict]:
|
||||
"""Get subtasks for a recipe (for filter dropdown).
|
||||
|
||||
Returns subtask id, marker_number, description, and parent task title.
|
||||
"""
|
||||
# Get version_id: use provided or find current version
|
||||
if version_id is None:
|
||||
ver_result = await db.execute(
|
||||
select(RecipeVersion.id).where(
|
||||
RecipeVersion.recipe_id == recipe_id,
|
||||
RecipeVersion.is_current == True,
|
||||
)
|
||||
)
|
||||
version_id = ver_result.scalar_one_or_none()
|
||||
if version_id is None:
|
||||
return []
|
||||
|
||||
# Get tasks and subtasks for this version
|
||||
tasks_result = await db.execute(
|
||||
select(RecipeTask).where(RecipeTask.version_id == version_id)
|
||||
.order_by(RecipeTask.order_index)
|
||||
)
|
||||
tasks = tasks_result.scalars().all()
|
||||
|
||||
subtasks_list = []
|
||||
for task in tasks:
|
||||
for st in sorted(task.subtasks, key=lambda s: s.marker_number):
|
||||
subtasks_list.append({
|
||||
"id": st.id,
|
||||
"marker_number": st.marker_number,
|
||||
"description": st.description,
|
||||
"task_title": task.title,
|
||||
"nominal": float(st.nominal) if st.nominal is not None else None,
|
||||
"utl": float(st.utl) if st.utl is not None else None,
|
||||
"ltl": float(st.ltl) if st.ltl is not None else None,
|
||||
})
|
||||
|
||||
return subtasks_list
|
||||
@@ -0,0 +1,261 @@
|
||||
"""SPC (Statistical Process Control) computation service.
|
||||
|
||||
Pure functions — no DB dependencies. Receives lists of floats and tolerance limits,
|
||||
returns structured data matching schemas in schemas/statistics.py.
|
||||
|
||||
Uses only Python stdlib (math, statistics). No numpy/scipy needed.
|
||||
"""
|
||||
import math
|
||||
import statistics as stats
|
||||
from datetime import datetime
|
||||
|
||||
from schemas.statistics import (
|
||||
CapabilityData,
|
||||
ControlChartData,
|
||||
HistogramData,
|
||||
SummaryData,
|
||||
)
|
||||
|
||||
|
||||
def compute_summary(
|
||||
pass_fail_values: list[str],
|
||||
) -> SummaryData:
|
||||
"""Compute pass/fail/warning summary from a list of pass_fail strings.
|
||||
|
||||
Args:
|
||||
pass_fail_values: List of "pass", "warning", or "fail" strings.
|
||||
|
||||
Returns:
|
||||
SummaryData with counts and rates.
|
||||
"""
|
||||
total = len(pass_fail_values)
|
||||
if total == 0:
|
||||
return SummaryData(
|
||||
total=0,
|
||||
pass_count=0,
|
||||
warning_count=0,
|
||||
fail_count=0,
|
||||
pass_rate=0.0,
|
||||
warning_rate=0.0,
|
||||
fail_rate=0.0,
|
||||
)
|
||||
|
||||
pass_count = pass_fail_values.count("pass")
|
||||
warning_count = pass_fail_values.count("warning")
|
||||
fail_count = pass_fail_values.count("fail")
|
||||
|
||||
return SummaryData(
|
||||
total=total,
|
||||
pass_count=pass_count,
|
||||
warning_count=warning_count,
|
||||
fail_count=fail_count,
|
||||
pass_rate=round(pass_count / total * 100, 2),
|
||||
warning_rate=round(warning_count / total * 100, 2),
|
||||
fail_rate=round(fail_count / total * 100, 2),
|
||||
)
|
||||
|
||||
|
||||
def compute_capability(
|
||||
values: list[float],
|
||||
utl: float | None,
|
||||
ltl: float | None,
|
||||
nominal: float | None,
|
||||
) -> CapabilityData:
|
||||
"""Compute capability indices Cp, Cpk, Pp, Ppk.
|
||||
|
||||
- Cp = (UTL - LTL) / (6 * sigma_within)
|
||||
- Cpk = min((UTL - mean) / (3 * sigma), (mean - LTL) / (3 * sigma))
|
||||
- Pp/Ppk: same formulas using population std dev (same data, no subgrouping)
|
||||
|
||||
Args:
|
||||
values: Measured values.
|
||||
utl: Upper Tolerance Limit (None if not defined).
|
||||
ltl: Lower Tolerance Limit (None if not defined).
|
||||
nominal: Nominal value (None if not defined).
|
||||
|
||||
Returns:
|
||||
CapabilityData with indices and statistics.
|
||||
"""
|
||||
n = len(values)
|
||||
if n < 2:
|
||||
return CapabilityData(
|
||||
cp=None, cpk=None, pp=None, ppk=None,
|
||||
mean=values[0] if n == 1 else 0.0,
|
||||
std_dev=0.0, n=n,
|
||||
utl=utl, ltl=ltl, nominal=nominal,
|
||||
)
|
||||
|
||||
mean = stats.mean(values)
|
||||
# Population std dev for Pp/Ppk
|
||||
std_dev_pop = stats.pstdev(values)
|
||||
# Sample std dev for Cp/Cpk
|
||||
std_dev_sample = stats.stdev(values)
|
||||
|
||||
cp = cpk = pp = ppk = None
|
||||
|
||||
if utl is not None and ltl is not None and std_dev_sample > 0:
|
||||
cp = round((utl - ltl) / (6 * std_dev_sample), 4)
|
||||
|
||||
if std_dev_sample > 0:
|
||||
cpk_values = []
|
||||
if utl is not None:
|
||||
cpk_values.append((utl - mean) / (3 * std_dev_sample))
|
||||
if ltl is not None:
|
||||
cpk_values.append((mean - ltl) / (3 * std_dev_sample))
|
||||
if cpk_values:
|
||||
cpk = round(min(cpk_values), 4)
|
||||
|
||||
if utl is not None and ltl is not None and std_dev_pop > 0:
|
||||
pp = round((utl - ltl) / (6 * std_dev_pop), 4)
|
||||
|
||||
if std_dev_pop > 0:
|
||||
ppk_values = []
|
||||
if utl is not None:
|
||||
ppk_values.append((utl - mean) / (3 * std_dev_pop))
|
||||
if ltl is not None:
|
||||
ppk_values.append((mean - ltl) / (3 * std_dev_pop))
|
||||
if ppk_values:
|
||||
ppk = round(min(ppk_values), 4)
|
||||
|
||||
return CapabilityData(
|
||||
cp=cp, cpk=cpk, pp=pp, ppk=ppk,
|
||||
mean=round(mean, 6),
|
||||
std_dev=round(std_dev_sample, 6),
|
||||
n=n,
|
||||
utl=utl, ltl=ltl, nominal=nominal,
|
||||
)
|
||||
|
||||
|
||||
def compute_control_chart(
|
||||
values: list[float],
|
||||
timestamps: list[datetime],
|
||||
utl: float | None,
|
||||
uwl: float | None,
|
||||
lwl: float | None,
|
||||
ltl: float | None,
|
||||
nominal: float | None,
|
||||
) -> ControlChartData:
|
||||
"""Compute control chart data with UCL/LCL and out-of-control detection.
|
||||
|
||||
UCL = mean + 3*sigma
|
||||
LCL = mean - 3*sigma
|
||||
Out-of-control: points outside UCL/LCL.
|
||||
|
||||
Args:
|
||||
values: Measured values in chronological order.
|
||||
timestamps: Corresponding timestamps.
|
||||
utl/uwl/lwl/ltl: Tolerance/warning limits.
|
||||
nominal: Nominal value.
|
||||
|
||||
Returns:
|
||||
ControlChartData with values, limits, and OOC indices.
|
||||
"""
|
||||
n = len(values)
|
||||
if n == 0:
|
||||
return ControlChartData(
|
||||
values=[], timestamps=[], mean=0.0, ucl=0.0, lcl=0.0,
|
||||
utl=utl, uwl=uwl, lwl=lwl, ltl=ltl, nominal=nominal,
|
||||
out_of_control=[],
|
||||
)
|
||||
|
||||
mean = stats.mean(values)
|
||||
|
||||
if n >= 2:
|
||||
sigma = stats.stdev(values)
|
||||
else:
|
||||
sigma = 0.0
|
||||
|
||||
ucl = mean + 3 * sigma
|
||||
lcl = mean - 3 * sigma
|
||||
|
||||
# Detect out-of-control points (outside UCL/LCL)
|
||||
out_of_control = []
|
||||
for i, v in enumerate(values):
|
||||
if v > ucl or v < lcl:
|
||||
out_of_control.append(i)
|
||||
|
||||
return ControlChartData(
|
||||
values=values,
|
||||
timestamps=timestamps,
|
||||
mean=round(mean, 6),
|
||||
ucl=round(ucl, 6),
|
||||
lcl=round(lcl, 6),
|
||||
utl=utl,
|
||||
uwl=uwl,
|
||||
lwl=lwl,
|
||||
ltl=ltl,
|
||||
nominal=nominal,
|
||||
out_of_control=out_of_control,
|
||||
)
|
||||
|
||||
|
||||
def compute_histogram(
|
||||
values: list[float],
|
||||
n_bins: int = 20,
|
||||
) -> HistogramData:
|
||||
"""Compute histogram bin data and normal curve overlay.
|
||||
|
||||
Args:
|
||||
values: Measured values.
|
||||
n_bins: Number of histogram bins (default 20).
|
||||
|
||||
Returns:
|
||||
HistogramData with bins, counts, and normal curve points.
|
||||
"""
|
||||
n = len(values)
|
||||
if n == 0:
|
||||
return HistogramData(
|
||||
bins=[], counts=[], normal_x=[], normal_y=[],
|
||||
mean=0.0, std_dev=0.0, n=0,
|
||||
)
|
||||
|
||||
mean = stats.mean(values)
|
||||
std_dev = stats.pstdev(values) if n >= 2 else 0.0
|
||||
|
||||
min_val = min(values)
|
||||
max_val = max(values)
|
||||
|
||||
# Avoid zero-width range
|
||||
if max_val == min_val:
|
||||
max_val = min_val + 1.0
|
||||
|
||||
bin_width = (max_val - min_val) / n_bins
|
||||
bins = [round(min_val + i * bin_width, 6) for i in range(n_bins + 1)]
|
||||
|
||||
# Count values per bin
|
||||
counts = [0] * n_bins
|
||||
for v in values:
|
||||
idx = int((v - min_val) / bin_width)
|
||||
if idx >= n_bins:
|
||||
idx = n_bins - 1
|
||||
counts[idx] += 1
|
||||
|
||||
# Normal curve overlay (100 points)
|
||||
normal_x: list[float] = []
|
||||
normal_y: list[float] = []
|
||||
if std_dev > 0:
|
||||
n_curve_points = 100
|
||||
x_min = mean - 4 * std_dev
|
||||
x_max = mean + 4 * std_dev
|
||||
x_step = (x_max - x_min) / (n_curve_points - 1)
|
||||
|
||||
for i in range(n_curve_points):
|
||||
x = x_min + i * x_step
|
||||
# Normal PDF: (1 / (sigma * sqrt(2*pi))) * exp(-0.5 * ((x-mu)/sigma)^2)
|
||||
y = (1.0 / (std_dev * math.sqrt(2 * math.pi))) * math.exp(
|
||||
-0.5 * ((x - mean) / std_dev) ** 2
|
||||
)
|
||||
# Scale to match histogram: y * n * bin_width
|
||||
y_scaled = y * n * bin_width
|
||||
normal_x.append(round(x, 6))
|
||||
normal_y.append(round(y_scaled, 4))
|
||||
|
||||
return HistogramData(
|
||||
bins=bins,
|
||||
counts=counts,
|
||||
normal_x=normal_x,
|
||||
normal_y=normal_y,
|
||||
mean=round(mean, 6),
|
||||
std_dev=round(std_dev, 6),
|
||||
n=n,
|
||||
)
|
||||
Reference in New Issue
Block a user