75 lines
4.4 KiB
Python
75 lines
4.4 KiB
Python
# ============================================================
|
||
# 主数据查询辅助（moduleId: engines-queries, 可重生 ✅）
|
||
# 从 legacy common.js 的全局查找器按行移植（P0 只读）：
|
||
# findRoutingSteps / findBomItems / findProductLines /
|
||
# findWorkstationForOperation / getShiftMinutes / getLineShifts / getAvailableMinutes
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
from typing import Any # 类型标注
|
||
|
||
# 世界状态的类型别名（camelCase 键的字典）
|
||
World = dict[str, Any]
|
||
|
||
|
||
def find_routing_steps(world: World, product_id: int) -> list[dict]:
|
||
"""取产品默认工艺路线的步骤（按顺序号排序）。"""
|
||
routing = next((r for r in world["routings"] if r["productId"] == product_id and r["isDefault"]), None) # 默认路线
|
||
if not routing: # 无路线 → 空列表（上游生成 NO_LINE 类冲突）
|
||
return []
|
||
steps = [s for s in world["routingSteps"] if s["routingId"] == routing["id"]] # 该路线全部步骤
|
||
return sorted(steps, key=lambda s: s["sequenceNo"]) # 按工序顺序返回
|
||
|
||
|
||
def find_bom_items(world: World, product_id: int) -> list[dict]:
|
||
"""取产品默认 BOM 的明细行。"""
|
||
bom = next((b for b in world["boms"] if b["productId"] == product_id and b["isDefault"]), None) # 默认 BOM
|
||
if not bom: # 无 BOM → 空（视为无料耗）
|
||
return []
|
||
return [i for i in world["bomItems"] if i["bomId"] == bom["id"]] # 该 BOM 全部明细
|
||
|
||
|
||
def find_product_lines(world: World, product_id: int) -> list[dict]:
|
||
"""取产品可用产线配置（priority 越小越优先；仅 ACTIVE 产线，停用产线不参与新排产 MD-01）。"""
|
||
active_line_ids = {ln["id"] for ln in world["lines"] if ln.get("status", "ACTIVE") == "ACTIVE"} # 在用产线集
|
||
rows = [lp for lp in world["lineProducts"]
|
||
if lp["productId"] == product_id and lp["lineId"] in active_line_ids] # 该产品的在用产线配置
|
||
return sorted(rows, key=lambda lp: lp["priority"]) # 按优先级排序
|
||
|
||
|
||
def find_workstation_for_operation(world: World, line_id: int, operation_id: int) -> dict | None:
|
||
"""在指定产线上找能执行某工序的工位（工位工序配置 ∩ 产线工位；仅 ACTIVE 工位 MD-01）。"""
|
||
ws_ids = {wo["workstationId"] for wo in world["workstationOperations"] if wo["operationId"] == operation_id} # 能做该工序的工位集
|
||
return next((ws for ws in world["workstations"]
|
||
if ws["lineId"] == line_id and ws["id"] in ws_ids
|
||
and ws.get("status", "ACTIVE") == "ACTIVE"), None) # 限定产线取第一个在用工位
|
||
|
||
|
||
def _hm_to_minutes(hm: str) -> int:
|
||
"""'HH:MM' → 当日分钟数（班次时间换算的基元）。"""
|
||
h, m = hm.split(":") # 拆时与分
|
||
return int(h) * 60 + int(m) # 换算为分钟
|
||
|
||
|
||
def get_shift_minutes(world: World, shift_id: int) -> int:
|
||
"""某班次的有效工作分钟（扣除休息段，与 legacy getShiftMinutes 对齐）。"""
|
||
shift = next((s for s in world["shifts"] if s["id"] == shift_id), None) # 定位班次
|
||
if not shift: # 班次缺失防御
|
||
return 0
|
||
total = _hm_to_minutes(shift["endTime"]) - _hm_to_minutes(shift["startTime"]) # 毛时长
|
||
for bp in shift.get("breakPeriods", []): # 逐个休息段扣减
|
||
total -= _hm_to_minutes(bp["end"]) - _hm_to_minutes(bp["start"])
|
||
return max(total, 0) # 不为负
|
||
|
||
|
||
def get_line_shifts(world: World, line_id: int, date_str: str) -> list[dict]:
|
||
"""某产线某日的工作班次列表（班次日历 isWorking 过滤）。"""
|
||
shift_ids = [sc["shiftId"] for sc in world["shiftCalendar"] # 扫描班次日历
|
||
if sc["lineId"] == line_id and sc["date"] == date_str and sc["isWorking"]]
|
||
return [s for s in world["shifts"] if s["id"] in shift_ids] # 映射回班次对象
|
||
|
||
|
||
def get_available_minutes(world: World, line_id: int, date_str: str) -> int:
|
||
"""某产线某日的总可用分钟（各工作班次有效分钟之和）。"""
|
||
return sum(get_shift_minutes(world, s["id"]) for s in get_line_shifts(world, line_id, date_str)) # 累加
|