niuniu/JhHardwareWRS_BackPoint
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 按设备匹配顺序拆分归属工时

Browse Source
This commit is contained in:
souplearn 2026-07-25 03:33:49 +08:00
parent fbbea51d5c
commit 891c10ddea
2 changed files with 162 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

154
app/services/report_allocations.py
View File

@ -11,8 +11,10 @@ from app.services.metrics import (
SHIFT_BUCKETS,
SHIFT_LABELS,
_blank_overtime_intervals,
_device_item_time_ranges,
_device_segment_ranges,
_iter_shift_base_dates,
_mold_key,
_overlap_minutes,
_schedule_interval,
format_hours,
@ -131,32 +133,130 @@ def _minutes_total(rows: dict[date, dict[str, float]]) -> float:
return sum(sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS) for minutes in rows.values())
def _target_effective_minutes(report, schedule: WorkScheduleConfig | None = None) -> float:
def _empty_shift_minutes() -> dict[str, float]:
return {key: 0.0 for key in SHIFT_BUCKETS}
def _add_date_bucket_minutes(
target: dict[date, dict[str, float]],
source: dict[date, dict[str, float]],
*,
scale: float = 1.0,
) -> None:
for allocation_date, minutes in source.items():
row = target.setdefault(allocation_date, _empty_shift_minutes())
for key in SHIFT_BUCKETS:
row[key] += minutes.get(key, 0.0) * scale
def _bucket_totals(rows_by_date: dict[date, dict[str, float]]) -> dict[str, float]:
return {
key: sum(minutes.get(key, 0.0) for minutes in rows_by_date.values())
for key in SHIFT_BUCKETS
}
def _item_bucket_totals(minutes_by_item: dict[int, dict[date, dict[str, float]]]) -> dict[str, float]:
totals = _empty_shift_minutes()
for rows_by_date in minutes_by_item.values():
for key, value in _bucket_totals(rows_by_date).items():
totals[key] += value
return totals
def _scale_item_minutes_to_bucket_targets(
minutes_by_item: dict[int, dict[date, dict[str, float]]],
target_buckets: dict[str, float],
) -> None:
raw_buckets = _item_bucket_totals(minutes_by_item)
for key in SHIFT_BUCKETS:
raw_value = raw_buckets.get(key, 0.0)
target_value = target_buckets.get(key, 0.0)
if raw_value <= 0 or abs(raw_value - target_value) <= 0.000001:
continue
scale = target_value / raw_value
for rows_by_date in minutes_by_item.values():
for minutes in rows_by_date.values():
minutes[key] = minutes.get(key, 0.0) * scale
def _items_by_device(items: list) -> dict[str, list]:
grouped: dict[str, list] = defaultdict(list)
for item in items:
mold_key = _mold_key(
getattr(item, "product_name", "") or item.device_no,
getattr(item, "process_name", ""),
)
grouped[mold_key].append(item)
return grouped
def _device_minutes_by_item(report, items: list, schedule: WorkScheduleConfig | None = None):
start_at = getattr(report, "start_at")
end_at = getattr(report, "end_at")
segment_ranges = _device_segment_ranges(start_at, end_at, _report_devices(report))
if segment_ranges:
return sum(
_minutes_total(_range_allocation_minutes(segment_start, segment_end, schedule))
for ranges in segment_ranges.values()
for segment_start, segment_end in ranges
if not segment_ranges:
return None
items_by_device = _items_by_device(items)
minutes_by_item: dict[int, dict[date, dict[str, float]]] = {}
target_minutes: dict[date, dict[str, float]] = defaultdict(_empty_shift_minutes)
unassigned_minutes: dict[date, dict[str, float]] = defaultdict(_empty_shift_minutes)
allocated_item_ids: set[int] = set()
for device_no, ranges in segment_ranges.items():
device_items = items_by_device.get(device_no, [])
for segment_start, segment_end in ranges:
segment_minutes = _range_allocation_minutes(segment_start, segment_end, schedule)
_add_date_bucket_minutes(target_minutes, segment_minutes)
if not device_items:
_add_date_bucket_minutes(unassigned_minutes, segment_minutes)
continue
for item_range in _device_item_time_ranges(device_items, segment_start, segment_end):
item_minutes = minutes_by_item.setdefault(id(item_range.item), defaultdict(_empty_shift_minutes))
range_minutes = _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule)
_add_date_bucket_minutes(
item_minutes,
range_minutes,
scale=as_float(getattr(item_range, "allocation_weight", 1.0)),
)
return _minutes_total(_range_allocation_minutes(start_at, end_at, schedule))
for item in device_items:
allocated_item_ids.add(id(item))
def _scale_minutes_to_target(
minutes_by_item: dict[int, dict[date, dict[str, float]]],
target_effective_minutes: float,
) -> None:
raw_total = sum(_minutes_total(item_minutes) for item_minutes in minutes_by_item.values())
if raw_total <= 0 or abs(raw_total - target_effective_minutes) <= 0.000001:
return
scale = target_effective_minutes / raw_total
for item_minutes in minutes_by_item.values():
for minutes in item_minutes.values():
unmatched_items = [item for item in items if id(item) not in allocated_item_ids]
if unmatched_items and _minutes_total(unassigned_minutes) > 0:
target_buckets = _bucket_totals(target_minutes)
matched_buckets = _item_bucket_totals(minutes_by_item)
unassigned_buckets = _bucket_totals(unassigned_minutes)
for unmatched_item in unmatched_items:
item_minutes = minutes_by_item.setdefault(id(unmatched_item), defaultdict(_empty_shift_minutes))
for allocation_date, minutes in unassigned_minutes.items():
row = item_minutes.setdefault(allocation_date, _empty_shift_minutes())
for key in SHIFT_BUCKETS:
minutes[key] = minutes.get(key, 0.0) * scale
remaining = max(0.0, target_buckets.get(key, 0.0) - matched_buckets.get(key, 0.0))
raw_unassigned = unassigned_buckets.get(key, 0.0)
if remaining <= 0 or raw_unassigned <= 0:
continue
row[key] += minutes.get(key, 0.0) * (remaining / raw_unassigned) / len(unmatched_items)
return minutes_by_item
_scale_item_minutes_to_bucket_targets(minutes_by_item, _bucket_totals(target_minutes))
return minutes_by_item
def _range_minutes_by_item(report, items: list, schedule: WorkScheduleConfig | None = None):
ranges = item_time_ranges(getattr(report, "start_at"), getattr(report, "end_at"), items, _report_devices(report))
minutes_by_item: dict[int, dict[date, dict[str, float]]] = {}
for item_range in ranges:
item_key = id(item_range.item)
target = minutes_by_item.setdefault(item_key, defaultdict(_empty_shift_minutes))
for allocation_date, minutes in _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule).items():
row = target.setdefault(allocation_date, _empty_shift_minutes())
for key in SHIFT_BUCKETS:
row[key] += minutes.get(key, 0.0) * as_float(
getattr(item_range, "allocation_weight", 1.0)
)
return minutes_by_item
def _fallback_row(report, item) -> ReportAllocationDraft:
@ -202,17 +302,9 @@ def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None =
if not items:
return []
ranges = item_time_ranges(getattr(report, "start_at"), getattr(report, "end_at"), items, _report_devices(report))
minutes_by_item: dict[int, dict[date, dict[str, float]]] = {}
for item_range in ranges:
item_key = id(item_range.item)
target = minutes_by_item.setdefault(item_key, defaultdict(lambda: {key: 0.0 for key in SHIFT_BUCKETS}))
for allocation_date, minutes in _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule).items():
for key in SHIFT_BUCKETS:
target[allocation_date][key] += minutes.get(key, 0.0) * as_float(
getattr(item_range, "allocation_weight", 1.0)
)
_scale_minutes_to_target(minutes_by_item, _target_effective_minutes(report, schedule))
minutes_by_item = _device_minutes_by_item(report, items, schedule)
if minutes_by_item is None:
minutes_by_item = _range_minutes_by_item(report, items, schedule)
rows: list[ReportAllocationDraft] = []
for item in items:

43
tests/test_report_allocations.py
View File

@ -52,14 +52,14 @@ def _item(
)
def _device(scanned_at: datetime):
def _device(scanned_at: datetime, *, device_no: str = "23#", process_name: str = "冲压", sort_order: int = 0):
return SimpleNamespace(
device_no="23#",
process_name="冲压",
device_no=device_no,
process_name=process_name,
scanned_at=scanned_at,
released_at=None,
release_reason=None,
sort_order=0,
sort_order=sort_order,
)
@ -262,6 +262,41 @@ def test_same_device_item_split_inside_meal_is_rescaled_to_metrics_effective_min
assert sum(row.effective_minutes for row in rows) == 240
def test_matched_meal_split_and_unmatched_device_range_follow_metrics_order():
start_at = datetime(2026, 7, 25, 8, 0)
second_started_at = datetime(2026, 7, 25, 12, 0)
switch_at = datetime(2026, 7, 25, 13, 0)
end_at = datetime(2026, 7, 25, 20, 0)
devices = [
_device(start_at, device_no="23#", sort_order=0),
_device(switch_at, device_no="77#", sort_order=1),
]
first_item = _item(id=1, good_qty=240, device_no="23#", started_at=start_at)
second_item = _item(id=2, good_qty=60, device_no="23#", started_at=second_started_at)
unmatched_item = _item(id=3, good_qty=320, device_no="99#")
items = [first_item, second_item, unmatched_item]
report = _report(start_at, end_at, items, devices=devices)
metrics = calculate_report_metrics(start_at, end_at, items, devices=devices)
rows = build_report_allocation_drafts(report)
assert metrics["effective_minutes"] == 620
assert {item.id: item.allocated_minutes for item in items} == {
1: 240,
2: 60,
3: 320,
}
assert {
item_id: round(sum(row.effective_minutes for row in rows if row.report_item_id == item_id), 2)
for item_id in [1, 2, 3]
} == {
1: 240,
2: 60,
3: 320,
}
assert round(sum(row.effective_minutes for row in rows), 2) == metrics["effective_minutes"]
def test_zero_effective_minutes_falls_back_to_original_report_date():
item = _item(good_qty=5, defect_qty=1, scrap_qty=0.5, changeover_count=2, process_unit_price_yuan=3)
report = _report(