feat(analytics): 设备流转甘特图数据底座（lead_time/started_at/task_type 主次标识）
- FlowDevice 新增 lead_time（生命周期总时长）、started_at（最早介入时间） - FlowSeries.data 改为区间数组 [device_index, start_offset, end_offset, task_name, duration, is_main] - get_flow_compare 按任务生成相对 T0 的偏移区间，task_type=SPAWN 标记为分支
This commit is contained in:
@ -51,11 +51,13 @@ class FlowDevice(BaseModel):
|
||||
external_serial: str | None
|
||||
material_name: str
|
||||
spec_model: str
|
||||
lead_time: float # 设备生命周期总时长（小时）= max_end - min_start
|
||||
started_at: str # 设备最早介入时间（T0），格式 MM-DD HH:mm
|
||||
|
||||
|
||||
class FlowSeries(BaseModel):
|
||||
name: str # 工序节点名
|
||||
data: list[float | None] # 每台设备在该工序的总耗时（未经过为 None）
|
||||
name: str # 人员姓名
|
||||
data: list[list] # 每项 = [device_index, start_offset, end_offset, task_name, duration]（小时）
|
||||
|
||||
|
||||
class FlowResponse(BaseModel):
|
||||
@ -255,17 +257,16 @@ async def get_flow_compare(
|
||||
spec_models: list[str] | None = None,
|
||||
) -> FlowResponse:
|
||||
"""
|
||||
查询每台设备上各操作人的累计耗时，拼装为堆叠柱状图：
|
||||
x 轴 = 设备身份证，每个「人员」一个 series，值为该员工在这台设备上的总耗时，
|
||||
谁的色块最长谁就是该设备的瓶颈。
|
||||
查询每台设备上各操作人的任务时间区间，拼装为「生命周期时间轴」区间图：
|
||||
x 轴 = 设备身份证，y 轴 = 相对该设备 T0（最早介入时间）的小时偏移。
|
||||
每个任务一根悬空区间柱（start_offset -> end_offset），并行任务可并排显示。
|
||||
|
||||
设备来源：
|
||||
- 传 product_sns：按给定身份证；
|
||||
- 仅传 spec_models：这些型号下最近有流转的 20 台设备；
|
||||
- 都未传：返回空。
|
||||
|
||||
排序：人员按「最早介入该设备的真实时间」（received_at/created_at 最小值）升序堆叠。
|
||||
未参与某设备的人员返回 None（null），不补 0。
|
||||
data 每项 = [device_index, start_offset, end_offset, task_name, duration]（单位小时）。
|
||||
"""
|
||||
from app.models.task import Task, TASK_STATUS_WIP, TASK_STATUS_PENDING, TASK_STATUS_COMPLETED
|
||||
from app.models.product import Product
|
||||
@ -309,6 +310,8 @@ async def get_flow_compare(
|
||||
FlowDevice(
|
||||
product_sn=r[1], external_serial=r[2],
|
||||
material_name=r[3] or "", spec_model=r[4] or "",
|
||||
lead_time=0.0,
|
||||
started_at="",
|
||||
)
|
||||
for r in product_rows
|
||||
]
|
||||
@ -321,11 +324,12 @@ async def get_flow_compare(
|
||||
sn_to_index = {d.product_sn: i for i, d in enumerate(devices)}
|
||||
product_ids = [r[0] for r in product_rows]
|
||||
|
||||
# ── 这些设备的全部任务（按操作人聚合） ──
|
||||
# ── 这些设备的全部任务（含工序名，用于区间图） ──
|
||||
task_rows = (await db.execute(
|
||||
select(
|
||||
Task.product_id, Task.assignee_id,
|
||||
Task.product_id, Task.assignee_id, Task.task_name,
|
||||
Task.received_at, Task.created_at, Task.completed_at,
|
||||
Task.task_type,
|
||||
)
|
||||
.where(
|
||||
Task.product_id.in_(product_ids),
|
||||
@ -334,41 +338,70 @@ async def get_flow_compare(
|
||||
)
|
||||
)).all()
|
||||
|
||||
# ── 聚合：人员(assignee_id) → {设备索引: 累计耗时}，同时记录人员最早介入时间 ──
|
||||
agg: dict[str, dict[int, float]] = {}
|
||||
person_earliest: dict[str, datetime] = {}
|
||||
# ── 解析为区间记录，并计算每台设备的 T0（最早）与 max_end（最晚） ──
|
||||
intervals: list[dict] = []
|
||||
t0_by_device: dict[int, datetime] = {}
|
||||
max_end_by_device: dict[int, datetime] = {}
|
||||
for row in task_rows:
|
||||
pid, assignee_id, received_at, created_at, completed_at = row
|
||||
pid, assignee_id, task_name, received_at, created_at, completed_at, task_type = row
|
||||
sn = id_to_sn.get(pid)
|
||||
if sn is None or sn not in sn_to_index:
|
||||
continue
|
||||
idx = sn_to_index[sn]
|
||||
start = _to_bj(received_at or created_at)
|
||||
end = _to_bj(completed_at) if completed_at else now
|
||||
agg.setdefault(assignee_id, {}).setdefault(idx, 0.0)
|
||||
agg[assignee_id][idx] += _duration_hours(start, end)
|
||||
if start and (person_earliest.get(assignee_id) is None or start < person_earliest[assignee_id]):
|
||||
person_earliest[assignee_id] = start
|
||||
if start is None:
|
||||
continue
|
||||
is_main = 0 if task_type == "SPAWN" else 1
|
||||
intervals.append({
|
||||
"idx": idx,
|
||||
"assignee_id": assignee_id,
|
||||
"task_name": (task_name or "").strip() or "未命名工序",
|
||||
"start": start,
|
||||
"end": end,
|
||||
"is_main": is_main,
|
||||
})
|
||||
if idx not in t0_by_device or start < t0_by_device[idx]:
|
||||
t0_by_device[idx] = start
|
||||
if idx not in max_end_by_device or end > max_end_by_device[idx]:
|
||||
max_end_by_device[idx] = end
|
||||
|
||||
# ── 计算每台设备 lead_time（最大 end - 最小 start，小时），重建 devices ──
|
||||
lead_time_by_index = {
|
||||
idx: round((max_end_by_device[idx] - t0_by_device[idx]).total_seconds() / 3600, 1)
|
||||
for idx in t0_by_device
|
||||
}
|
||||
devices = [
|
||||
FlowDevice(
|
||||
product_sn=d.product_sn, external_serial=d.external_serial,
|
||||
material_name=d.material_name, spec_model=d.spec_model,
|
||||
lead_time=lead_time_by_index.get(i, 0.0),
|
||||
started_at=t0_by_device[i].strftime("%m-%d %H:%M") if i in t0_by_device else "",
|
||||
)
|
||||
for i, d in enumerate(devices)
|
||||
]
|
||||
|
||||
# ── 按人分组，转为相对 T0 的小时偏移区间 ──
|
||||
by_assignee: dict[str, list[list]] = {}
|
||||
for it in intervals:
|
||||
t0 = t0_by_device[it["idx"]]
|
||||
start_offset = round((it["start"] - t0).total_seconds() / 3600, 1)
|
||||
end_offset = round((it["end"] - t0).total_seconds() / 3600, 1)
|
||||
duration = round(end_offset - start_offset, 1)
|
||||
by_assignee.setdefault(it["assignee_id"], []).append(
|
||||
[it["idx"], start_offset, end_offset, it["task_name"], duration, it["is_main"]]
|
||||
)
|
||||
|
||||
# 翻译人员姓名
|
||||
raw_ids = list(agg.keys())
|
||||
raw_ids = list(by_assignee.keys())
|
||||
name_map: dict[str, str] = {}
|
||||
if raw_ids:
|
||||
from app.services.mom_cache import get_display_names
|
||||
name_map = get_display_names(raw_ids)
|
||||
|
||||
# ── 按人员最早介入时间升序排序（从下到上 = 谁先接手谁后接手） ──
|
||||
ordered_persons = sorted(agg.keys(), key=lambda p: person_earliest.get(p) or now)
|
||||
|
||||
series = [
|
||||
FlowSeries(
|
||||
name=name_map.get(p, p),
|
||||
data=[
|
||||
round(agg[p][i], 1) if i in agg[p] else None
|
||||
for i in range(len(devices))
|
||||
],
|
||||
)
|
||||
for p in ordered_persons
|
||||
FlowSeries(name=name_map.get(a, a), data=by_assignee[a])
|
||||
for a in sorted(by_assignee.keys())
|
||||
]
|
||||
return FlowResponse(devices=devices, series=series)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user