300 lines
9.6 KiB
Python
300 lines
9.6 KiB
Python
import json
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
from app.services.cleaning import is_cleaning_item
|
||
from app.services.common import as_float, normalize_device_no, round2
|
||
from app.services.continuous_die import is_continuous_die_item
|
||
from app.services.display_names import point_mold_display_name
|
||
from app.services.misc_work import is_misc_item
|
||
from app.services.multi_person import is_multi_person_item
|
||
|
||
|
||
DEVICE_TYPE_STAMPING = "冲压设备"
|
||
DEVICE_TYPE_CLEANING = "清洗设备"
|
||
MOLD_TYPE_NORMAL = "普通模具"
|
||
MOLD_TYPE_CLEANING = "清洗模具"
|
||
MOLD_TYPE_CONTINUOUS_DIE = "连续模"
|
||
MOLD_TYPE_MULTI_PERSON = "多人工序"
|
||
METRIC_KIND_MINUTES = "minutes"
|
||
METRIC_KIND_QUANTITY = "quantity"
|
||
TAG_CLEANING = "清洗"
|
||
TAG_CONTINUOUS_DIE = "连续模"
|
||
TAG_MULTI_PERSON = "多人协作"
|
||
|
||
|
||
@dataclass
|
||
class UsageStatRow:
|
||
id: str
|
||
category: str
|
||
name: str
|
||
metric_kind: str
|
||
object_type: str
|
||
value: float = 0
|
||
report_count: int = 0
|
||
report_ids: set[int] = field(default_factory=set, repr=False)
|
||
tags: list[str] = field(default_factory=list)
|
||
attendance_point_name: str = ""
|
||
product_name: str = ""
|
||
process_name: str = ""
|
||
stamping_method: str = ""
|
||
|
||
|
||
class UsageStatAccumulator:
|
||
def __init__(self, category: str):
|
||
self.category = category
|
||
self._rows: dict[tuple[Any, ...], UsageStatRow] = {}
|
||
|
||
def add(
|
||
self,
|
||
key: tuple[Any, ...],
|
||
*,
|
||
name: str,
|
||
metric_kind: str,
|
||
object_type: str,
|
||
value: float,
|
||
tags: list[str] | None = None,
|
||
attendance_point_name: str = "",
|
||
product_name: str = "",
|
||
process_name: str = "",
|
||
stamping_method: str = "",
|
||
report_id: int | None = None,
|
||
) -> None:
|
||
row = self._rows.get(key)
|
||
if row is None:
|
||
row = UsageStatRow(
|
||
id=_build_row_id(
|
||
self.category,
|
||
attendance_point_name,
|
||
name,
|
||
product_name,
|
||
process_name,
|
||
stamping_method,
|
||
metric_kind,
|
||
),
|
||
category=self.category,
|
||
name=name,
|
||
metric_kind=metric_kind,
|
||
object_type=object_type,
|
||
tags=[],
|
||
attendance_point_name=attendance_point_name,
|
||
product_name=product_name,
|
||
process_name=process_name,
|
||
stamping_method=stamping_method,
|
||
)
|
||
self._rows[key] = row
|
||
elif _object_type_priority(object_type) > _object_type_priority(row.object_type):
|
||
row.object_type = object_type
|
||
|
||
row.value += as_float(value)
|
||
if report_id is None:
|
||
row.report_count += 1
|
||
elif report_id not in row.report_ids:
|
||
row.report_ids.add(report_id)
|
||
row.report_count += 1
|
||
for tag in tags or []:
|
||
if tag and tag not in row.tags:
|
||
row.tags.append(tag)
|
||
|
||
def rows(self) -> list[UsageStatRow]:
|
||
rows = list(self._rows.values())
|
||
for row in rows:
|
||
row.value = round2(row.value)
|
||
return rows
|
||
|
||
|
||
def split_cleaning_device_nos(value: str | None) -> list[str]:
|
||
device_nos: list[str] = []
|
||
for part in (part.strip() for part in re.split(r"[、,，]", _clean(value))):
|
||
if part and part not in device_nos:
|
||
device_nos.append(part)
|
||
return device_nos
|
||
|
||
|
||
def build_usage_stats(
|
||
*,
|
||
reports: list,
|
||
category: str,
|
||
equipment_type_by_key: dict[tuple[str, str], str],
|
||
) -> list[UsageStatRow]:
|
||
if category == "device":
|
||
return _build_device_usage_stats(reports, equipment_type_by_key)
|
||
if category == "mold":
|
||
return _build_mold_usage_stats(reports)
|
||
raise ValueError("category must be device or mold")
|
||
|
||
|
||
def build_usage_stats_from_allocations(
|
||
*,
|
||
allocations: list,
|
||
category: str,
|
||
equipment_type_by_key: dict[tuple[str, str], str],
|
||
) -> list[UsageStatRow]:
|
||
reports = []
|
||
for allocation in allocations:
|
||
item = allocation.item
|
||
if item is None:
|
||
continue
|
||
proxy = SimpleNamespace(
|
||
attendance_point_name=item.attendance_point_name,
|
||
product_name=item.product_name,
|
||
process_name=item.process_name,
|
||
stamping_method=item.stamping_method,
|
||
operator_count=item.operator_count,
|
||
device_no=item.device_no,
|
||
allocated_minutes=allocation.effective_minutes,
|
||
good_qty=allocation.good_qty,
|
||
report_id=allocation.report_id,
|
||
)
|
||
reports.append(SimpleNamespace(items=[proxy]))
|
||
return build_usage_stats(
|
||
reports=reports,
|
||
category=category,
|
||
equipment_type_by_key=equipment_type_by_key,
|
||
)
|
||
|
||
|
||
def _build_device_usage_stats(
|
||
reports: list,
|
||
equipment_type_by_key: dict[tuple[str, str], str],
|
||
) -> list[UsageStatRow]:
|
||
acc = UsageStatAccumulator(category="device")
|
||
for item, report_id in _iter_report_items(reports):
|
||
if is_misc_item(item):
|
||
continue
|
||
|
||
point_name = _clean(getattr(item, "attendance_point_name", None))
|
||
if is_cleaning_item(item):
|
||
device_nos = split_cleaning_device_nos(getattr(item, "device_no", None))
|
||
if not device_nos:
|
||
continue
|
||
value = as_float(getattr(item, "good_qty", 0)) / len(device_nos)
|
||
for device_no in device_nos:
|
||
object_type = equipment_type_by_key.get((point_name, device_no), DEVICE_TYPE_CLEANING)
|
||
acc.add(
|
||
(point_name, device_no, METRIC_KIND_QUANTITY),
|
||
name=device_no,
|
||
metric_kind=METRIC_KIND_QUANTITY,
|
||
object_type=object_type,
|
||
value=value,
|
||
tags=[TAG_CLEANING],
|
||
attendance_point_name=point_name,
|
||
report_id=report_id,
|
||
)
|
||
continue
|
||
|
||
device_no = normalize_device_no(getattr(item, "device_no", None))
|
||
if not device_no:
|
||
continue
|
||
acc.add(
|
||
(point_name, device_no, METRIC_KIND_MINUTES),
|
||
name=device_no,
|
||
metric_kind=METRIC_KIND_MINUTES,
|
||
object_type=equipment_type_by_key.get((point_name, device_no), DEVICE_TYPE_STAMPING),
|
||
value=as_float(getattr(item, "allocated_minutes", 0)),
|
||
attendance_point_name=point_name,
|
||
report_id=report_id,
|
||
)
|
||
return acc.rows()
|
||
|
||
|
||
def _build_mold_usage_stats(reports: list) -> list[UsageStatRow]:
|
||
acc = UsageStatAccumulator(category="mold")
|
||
for item, report_id in _iter_report_items(reports):
|
||
if is_misc_item(item):
|
||
continue
|
||
|
||
point_name = _clean(getattr(item, "attendance_point_name", None))
|
||
product_name = _clean(getattr(item, "product_name", None))
|
||
process_name = _clean(getattr(item, "process_name", None))
|
||
stamping_method = _clean(getattr(item, "stamping_method", None))
|
||
metric_kind = METRIC_KIND_QUANTITY if is_cleaning_item(item) else METRIC_KIND_MINUTES
|
||
value_field = "good_qty" if metric_kind == METRIC_KIND_QUANTITY else "allocated_minutes"
|
||
tags = _item_tags(item)
|
||
object_type = _mold_object_type(item)
|
||
|
||
acc.add(
|
||
(point_name, product_name, process_name, stamping_method, metric_kind),
|
||
name=point_mold_display_name(point_name, product_name, process_name, stamping_method),
|
||
metric_kind=metric_kind,
|
||
object_type=object_type,
|
||
value=as_float(getattr(item, value_field, 0)),
|
||
tags=tags,
|
||
attendance_point_name=point_name,
|
||
product_name=product_name,
|
||
process_name=process_name,
|
||
stamping_method=stamping_method,
|
||
report_id=report_id,
|
||
)
|
||
return acc.rows()
|
||
|
||
|
||
def _iter_report_items(reports: list):
|
||
for report in reports:
|
||
report_id = getattr(report, "id", None)
|
||
for item in getattr(report, "items", []) or []:
|
||
item_report_id = getattr(item, "report_id", None)
|
||
yield item, item_report_id if item_report_id is not None else report_id
|
||
|
||
|
||
def _item_tags(item) -> list[str]:
|
||
tags: list[str] = []
|
||
if is_cleaning_item(item):
|
||
tags.append(TAG_CLEANING)
|
||
if is_continuous_die_item(item):
|
||
tags.append(TAG_CONTINUOUS_DIE)
|
||
if is_multi_person_item(item):
|
||
tags.append(TAG_MULTI_PERSON)
|
||
return tags
|
||
|
||
|
||
def _mold_object_type(item) -> str:
|
||
if is_cleaning_item(item):
|
||
return MOLD_TYPE_CLEANING
|
||
if is_continuous_die_item(item):
|
||
return MOLD_TYPE_CONTINUOUS_DIE
|
||
if is_multi_person_item(item):
|
||
return MOLD_TYPE_MULTI_PERSON
|
||
return MOLD_TYPE_NORMAL
|
||
|
||
|
||
def _object_type_priority(value: str) -> int:
|
||
return {
|
||
MOLD_TYPE_NORMAL: 0,
|
||
DEVICE_TYPE_STAMPING: 0,
|
||
DEVICE_TYPE_CLEANING: 0,
|
||
MOLD_TYPE_MULTI_PERSON: 1,
|
||
MOLD_TYPE_CONTINUOUS_DIE: 2,
|
||
MOLD_TYPE_CLEANING: 3,
|
||
}.get(value, 0)
|
||
|
||
|
||
def _build_row_id(
|
||
category: str,
|
||
attendance_point_name: str,
|
||
name: str,
|
||
product_name: str,
|
||
process_name: str,
|
||
stamping_method: str,
|
||
metric_kind: str,
|
||
) -> str:
|
||
return json.dumps(
|
||
[
|
||
_clean(category),
|
||
_clean(attendance_point_name),
|
||
_clean(name),
|
||
_clean(product_name),
|
||
_clean(process_name),
|
||
_clean(stamping_method),
|
||
_clean(metric_kind),
|
||
],
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
)
|
||
|
||
|
||
def _clean(value: str | None) -> str:
|
||
return "" if value is None else str(value).strip()
|