pickling-mes/backend/app/services/line_service.py

"""产线联动引擎：计划上线 / 上卷鞍座 / 生产实绩 / 停机自动检测。

状态流转：
  ready(准备好) --自动--> online(在线, 队首唯一) --移动--> 上卷鞍座(staged)
  --有速度/投入生产--> producing(生产中) --带头2000m--> produced(生产完成,产生实绩)

停机：当产线无生产(速度=0)持续超过 10min，自动新增一条未结束停机记录；
      恢复速度后自动结束该记录，原因由用户后续补充录入。
"""
from datetime import datetime
from sqlalchemy import select, asc
from sqlalchemy.ext.asyncio import AsyncSession
from loguru import logger

from app.models.plan import ProductionPlan
from app.models.production import ProductionRecord
from app.models.downtime import DowntimeRecord
from app.models.line_state import LineState

# ── 入口位置 ──
SADDLE_NAME = "上卷鞍座"        # 唯一会触发生产联动的位置
ENTRY_POSITIONS = [
    "1#上卷小车", "2#上卷小车",
    "1#称重位", "2#称重位",
    "1#地辊", "2#地辊",
    "1#倒卷小车", "2#倒卷小车",
    SADDLE_NAME,
]

# ── 仿真参数 ──
TARGET_LENGTH_M = 2000.0       # 带头目标长度（生产完成阈值）
SIM_SPEED_M_MIN = 600.0        # 仿真线速度 m/min（2000m ≈ 200s）
DOWNTIME_THRESHOLD_S = 600      # 速度为0持续超过该秒数判定停机（10min）


async def place_at_position(db: AsyncSession, plan: ProductionPlan, position: str):
    """把计划放到任意入口位置；位置唯一占用。上卷鞍座单独触发生产联动。"""
    if position not in ENTRY_POSITIONS:
        raise ValueError("未知入口位置")
    if position == SADDLE_NAME:
        await move_to_saddle(db, plan)
        return
    # 普通位置：清除同位置上的其它计划，移出鞍座（若在）
    res = await db.execute(
        select(ProductionPlan).where(ProductionPlan.position == position, ProductionPlan.id != plan.id)
    )
    for o in res.scalars():
        o.position = None
    plan.on_saddle = 0
    plan.run_started_at = None
    plan.run_speed = 0
    plan.run_length_m = 0
    plan.position = position
    if plan.status == "producing":
        plan.status = "online"
    await ensure_online(db)


def _shift_of(dt: datetime) -> str:
    return "甲" if 8 <= dt.hour < 20 else "乙"


async def _saddle_plan(db: AsyncSession):
    res = await db.execute(select(ProductionPlan).where(ProductionPlan.on_saddle == 1))
    return res.scalars().first()


async def ensure_online(db: AsyncSession):
    """保证恰好一条 online（队首，最早录入的 ready）。"""
    res = await db.execute(select(ProductionPlan).where(ProductionPlan.status == "online"))
    online = list(res.scalars())
    if len(online) > 1:
        # 仅保留最早的一条为 online，其余回退 ready
        online.sort(key=lambda p: (p.plan_date or datetime.max, p.id))
        for p in online[1:]:
            p.status = "ready"
        online = online[:1]
    if not online:
        res = await db.execute(
            select(ProductionPlan)
            .where(ProductionPlan.status == "ready")
            .order_by(asc(ProductionPlan.plan_date), asc(ProductionPlan.id))
            .limit(1)
        )
        head = res.scalar_one_or_none()
        if head:
            head.status = "online"


async def move_to_saddle(db: AsyncSession, plan: ProductionPlan):
    """把在线计划移动到上卷鞍座（staged，等待速度/投入生产）。"""
    occupied = await _saddle_plan(db)
    if occupied and occupied.id != plan.id:
        raise ValueError("上卷鞍座已被占用，请等待当前钢卷生产完成")
    plan.on_saddle = 1
    plan.position = SADDLE_NAME
    plan.saddle_at = datetime.now()
    plan.run_started_at = None
    plan.run_speed = 0
    plan.run_length_m = 0
    if plan.status not in ("producing", "produced"):
        plan.status = "online"
    await ensure_online(db)


async def commit_plan(db: AsyncSession, plan: ProductionPlan):
    """投入生产：鞍座计划有速度 → 生产中。"""
    if plan.on_saddle != 1:
        await move_to_saddle(db, plan)
    if plan.run_started_at is None:
        plan.run_started_at = datetime.now()
    plan.run_speed = SIM_SPEED_M_MIN
    plan.status = "producing"
    await ensure_online(db)


async def _produce(db: AsyncSession, plan: ProductionPlan):
    """带头到达目标长度 → 生产完成，并生成实绩记录。"""
    now = datetime.now()
    plan.status = "produced"
    plan.produced_at = now
    plan.on_saddle = 0
    plan.position = None
    plan.run_speed = 0
    plan.run_length_m = TARGET_LENGTH_M

    weight = plan.incoming_weight or plan.plan_weight or 0
    length = TARGET_LENGTH_M
    rec = ProductionRecord(
        coil_no=plan.cold_coil_no or plan.plan_no,
        sub_coil_no=plan.cold_coil_no,
        hot_coil_no=plan.hot_coil_no,
        plan_id=plan.id,
        shift=_shift_of(now),
        steel_grade=plan.steel_grade,
        incoming_thickness=plan.incoming_thickness,
        outlet_thickness=plan.product_thickness,
        deviation_upper=plan.deviation_upper,
        deviation_lower=plan.deviation_lower,
        incoming_width=plan.incoming_width,
        outlet_width=plan.product_width,
        incoming_weight=weight,
        weighed_weight=round(weight * 0.985, 4) if weight else None,
        packaging_req=plan.packaging_req,
        trim_req=plan.trim_req,
        surface_quality="合格",
        product_quality=99.0,
        product_length=length,
        length_per_ton=round(length / weight, 2) if weight else None,
        offline_time=now,
        status="PRODUCT",
        shift_date=now,
        start_time=plan.run_started_at or now,
        end_time=now,
        process_length=length,
        process_weight=weight,
        avg_speed=SIM_SPEED_M_MIN,
        max_speed=round(SIM_SPEED_M_MIN * 1.05, 1),
        inlet_thickness=plan.incoming_thickness,
        inlet_width=plan.incoming_width,
        quality_grade="A",
        operator="系统",
    )
    db.add(rec)
    logger.info(f"生产完成并产生实绩: {plan.cold_coil_no or plan.plan_no}")


async def advance_saddle(db: AsyncSession):
    """推进鞍座计划：staged 自动获得速度 → 生产中；累计长度到 2000m → 完成。"""
    plan = await _saddle_plan(db)
    if not plan:
        return
    now = datetime.now()
    # staged（在线且在鞍座）自动获得速度，模拟 PLC 速度信号
    if plan.status == "online" and plan.run_started_at is None:
        plan.run_started_at = now
        plan.run_speed = SIM_SPEED_M_MIN
        plan.status = "producing"
        await ensure_online(db)
    if plan.status == "producing" and plan.run_started_at:
        elapsed = (now - plan.run_started_at).total_seconds()
        plan.run_length_m = min(TARGET_LENGTH_M, (plan.run_speed or 0) / 60.0 * elapsed)
        if plan.run_length_m >= TARGET_LENGTH_M:
            await _produce(db, plan)


async def _get_line_state(db: AsyncSession) -> LineState:
    st = await db.get(LineState, 1)
    if not st:
        st = LineState(id=1, speed=0)
        db.add(st)
        await db.flush()
    return st


async def detect_downtime(db: AsyncSession):
    """速度为0持续超过阈值 → 自动新增停机；恢复速度 → 自动结束。"""
    res = await db.execute(select(ProductionPlan).where(ProductionPlan.status == "producing"))
    running = res.scalars().first() is not None
    now = datetime.now()
    st = await _get_line_state(db)

    if running:
        st.speed = SIM_SPEED_M_MIN
        st.zero_since = None
        if st.open_downtime_id:
            rec = await db.get(DowntimeRecord, st.open_downtime_id)
            if rec and rec.end_time is None:
                rec.end_time = now
                rec.duration = round((now - rec.start_time).total_seconds() / 60.0, 1)
            st.open_downtime_id = None
    else:
        st.speed = 0
        if st.zero_since is None:
            st.zero_since = now
        elif st.open_downtime_id is None and (now - st.zero_since).total_seconds() >= DOWNTIME_THRESHOLD_S:
            rec = DowntimeRecord(
                category_code="AUTO",
                category_name="待定",
                shift=_shift_of(now),
                shift_date=now,
                start_time=st.zero_since,
                fault_desc="线速度为0持续超过10分钟（系统自动检测，原因待补充）",
                reporter="系统",
                is_planned=0,
            )
            db.add(rec)
            await db.flush()
            st.open_downtime_id = rec.id
            logger.info("自动检测到停机，已新增待补充停机记录")


async def tick(db: AsyncSession):
    """引擎单步：上线 + 推进鞍座 + 停机检测。"""
    await ensure_online(db)
    await advance_saddle(db)
    await detect_downtime(db)


async def run_engine_loop(interval_s: int = 15):
    """后台循环，使联动在无人查看时也能自动推进。"""
    import asyncio
    from app.database import AsyncSessionLocal
    while True:
        try:
            async with AsyncSessionLocal() as db:
                await tick(db)
                await db.commit()
        except Exception as e:  # noqa
            logger.warning(f"line engine tick 失败: {e}")
        await asyncio.sleep(interval_s)