"""Регламентное обслуживание: расчёт состояния узлов (по наработке и/или по календарю), сброс, прогноз.""" import calendar from datetime import date, datetime, timedelta from sqlalchemy import select, update from sqlalchemy.orm import Session from . import config from .models import DailyTask, Equipment, IncidentUnit, MeterReading, Order, OrderItem, ServiceEvent, ServiceUnit UNIT_RU = {"day": ("день", "дня", "дней"), "week": ("неделю", "недели", "недель"), "month": ("месяц", "месяца", "месяцев")} STATE_RU = {"ok": "В норме", "soon": "Скоро", "due": "Требуется", "nodata": "Нет данных"} RATE_WINDOW_DAYS = 30 # по какому окну считаем среднюю наработку в сутки для прогноза def plural(n: int, forms: tuple) -> str: n10, n100 = n % 10, n % 100 if n10 == 1 and n100 != 11: return forms[0] if 2 <= n10 <= 4 and not 12 <= n100 <= 14: return forms[1] return forms[2] def num(x: float) -> str: return str(int(x)) if float(x) == int(x) else f"{x:.1f}".rstrip("0").rstrip(".") def add_months(d: date, months: int) -> date: m = d.month - 1 + months y, m = d.year + m // 12, m % 12 + 1 return date(y, m, min(d.day, calendar.monthrange(y, m)[1])) def add_interval(d: date, value: int, unit: str) -> date: if unit == "month": return add_months(d, value) return d + timedelta(days=value * (7 if unit == "week" else 1)) def uses_hours(u: ServiceUnit) -> bool: return u.mode in ("hours", "both") and bool(u.interval_hours) def uses_calendar(u: ServiceUnit) -> bool: return u.mode in ("calendar", "both") and bool(u.interval_value) and u.interval_unit in UNIT_RU def rule_text(u: ServiceUnit) -> str: parts = [] if uses_hours(u): parts.append(f"каждые {num(u.interval_hours)} м/ч") if uses_calendar(u): v = u.interval_value parts.append("раз в " + (UNIT_RU[u.interval_unit][0] if v == 1 else f"{v} {plural(v, UNIT_RU[u.interval_unit])}")) return " или ".join(parts) or "правило не задано" def daily_rate(db: Session, equipment_id: int, now: datetime | None = None) -> float | None: """Средняя наработка в сутки за последние RATE_WINDOW_DAYS по показаниям счётчика (None — данных мало).""" now = now or config.now() rows = db.execute(select(MeterReading.at, MeterReading.hours).where( MeterReading.equipment_id == equipment_id, MeterReading.at >= now - timedelta(days=RATE_WINDOW_DAYS), MeterReading.source != "correction").order_by(MeterReading.at)).all() if len(rows) < 2: return None (t0, h0), (t1, h1) = rows[0], rows[-1] days = (t1 - t0).total_seconds() / 86400 if days < 1 or h1 <= h0: return None return (h1 - h0) / days def evaluate(u: ServiceUnit, meter_hours: float, rate: float | None = None, now: datetime | None = None) -> dict: """Состояние узла: остаток по наработке/календарю, процент выработки ресурса, срок и прогноз.""" now = now or config.now() today = now.date() out = {"state": "ok", "percent": 0, "used_hours": None, "remaining_hours": None, "due_date": None, "remaining_days": None, "forecast_date": None, "rule": rule_text(u)} warn = max(0, min(90, u.warn_percent or 0)) / 100 pcts, soon, due = [], False, False if not (uses_hours(u) or uses_calendar(u)): # правило не задано — считать нечего out["state"] = "nodata" return out if uses_hours(u): used = max(0.0, meter_hours - (u.last_service_hours or 0)) rem = u.interval_hours - used out.update(used_hours=round(used, 1), remaining_hours=round(rem, 1)) pcts.append(used / u.interval_hours * 100) due |= rem <= 0 soon |= rem <= u.interval_hours * warn if uses_calendar(u): base = (u.last_service_at or u.created_at or now).date() due_date = add_interval(base, u.interval_value, u.interval_unit) total = max(1, (due_date - base).days) rem_d = (due_date - today).days out.update(due_date=due_date.isoformat(), remaining_days=rem_d) pcts.append((today - base).days / total * 100) due |= rem_d <= 0 soon |= rem_d <= total * warn out["percent"] = round(max(pcts)) if pcts else 0 out["state"] = "due" if due else "soon" if soon else "ok" # прогноз: ближайшая из дат (календарная и по средней наработке) cands = [] if out["due_date"]: cands.append(date.fromisoformat(out["due_date"])) if out["remaining_hours"] is not None and rate: cands.append(today + timedelta(days=max(0, int(out["remaining_hours"] / rate)))) if cands: out["forecast_date"] = min(cands).isoformat() return out def state_rank(state: str) -> int: return {"due": 0, "soon": 1, "nodata": 2, "ok": 3}[state] # ---------- действия ---------- def record_meter(db: Session, eq: Equipment, hours: float, *, at: datetime | None = None, source: str = "manual", user_id: int | None = None, order_id: int | None = None, note: str = "", correction: bool = False) -> MeterReading: """Новое показание счётчика. Меньше текущего допускается только как коррекция (замена/сброс счётчика).""" at = at or config.now() if hours < (eq.meter_hours or 0) and not correction: raise ValueError("Наработка не может быть меньше текущей (для замены счётчика отметьте «Коррекция»)") r = MeterReading(equipment_id=eq.id, hours=hours, at=at, source="correction" if correction else source, user_id=user_id, order_id=order_id, note=note) db.add(r) if eq.meter_updated_at is None or at >= eq.meter_updated_at or correction: # более раннее показание не «откатывает» счётчик eq.meter_hours, eq.meter_updated_at = hours, at return r def reset_unit(db: Session, u: ServiceUnit, *, at: datetime | None = None, hours: float | None = None, kind: str = "reset", user_id: int | None = None, order_id: int | None = None, note: str = "") -> ServiceEvent: """Сброс узла: обслуживание выполнено (по наряду) или проведено отдельно / при внештатной ситуации.""" eq = db.get(Equipment, u.equipment_id) at = at or config.now() hours = (eq.meter_hours or 0) if hours is None else hours u.last_service_at, u.last_service_hours = at, hours ev = ServiceEvent(unit_id=u.id, at=at, hours=hours, kind=kind, order_id=order_id, user_id=user_id, note=note) db.add(ev) return ev def describe(u: ServiceUnit, ev: dict) -> str: """Короткий текст о состоянии для пункта наряда.""" bits = [u.name, rule_text(u)] if ev["remaining_hours"] is not None: r = ev["remaining_hours"] bits.append(f"осталось {num(r)} м/ч" if r > 0 else f"просрочено на {num(-r)} м/ч") if ev["remaining_days"] is not None: r = ev["remaining_days"] bits.append(f"осталось {r} дн." if r > 0 else f"просрочено на {-r} дн." if r < 0 else "срок сегодня") return " · ".join(bits) def order_units(db: Session, equipment_id: int, only_due: bool) -> list[tuple[ServiceUnit, dict]]: eq = db.get(Equipment, equipment_id) rate = daily_rate(db, equipment_id) units = db.scalars(select(ServiceUnit).where(ServiceUnit.equipment_id == equipment_id, ServiceUnit.is_active.is_(True)).order_by(ServiceUnit.id)).all() pairs = [(u, evaluate(u, eq.meter_hours or 0, rate)) for u in units] if only_due: pairs = [p for p in pairs if p[1]["state"] in ("due", "soon")] return pairs def apply_completed_order(db: Session, o: Order) -> None: """Наряд завершён: фиксируем наработку и сбрасываем узлы, отмеченные «обслужено». Повторный вызов безопасен.""" at = o.completed_at or config.now() eq = db.get(Equipment, o.equipment_id) if o.equipment_id else None for it in o.items: if it.is_meter and eq and it.value not in (None, ""): try: hours = float(it.value.replace(",", ".")) except ValueError: continue seen = db.scalar(select(MeterReading.id).where(MeterReading.order_id == o.id)) if not seen and hours >= (eq.meter_hours or 0): record_meter(db, eq, hours, at=at, source="order", user_id=o.executor_id, order_id=o.id) for it in o.items: if not it.service_unit_id or it.value != "done": continue u = db.get(ServiceUnit, it.service_unit_id) if not u: continue if db.scalar(select(ServiceEvent.id).where(ServiceEvent.unit_id == u.id, ServiceEvent.order_id == o.id)): continue reset_unit(db, u, at=at, kind="order", user_id=o.executor_id, order_id=o.id, note=f"Наряд №{o.id}") def unlink_unit(db: Session, unit_id: int) -> None: """Перед удалением узла: пункты нарядов остаются, но больше не ссылаются на него.""" db.execute(update(OrderItem).where(OrderItem.service_unit_id == unit_id).values(service_unit_id=None)) db.execute(update(IncidentUnit).where(IncidentUnit.unit_id == unit_id).values(unit_id=None)) db.execute(update(DailyTask).where(DailyTask.unit_id == unit_id).values(unit_id=None)) db.execute(ServiceEvent.__table__.delete().where(ServiceEvent.unit_id == unit_id))