85 lines
4.7 KiB
Python
85 lines
4.7 KiB
Python
"""Раздел «Анализ надёжности» (инженеры и руководители; включается в настройках сервера)."""
|
|
from datetime import datetime, timedelta
|
|
from typing import Literal, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import Response
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import config, reliability as R, schemas as S, scope, settings_store as ss
|
|
from ..db import get_db
|
|
from ..models import Equipment, User
|
|
from ..security import require_engineer
|
|
|
|
router = APIRouter(prefix="/reliability", tags=["reliability"])
|
|
|
|
|
|
def enabled(db: Session) -> bool:
|
|
return ss.get(db, "reliability") != "0"
|
|
|
|
|
|
def guard(db: Session = Depends(get_db), u: User = Depends(require_engineer)) -> User:
|
|
if not enabled(db):
|
|
raise HTTPException(404, "Анализ надёжности выключен в настройках сервера")
|
|
return u
|
|
|
|
|
|
def period(date_from: Optional[datetime], date_to: Optional[datetime]) -> tuple[datetime, datetime]:
|
|
end = S._dt_in(date_to) if date_to else config.now()
|
|
start = S._dt_in(date_from) if date_from else end - timedelta(days=365)
|
|
if start >= end:
|
|
raise HTTPException(422, "Начало периода должно быть раньше конца")
|
|
return start, end
|
|
|
|
|
|
def machines(db: Session, u: User, department_id: Optional[int], cranes: Optional[bool], equipment_id: Optional[int]) -> set[int]:
|
|
"""Машины, которые видит пользователь (отделы), с фильтрами отчёта."""
|
|
stmt = select(Equipment.id)
|
|
cond = scope.equipment_filter(db, u)
|
|
if cond is not None:
|
|
stmt = stmt.where(cond)
|
|
if department_id:
|
|
stmt = stmt.where(Equipment.department_id.in_(scope.dept_tree(db, department_id)))
|
|
if cranes is not None:
|
|
stmt = stmt.where(Equipment.is_crane.is_(cranes))
|
|
if equipment_id:
|
|
stmt = stmt.where(Equipment.id == equipment_id)
|
|
return set(db.scalars(stmt))
|
|
|
|
|
|
@router.get("/dictionaries")
|
|
def dictionaries(_: User = Depends(guard)):
|
|
"""Справочники формы закрытия отказа: причины, характер, что сделано, признаки заранее."""
|
|
return {"cause": R.CAUSE_RU, "nature": R.NATURE_RU, "action": R.ACTION_RU, "precursor": R.PRECURSOR_RU}
|
|
|
|
|
|
@router.get("")
|
|
def summary(date_from: Optional[datetime] = Query(None, alias="from"), date_to: Optional[datetime] = Query(None, alias="to"),
|
|
department_id: Optional[int] = None, cranes: Optional[bool] = None, equipment_id: Optional[int] = None,
|
|
confidence: float = Query(0.9, ge=0.5, le=0.99), horizon_days: int = Query(30, ge=1, le=365),
|
|
db: Session = Depends(get_db), u: User = Depends(guard)):
|
|
"""Показатели надёжности по машинам и парку за период: λ, θ, оценки по малоотказному опыту, Tв, Kг, прогноз, причины."""
|
|
start, end = period(date_from, date_to)
|
|
return R.analyze(db, start, end, machines(db, u, department_id, cranes, equipment_id), confidence, horizon_days)
|
|
|
|
|
|
@router.get("/drift")
|
|
def drift(days: int = Query(180, ge=14, le=1095), horizon: int = Query(90, ge=7, le=730), department_id: Optional[int] = None,
|
|
cranes: Optional[bool] = None, equipment_id: Optional[int] = None, db: Session = Depends(get_db), u: User = Depends(guard)):
|
|
"""«Параметр — граница работоспособности»: дрейф числовых пунктов чек-листов к границе нормы и выходы за неё."""
|
|
return R.drift(db, machines(db, u, department_id, cranes, equipment_id), days, horizon)
|
|
|
|
|
|
@router.get("/export")
|
|
def export(kind: Literal["failures", "service"] = "failures",
|
|
date_from: Optional[datetime] = Query(None, alias="from"), date_to: Optional[datetime] = Query(None, alias="to"),
|
|
department_id: Optional[int] = None, cranes: Optional[bool] = None, equipment_id: Optional[int] = None,
|
|
db: Session = Depends(get_db), u: User = Depends(guard)):
|
|
"""CSV для расчётов: failures — отказы и неисправности (строка на запись); service — история обслуживания узлов."""
|
|
start, end = period(date_from, date_to)
|
|
ids = machines(db, u, department_id, cranes, equipment_id)
|
|
text = (R.export_failures if kind == "failures" else R.export_service_events)(db, start, end, ids)
|
|
name = f"{'failures' if kind == 'failures' else 'service_events'}_{start:%Y%m%d}_{end:%Y%m%d}.csv"
|
|
return Response(text, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{name}"'})
|