664 lines
24 KiB
Python
664 lines
24 KiB
Python
from collections import defaultdict
|
||
from dataclasses import dataclass
|
||
from datetime import date, datetime, time, timedelta
|
||
from typing import Protocol
|
||
|
||
from app.services.common import as_float, normalize_device_no, round1, round2
|
||
from app.services.misc_work import is_misc_item
|
||
from app.services.work_schedule import DEFAULT_WORK_SCHEDULE_CONFIG, WorkScheduleConfig
|
||
|
||
|
||
class ReportItemLike(Protocol):
|
||
device_no: str
|
||
product_name: str
|
||
process_name: str | None
|
||
started_at: datetime | None
|
||
standard_beat: float
|
||
standard_workload: float
|
||
stamping_method: str | None
|
||
good_qty: float
|
||
defect_qty: float
|
||
scrap_qty: float
|
||
allocated_minutes: float
|
||
|
||
|
||
class ReportDeviceLike(Protocol):
|
||
device_no: str
|
||
process_name: str | None
|
||
scanned_at: datetime
|
||
released_at: datetime | None
|
||
sort_order: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ItemTimeRange:
|
||
item: ReportItemLike
|
||
start_at: datetime
|
||
end_at: datetime
|
||
|
||
|
||
SHIFT_BUCKETS = ("day", "overtime", "night")
|
||
SHIFT_LABELS = {
|
||
"day": "白班",
|
||
"overtime": "加班",
|
||
"night": "夜班",
|
||
}
|
||
MEAL_END_GRACE_MINUTES = 5
|
||
|
||
|
||
def minutes_between(start_at: datetime, end_at: datetime) -> float:
|
||
seconds = max(0.0, (end_at - start_at).total_seconds())
|
||
return seconds / 60
|
||
|
||
|
||
def _combine(day: date, value: time) -> datetime:
|
||
return datetime.combine(day, value)
|
||
|
||
|
||
def _parse_schedule_time(value: str) -> time:
|
||
hour, minute = [int(part) for part in str(value).split(":", 1)]
|
||
return time(hour, minute)
|
||
|
||
|
||
def _schedule_interval(base_date: date, start_text: str, end_text: str) -> tuple[datetime, datetime]:
|
||
start = _combine(base_date, _parse_schedule_time(start_text))
|
||
end = _combine(base_date, _parse_schedule_time(end_text))
|
||
if end <= start:
|
||
end += timedelta(days=1)
|
||
return start, end
|
||
|
||
|
||
def _overlap_minutes(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
interval_start: datetime,
|
||
interval_end: datetime,
|
||
) -> float:
|
||
start = max(start_at, interval_start)
|
||
end = min(end_at, interval_end)
|
||
return minutes_between(start, end) if end > start else 0.0
|
||
|
||
|
||
def _clip_interval(
|
||
interval_start: datetime,
|
||
interval_end: datetime,
|
||
boundary_start: datetime,
|
||
boundary_end: datetime,
|
||
) -> tuple[datetime, datetime] | None:
|
||
start = max(interval_start, boundary_start)
|
||
end = min(interval_end, boundary_end)
|
||
if end <= start:
|
||
return None
|
||
return start, end
|
||
|
||
|
||
def _merge_intervals(intervals: list[tuple[datetime, datetime]]) -> list[tuple[datetime, datetime]]:
|
||
ordered = sorted(intervals, key=lambda item: item[0])
|
||
merged: list[tuple[datetime, datetime]] = []
|
||
for start, end in ordered:
|
||
if not merged or start > merged[-1][1]:
|
||
merged.append((start, end))
|
||
elif end > merged[-1][1]:
|
||
merged[-1] = (merged[-1][0], end)
|
||
return merged
|
||
|
||
|
||
def _subtract_intervals(
|
||
base_start: datetime,
|
||
base_end: datetime,
|
||
occupied: list[tuple[datetime, datetime]],
|
||
) -> list[tuple[datetime, datetime]]:
|
||
gaps: list[tuple[datetime, datetime]] = []
|
||
cursor = base_start
|
||
for start, end in _merge_intervals(occupied):
|
||
if start > cursor:
|
||
gaps.append((cursor, start))
|
||
if end > cursor:
|
||
cursor = end
|
||
if cursor < base_end:
|
||
gaps.append((cursor, base_end))
|
||
return gaps
|
||
|
||
|
||
def _calendar_dates_between(start_at: datetime, end_at: datetime):
|
||
current = start_at.date()
|
||
last = end_at.date()
|
||
while current <= last:
|
||
yield current
|
||
current += timedelta(days=1)
|
||
|
||
|
||
def _blank_overtime_intervals(
|
||
current_date: date,
|
||
schedule: WorkScheduleConfig,
|
||
) -> list[tuple[datetime, datetime]]:
|
||
day_start = datetime.combine(current_date, time.min)
|
||
day_end = day_start + timedelta(days=1)
|
||
occupied: list[tuple[datetime, datetime]] = []
|
||
raw_intervals = [
|
||
_schedule_interval(current_date, schedule.day_start, schedule.day_end),
|
||
_schedule_interval(current_date, schedule.overtime_start, schedule.overtime_end),
|
||
_schedule_interval(current_date, schedule.night_start, schedule.night_end),
|
||
_schedule_interval(current_date - timedelta(days=1), schedule.night_start, schedule.night_end),
|
||
]
|
||
for interval_start, interval_end in raw_intervals:
|
||
clipped = _clip_interval(interval_start, interval_end, day_start, day_end)
|
||
if clipped is not None:
|
||
occupied.append(clipped)
|
||
return _subtract_intervals(day_start, day_end, occupied)
|
||
|
||
|
||
def _covered_meal_overlap_minutes(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
meal_start: datetime,
|
||
meal_end: datetime,
|
||
) -> float:
|
||
if end_at <= meal_end + timedelta(minutes=MEAL_END_GRACE_MINUTES):
|
||
return 0.0
|
||
if start_at <= meal_start and end_at >= meal_end:
|
||
return _overlap_minutes(start_at, end_at, meal_start, meal_end)
|
||
return 0.0
|
||
|
||
|
||
def _iter_shift_base_dates(start_at: datetime, end_at: datetime):
|
||
current = start_at.date() - timedelta(days=1)
|
||
last = end_at.date()
|
||
while current <= last:
|
||
yield current
|
||
current += timedelta(days=1)
|
||
|
||
|
||
def _raw_shift_minutes(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
schedule: WorkScheduleConfig | None = None,
|
||
) -> dict[str, float]:
|
||
active_schedule = schedule or DEFAULT_WORK_SCHEDULE_CONFIG
|
||
minutes = {
|
||
"day": 0.0,
|
||
"overtime": 0.0,
|
||
"night": 0.0,
|
||
"meal_break": 0.0,
|
||
}
|
||
if end_at <= start_at:
|
||
return minutes
|
||
|
||
for current_date in _iter_shift_base_dates(start_at, end_at):
|
||
meal_intervals = [
|
||
_schedule_interval(current_date, active_schedule.lunch_start, active_schedule.lunch_end),
|
||
_schedule_interval(current_date, active_schedule.dinner_start, active_schedule.dinner_end),
|
||
]
|
||
shift_intervals = {
|
||
"day": [_schedule_interval(current_date, active_schedule.day_start, active_schedule.day_end)],
|
||
"overtime": [
|
||
_schedule_interval(current_date, active_schedule.overtime_start, active_schedule.overtime_end)
|
||
],
|
||
"night": [_schedule_interval(current_date, active_schedule.night_start, active_schedule.night_end)],
|
||
}
|
||
|
||
for meal_start, meal_end in meal_intervals:
|
||
minutes["meal_break"] += _covered_meal_overlap_minutes(
|
||
start_at,
|
||
end_at,
|
||
meal_start,
|
||
meal_end,
|
||
)
|
||
|
||
for bucket, intervals in shift_intervals.items():
|
||
for interval_start, interval_end in intervals:
|
||
raw_minutes = _overlap_minutes(start_at, end_at, interval_start, interval_end)
|
||
shift_start = max(start_at, interval_start)
|
||
shift_end = min(end_at, interval_end)
|
||
meal_minutes = sum(
|
||
_overlap_minutes(shift_start, shift_end, meal_start, meal_end)
|
||
for meal_start, meal_end in meal_intervals
|
||
if (
|
||
start_at <= meal_start
|
||
and end_at >= meal_end
|
||
and end_at > meal_end + timedelta(minutes=MEAL_END_GRACE_MINUTES)
|
||
)
|
||
)
|
||
minutes[bucket] += max(0.0, raw_minutes - meal_minutes)
|
||
|
||
for blank_start, blank_end in _blank_overtime_intervals(current_date, active_schedule):
|
||
raw_minutes = _overlap_minutes(start_at, end_at, blank_start, blank_end)
|
||
blank_start_at = max(start_at, blank_start)
|
||
blank_end_at = min(end_at, blank_end)
|
||
meal_minutes = sum(
|
||
_overlap_minutes(blank_start_at, blank_end_at, meal_start, meal_end)
|
||
for meal_start, meal_end in meal_intervals
|
||
if (
|
||
start_at <= meal_start
|
||
and end_at >= meal_end
|
||
and end_at > meal_end + timedelta(minutes=MEAL_END_GRACE_MINUTES)
|
||
)
|
||
)
|
||
minutes["overtime"] += max(0.0, raw_minutes - meal_minutes)
|
||
return minutes
|
||
|
||
|
||
def shift_minutes_between(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
break_minutes: float = 0,
|
||
schedule: WorkScheduleConfig | None = None,
|
||
) -> dict[str, float]:
|
||
return _raw_shift_minutes(start_at, end_at, schedule)
|
||
|
||
|
||
def effective_work_minutes_between(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
break_minutes: float = 0,
|
||
schedule: WorkScheduleConfig | None = None,
|
||
) -> float:
|
||
shift_minutes = shift_minutes_between(start_at, end_at, break_minutes, schedule)
|
||
return sum(shift_minutes.get(key, 0.0) for key in SHIFT_BUCKETS)
|
||
|
||
|
||
def format_hours(minutes: float) -> str:
|
||
text = f"{round(as_float(minutes) / 60, 2):.2f}".rstrip("0").rstrip(".")
|
||
return text or "0"
|
||
|
||
|
||
def shift_distribution_text(shift_minutes: dict[str, float]) -> str:
|
||
effective_minutes = sum(shift_minutes.get(key, 0.0) for key in SHIFT_BUCKETS)
|
||
parts = [
|
||
f"{SHIFT_LABELS[key]}{format_hours(shift_minutes.get(key, 0.0))}小时"
|
||
for key in SHIFT_BUCKETS
|
||
if round(as_float(shift_minutes.get(key, 0.0)), 2) > 0
|
||
]
|
||
detail = f"（{'、'.join(parts)}）" if parts else ""
|
||
return f"工时 {format_hours(effective_minutes)}小时{detail}"
|
||
|
||
|
||
def _set_item_shift_minutes(item: ReportItemLike, shift_minutes: dict[str, float]) -> None:
|
||
for key in SHIFT_BUCKETS:
|
||
setattr(item, f"_shift_{key}_minutes", as_float(shift_minutes.get(key, 0.0)))
|
||
|
||
|
||
def _add_item_shift_minutes(item: ReportItemLike, shift_minutes: dict[str, float]) -> None:
|
||
for key in SHIFT_BUCKETS:
|
||
setattr(
|
||
item,
|
||
f"_shift_{key}_minutes",
|
||
_get_item_shift_minutes(item, key) + as_float(shift_minutes.get(key, 0.0)),
|
||
)
|
||
|
||
|
||
def _get_item_shift_minutes(item: ReportItemLike, key: str) -> float:
|
||
return as_float(getattr(item, f"_shift_{key}_minutes", 0.0))
|
||
|
||
|
||
def _scale_item_shift_minutes(item: ReportItemLike, scale: float) -> None:
|
||
for key in SHIFT_BUCKETS:
|
||
setattr(item, f"_shift_{key}_minutes", _get_item_shift_minutes(item, key) * scale)
|
||
|
||
|
||
def _mold_key(product_name: str | None, process_name: str | None = "") -> str:
|
||
return f"{normalize_device_no(product_name)}\0{str(process_name or '').strip()}"
|
||
|
||
|
||
WORK_TIME_IGNORED_RELEASE_REASONS = {
|
||
"换模具释放占用",
|
||
"报工提交释放占用",
|
||
"超时-系统自动提交",
|
||
}
|
||
|
||
|
||
def release_at_caps_work_time(device: ReportDeviceLike) -> bool:
|
||
if getattr(device, "released_at", None) is None:
|
||
return False
|
||
reason = str(getattr(device, "release_reason", "") or "").strip()
|
||
return reason not in WORK_TIME_IGNORED_RELEASE_REASONS
|
||
|
||
|
||
def _device_segment_ranges(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
devices: list[ReportDeviceLike] | None,
|
||
) -> dict[str, list[tuple[datetime, datetime]]]:
|
||
ordered_devices = sorted(
|
||
devices or [],
|
||
key=lambda device: (int(getattr(device, "sort_order", 0) or 0), device.scanned_at),
|
||
)
|
||
if not ordered_devices:
|
||
return {}
|
||
|
||
segment_ranges: dict[str, list[tuple[datetime, datetime]]] = defaultdict(list)
|
||
for index, device in enumerate(ordered_devices):
|
||
segment_start = start_at if index == 0 else max(start_at, device.scanned_at)
|
||
segment_end = end_at
|
||
if index + 1 < len(ordered_devices):
|
||
segment_end = min(end_at, ordered_devices[index + 1].scanned_at)
|
||
released_at = getattr(device, "released_at", None)
|
||
if release_at_caps_work_time(device) and released_at is not None:
|
||
segment_end = min(segment_end, released_at)
|
||
if segment_end > segment_start:
|
||
segment_ranges[_mold_key(device.device_no, getattr(device, "process_name", ""))].append(
|
||
(segment_start, segment_end)
|
||
)
|
||
return segment_ranges
|
||
|
||
|
||
def _segment_ranges_shift_minutes(
|
||
segment_ranges: dict[str, list[tuple[datetime, datetime]]],
|
||
schedule: WorkScheduleConfig | None = None,
|
||
) -> dict[str, float]:
|
||
minutes = {key: 0.0 for key in SHIFT_BUCKETS}
|
||
for ranges in segment_ranges.values():
|
||
for segment_start, segment_end in ranges:
|
||
shift_minutes = shift_minutes_between(segment_start, segment_end, schedule=schedule)
|
||
for key in SHIFT_BUCKETS:
|
||
minutes[key] += shift_minutes.get(key, 0.0)
|
||
return minutes
|
||
|
||
|
||
def _clamp_datetime(value: datetime, start_at: datetime, end_at: datetime) -> datetime:
|
||
if value < start_at:
|
||
return start_at
|
||
if value > end_at:
|
||
return end_at
|
||
return value
|
||
|
||
|
||
def _allocate_device_item_minutes(
|
||
device_items: list[ReportItemLike],
|
||
segment_start: datetime,
|
||
segment_end: datetime,
|
||
schedule: WorkScheduleConfig | None = None,
|
||
) -> None:
|
||
if not device_items:
|
||
return
|
||
|
||
grouped_items: dict[datetime, list[ReportItemLike]] = defaultdict(list)
|
||
for item in device_items:
|
||
item_start = getattr(item, "started_at", None) or segment_start
|
||
grouped_items[_clamp_datetime(item_start, segment_start, segment_end)].append(item)
|
||
first_item_start = min(grouped_items)
|
||
if first_item_start > segment_start:
|
||
grouped_items[segment_start].extend(grouped_items.pop(first_item_start))
|
||
|
||
ordered_starts = sorted(grouped_items)
|
||
for index, item_start in enumerate(ordered_starts):
|
||
item_end = segment_end
|
||
if index + 1 < len(ordered_starts):
|
||
item_end = ordered_starts[index + 1]
|
||
shift_minutes = shift_minutes_between(item_start, item_end, schedule=schedule)
|
||
minutes = sum(shift_minutes.get(key, 0.0) for key in SHIFT_BUCKETS)
|
||
items_at_start = grouped_items[item_start]
|
||
minutes_per_item = minutes / len(items_at_start)
|
||
for item in items_at_start:
|
||
item.allocated_minutes = as_float(item.allocated_minutes) + minutes_per_item
|
||
_add_item_shift_minutes(
|
||
item,
|
||
{key: shift_minutes.get(key, 0.0) / len(items_at_start) for key in SHIFT_BUCKETS},
|
||
)
|
||
|
||
|
||
def _device_item_time_ranges(
|
||
device_items: list[ReportItemLike],
|
||
segment_start: datetime,
|
||
segment_end: datetime,
|
||
) -> list[ItemTimeRange]:
|
||
if not device_items:
|
||
return []
|
||
|
||
grouped_items: dict[datetime, list[ReportItemLike]] = defaultdict(list)
|
||
for item in device_items:
|
||
item_start = getattr(item, "started_at", None) or segment_start
|
||
grouped_items[_clamp_datetime(item_start, segment_start, segment_end)].append(item)
|
||
first_item_start = min(grouped_items)
|
||
if first_item_start > segment_start:
|
||
grouped_items[segment_start].extend(grouped_items.pop(first_item_start))
|
||
|
||
ranges: list[ItemTimeRange] = []
|
||
ordered_starts = sorted(grouped_items)
|
||
for index, item_start in enumerate(ordered_starts):
|
||
item_end = segment_end
|
||
if index + 1 < len(ordered_starts):
|
||
item_end = ordered_starts[index + 1]
|
||
for item in grouped_items[item_start]:
|
||
ranges.append(ItemTimeRange(item=item, start_at=item_start, end_at=item_end))
|
||
return ranges
|
||
|
||
|
||
def item_time_ranges(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
items: list[ReportItemLike],
|
||
devices: list[ReportDeviceLike] | None = None,
|
||
) -> list[ItemTimeRange]:
|
||
if not items or end_at <= start_at:
|
||
return []
|
||
|
||
segment_ranges = _device_segment_ranges(start_at, end_at, devices)
|
||
if not segment_ranges:
|
||
return [ItemTimeRange(item=item, start_at=start_at, end_at=end_at) for item in items]
|
||
|
||
items_by_device: dict[str, list[ReportItemLike]] = defaultdict(list)
|
||
for item in items:
|
||
mold_key = _mold_key(
|
||
getattr(item, "product_name", "") or item.device_no,
|
||
getattr(item, "process_name", ""),
|
||
)
|
||
items_by_device[mold_key].append(item)
|
||
|
||
ranges: list[ItemTimeRange] = []
|
||
allocated_item_ids: set[int] = set()
|
||
for device_no, device_ranges in segment_ranges.items():
|
||
device_items = items_by_device.get(device_no, [])
|
||
if not device_items:
|
||
continue
|
||
for segment_start, segment_end in device_ranges:
|
||
ranges.extend(_device_item_time_ranges(device_items, segment_start, segment_end))
|
||
for item in device_items:
|
||
allocated_item_ids.add(id(item))
|
||
|
||
unmatched_items = [item for item in items if id(item) not in allocated_item_ids]
|
||
if unmatched_items:
|
||
ranges.extend(ItemTimeRange(item=item, start_at=start_at, end_at=end_at) for item in unmatched_items)
|
||
return ranges
|
||
|
||
|
||
def _allocate_item_minutes(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
items: list[ReportItemLike],
|
||
effective_minutes: float,
|
||
devices: list[ReportDeviceLike] | None,
|
||
schedule: WorkScheduleConfig | None = None,
|
||
) -> None:
|
||
if not items:
|
||
return
|
||
|
||
for item in items:
|
||
item.allocated_minutes = 0.0
|
||
_set_item_shift_minutes(item, {})
|
||
|
||
segment_ranges = _device_segment_ranges(start_at, end_at, devices)
|
||
if not segment_ranges:
|
||
shift_minutes = shift_minutes_between(start_at, end_at, schedule=schedule)
|
||
minutes_per_item = effective_minutes / len(items)
|
||
for item in items:
|
||
item.allocated_minutes = minutes_per_item
|
||
_set_item_shift_minutes(
|
||
item,
|
||
{key: shift_minutes.get(key, 0.0) / len(items) for key in SHIFT_BUCKETS},
|
||
)
|
||
return
|
||
|
||
covered_shift_minutes = _segment_ranges_shift_minutes(segment_ranges, schedule)
|
||
target_effective_minutes = min(effective_minutes, sum(covered_shift_minutes.get(key, 0.0) for key in SHIFT_BUCKETS))
|
||
target_shift_minutes = covered_shift_minutes
|
||
|
||
items_by_device: dict[str, list[ReportItemLike]] = defaultdict(list)
|
||
for item in items:
|
||
mold_key = _mold_key(
|
||
getattr(item, "product_name", "") or item.device_no,
|
||
getattr(item, "process_name", ""),
|
||
)
|
||
items_by_device[mold_key].append(item)
|
||
|
||
allocated_item_ids: set[int] = set()
|
||
for device_no, ranges in segment_ranges.items():
|
||
device_items = items_by_device.get(device_no, [])
|
||
if not device_items:
|
||
continue
|
||
for segment_range in ranges:
|
||
_allocate_device_item_minutes(device_items, segment_range[0], segment_range[1], schedule)
|
||
for item in device_items:
|
||
allocated_item_ids.add(id(item))
|
||
|
||
unmatched_items = [item for item in items if id(item) not in allocated_item_ids]
|
||
if unmatched_items:
|
||
allocated_minutes = sum(as_float(item.allocated_minutes) for item in items)
|
||
remaining_minutes = max(0.0, target_effective_minutes - allocated_minutes)
|
||
minutes_per_item = remaining_minutes / len(unmatched_items)
|
||
allocated_shift_minutes = {
|
||
key: sum(_get_item_shift_minutes(item, key) for item in items)
|
||
for key in SHIFT_BUCKETS
|
||
}
|
||
remaining_shift_minutes = {
|
||
key: max(0.0, target_shift_minutes.get(key, 0.0) - allocated_shift_minutes[key])
|
||
for key in SHIFT_BUCKETS
|
||
}
|
||
for item in unmatched_items:
|
||
item.allocated_minutes = minutes_per_item
|
||
_set_item_shift_minutes(
|
||
item,
|
||
{key: remaining_shift_minutes[key] / len(unmatched_items) for key in SHIFT_BUCKETS},
|
||
)
|
||
|
||
allocated_minutes = sum(as_float(item.allocated_minutes) for item in items)
|
||
if target_effective_minutes <= 0:
|
||
for item in items:
|
||
item.allocated_minutes = 0.0
|
||
_set_item_shift_minutes(item, {})
|
||
elif allocated_minutes <= 0:
|
||
minutes_per_item = target_effective_minutes / len(items)
|
||
for item in items:
|
||
item.allocated_minutes = minutes_per_item
|
||
_set_item_shift_minutes(
|
||
item,
|
||
{key: target_shift_minutes.get(key, 0.0) / len(items) for key in SHIFT_BUCKETS},
|
||
)
|
||
elif abs(allocated_minutes - target_effective_minutes) > 0.000001:
|
||
scale = target_effective_minutes / allocated_minutes
|
||
for item in items:
|
||
item.allocated_minutes = as_float(item.allocated_minutes) * scale
|
||
_scale_item_shift_minutes(item, scale)
|
||
|
||
|
||
def calculate_report_metrics(
|
||
start_at: datetime,
|
||
end_at: datetime,
|
||
items: list[ReportItemLike],
|
||
break_minutes: float = 0,
|
||
devices: list[ReportDeviceLike] | None = None,
|
||
schedule: WorkScheduleConfig | None = None,
|
||
) -> dict[str, float]:
|
||
segment_ranges = _device_segment_ranges(start_at, end_at, devices)
|
||
if segment_ranges:
|
||
duration_minutes = sum(
|
||
minutes_between(segment_start, segment_end)
|
||
for ranges in segment_ranges.values()
|
||
for segment_start, segment_end in ranges
|
||
)
|
||
shift_minutes = _segment_ranges_shift_minutes(segment_ranges, schedule)
|
||
shift_minutes["meal_break"] = sum(
|
||
shift_minutes_between(segment_start, segment_end, break_minutes, schedule).get("meal_break", 0.0)
|
||
for ranges in segment_ranges.values()
|
||
for segment_start, segment_end in ranges
|
||
)
|
||
else:
|
||
duration_minutes = minutes_between(start_at, end_at)
|
||
shift_minutes = shift_minutes_between(start_at, end_at, break_minutes, schedule)
|
||
effective_minutes = sum(shift_minutes.get(key, 0.0) for key in SHIFT_BUCKETS)
|
||
_allocate_item_minutes(start_at, end_at, items, effective_minutes, devices, schedule)
|
||
|
||
total_good_qty = 0.0
|
||
total_output_qty = 0.0
|
||
weighted_standard_beat = 0.0
|
||
standard_weight = 0.0
|
||
expected_workload = 0.0
|
||
productive_minutes = 0.0
|
||
|
||
for item in items:
|
||
if is_misc_item(item):
|
||
try:
|
||
item.standard_workload = 0
|
||
except AttributeError:
|
||
pass
|
||
continue
|
||
good_qty = as_float(item.good_qty)
|
||
defect_qty = as_float(item.defect_qty)
|
||
scrap_qty = as_float(item.scrap_qty)
|
||
output_qty = good_qty + defect_qty + scrap_qty
|
||
beat = as_float(item.standard_beat)
|
||
workload = as_float(item.allocated_minutes) * 60 / beat if beat > 0 else 0
|
||
weight = max(1.0, output_qty)
|
||
try:
|
||
item.standard_workload = round1(workload)
|
||
except AttributeError:
|
||
pass
|
||
|
||
total_good_qty += good_qty
|
||
total_output_qty += output_qty
|
||
weighted_standard_beat += beat * weight
|
||
standard_weight += weight
|
||
expected_workload += workload
|
||
productive_minutes += as_float(item.allocated_minutes)
|
||
|
||
actual_beat = productive_minutes * 60 / total_output_qty if total_output_qty > 0 else 0
|
||
standard_beat = weighted_standard_beat / standard_weight if standard_weight > 0 else 0
|
||
pace_rate = (actual_beat - standard_beat) / standard_beat * 100 if standard_beat > 0 else 0
|
||
workload_rate = (
|
||
(total_good_qty - expected_workload) / expected_workload * 100
|
||
if expected_workload > 0
|
||
else 0
|
||
)
|
||
|
||
return {
|
||
"duration_minutes": round2(duration_minutes),
|
||
"effective_minutes": round2(effective_minutes),
|
||
"shift_day_minutes": round2(shift_minutes["day"]),
|
||
"shift_overtime_minutes": round2(shift_minutes["overtime"]),
|
||
"shift_night_minutes": round2(shift_minutes["night"]),
|
||
"shift_other_minutes": 0,
|
||
"shift_meal_break_minutes": round2(shift_minutes["meal_break"]),
|
||
"shift_distribution_text": shift_distribution_text(shift_minutes),
|
||
"total_good_qty": round2(total_good_qty),
|
||
"total_output_qty": round2(total_output_qty),
|
||
"actual_beat": round2(actual_beat),
|
||
"standard_beat": round2(standard_beat),
|
||
"expected_workload": round1(expected_workload),
|
||
"pace_rate": round1(pace_rate),
|
||
"workload_rate": round1(workload_rate),
|
||
}
|
||
|
||
|
||
def build_result_text(metrics: dict[str, float]) -> str:
|
||
pace_rate = as_float(metrics.get("pace_rate"))
|
||
workload_rate = as_float(metrics.get("workload_rate"))
|
||
fast_abnormal = pace_rate < -30
|
||
large_abnormal = workload_rate > 30
|
||
pace_text = (
|
||
f"今日节拍慢于标准节拍{abs(pace_rate)}%"
|
||
if pace_rate > 0
|
||
else f"今日节拍快于标准节拍{abs(pace_rate)}%"
|
||
)
|
||
workload_text = (
|
||
f"今日工作量大于标准工作量{abs(workload_rate)}%"
|
||
if workload_rate >= 0
|
||
else f"今日工作量小于标准工作量{abs(workload_rate)}%"
|
||
)
|
||
if fast_abnormal or large_abnormal:
|
||
suffix = "数据明显超出标准，请自我核对一下是否报工填错。"
|
||
elif pace_rate <= 0 and workload_rate >= 0:
|
||
suffix = "今天真的很棒呢！"
|
||
else:
|
||
suffix = "后续需要加油喽！"
|
||
return f"{pace_text}，{workload_text}，{suffix}"
|