mirror of
https://github.com/frappe/erpnext.git
synced 2026-08-15 15:38:39 +00:00
feat: cross-plan load and overlap validation in production plan scheduling (#58127)
* feat: cross-plan load and overlap validation in production plan scheduling * fix: row-level job card exclusion and locking read in schedule overlap check * fix: operation-level job card exclusion and workstation locking in capacity check * fix: keep plan schedule load when job cards carry no booked time * fix: qty-coverage based job card exclusion for plan schedule load
This commit is contained in:
rohitwaghchaure
@@ -150,13 +150,14 @@
|
||||
],
|
||||
"index_web_pages_for_search": 0,
|
||||
"links": [],
|
||||
"modified": "2026-08-12 18:00:00.000000",
|
||||
"modified": "2026-08-13 11:00:00.000000",
|
||||
"modified_by": "Administrator",
|
||||
"module": "Manufacturing",
|
||||
"name": "Production Plan Schedule",
|
||||
"owner": "Administrator",
|
||||
"permissions": [
|
||||
{
|
||||
"delete": 1,
|
||||
"email": 1,
|
||||
"export": 1,
|
||||
"print": 1,
|
||||
@@ -166,6 +167,7 @@
|
||||
"share": 1
|
||||
},
|
||||
{
|
||||
"delete": 1,
|
||||
"email": 1,
|
||||
"export": 1,
|
||||
"print": 1,
|
||||
|
||||
@@ -1,9 +1,15 @@
|
||||
# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and contributors
|
||||
# For license information, please see license.txt
|
||||
|
||||
import itertools
|
||||
|
||||
import frappe
|
||||
from frappe import _
|
||||
from frappe.model.document import Document
|
||||
from frappe.utils import cint, get_datetime
|
||||
|
||||
from erpnext.manufacturing.doctype.job_card.job_card import OverlapError
|
||||
from erpnext.manufacturing.scheduling.loaders import filter_covered_schedule_rows
|
||||
|
||||
|
||||
class ProductionPlanSchedule(Document):
|
||||
@@ -38,6 +44,111 @@ class ProductionPlanSchedule(Document):
|
||||
)
|
||||
)
|
||||
|
||||
def validate(self):
|
||||
if get_datetime(self.from_time) >= get_datetime(self.to_time):
|
||||
frappe.throw(_("From Time must be before To Time"))
|
||||
|
||||
self.validate_workstation_overlap()
|
||||
|
||||
def validate_workstation_overlap(self):
|
||||
if not self.workstation:
|
||||
return
|
||||
|
||||
frappe.db.get_value("Workstation", self.workstation, "name", for_update=True)
|
||||
from_time, to_time = get_datetime(self.from_time), get_datetime(self.to_time)
|
||||
bookings = get_overlapping_bookings(self, from_time, to_time)
|
||||
if not bookings:
|
||||
return
|
||||
|
||||
capacity = cint(frappe.get_cached_value("Workstation", self.workstation, "production_capacity")) or 1
|
||||
conflict = get_capacity_conflict(bookings, from_time, to_time, capacity)
|
||||
if conflict:
|
||||
frappe.throw(
|
||||
_("Workstation {0} has no free capacity between {1} and {2}: overlaps with {3}").format(
|
||||
self.workstation, self.from_time, self.to_time, conflict
|
||||
),
|
||||
OverlapError,
|
||||
)
|
||||
|
||||
|
||||
def get_overlapping_bookings(doc, from_time, to_time):
|
||||
bookings = get_schedule_bookings(doc, from_time, to_time)
|
||||
bookings += get_job_card_bookings(doc, from_time, to_time, "Job Card Scheduled Time", drafts_only=True)
|
||||
bookings += get_job_card_bookings(doc, from_time, to_time, "Job Card Time Log", drafts_only=False)
|
||||
return bookings
|
||||
|
||||
|
||||
def get_schedule_bookings(doc, from_time, to_time):
|
||||
schedule = frappe.qb.DocType("Production Plan Schedule")
|
||||
plan = frappe.qb.DocType("Production Plan")
|
||||
|
||||
rows = (
|
||||
frappe.qb.from_(schedule)
|
||||
.join(plan)
|
||||
.on(schedule.production_plan == plan.name)
|
||||
.select(
|
||||
schedule.name.as_("source"),
|
||||
schedule.from_time,
|
||||
schedule.to_time,
|
||||
schedule.production_plan,
|
||||
schedule.plan_row,
|
||||
schedule.operation,
|
||||
)
|
||||
.where(
|
||||
(schedule.workstation == doc.workstation)
|
||||
& (schedule.name != (doc.name or "New"))
|
||||
& (schedule.from_time < to_time)
|
||||
& (schedule.to_time > from_time)
|
||||
& (plan.docstatus < 2)
|
||||
& (plan.status != "Closed")
|
||||
)
|
||||
.for_update()
|
||||
).run(as_dict=True)
|
||||
|
||||
return filter_covered_schedule_rows(rows)
|
||||
|
||||
|
||||
def get_job_card_bookings(doc, from_time, to_time, doctype, drafts_only):
|
||||
child = frappe.qb.DocType(doctype)
|
||||
job_card = frappe.qb.DocType("Job Card")
|
||||
|
||||
query = (
|
||||
frappe.qb.from_(child)
|
||||
.join(job_card)
|
||||
.on(child.parent == job_card.name)
|
||||
.select(job_card.name.as_("source"), child.from_time, child.to_time)
|
||||
.where(
|
||||
(job_card.workstation == doc.workstation)
|
||||
& (child.from_time < to_time)
|
||||
& (child.to_time > from_time)
|
||||
)
|
||||
)
|
||||
|
||||
if drafts_only:
|
||||
query = query.where((job_card.docstatus == 0) & (job_card.total_time_in_mins == 0))
|
||||
else:
|
||||
query = query.where(job_card.docstatus < 2)
|
||||
|
||||
return query.run(as_dict=True)
|
||||
|
||||
|
||||
def get_capacity_conflict(bookings, from_time, to_time, capacity):
|
||||
points = {from_time, to_time}
|
||||
for row in bookings:
|
||||
points.add(max(get_datetime(row.from_time), from_time))
|
||||
points.add(min(get_datetime(row.to_time), to_time))
|
||||
|
||||
for segment_start, segment_end in itertools.pairwise(sorted(points)):
|
||||
overlapping = [
|
||||
row
|
||||
for row in bookings
|
||||
if get_datetime(row.from_time) < segment_end and get_datetime(row.to_time) > segment_start
|
||||
]
|
||||
if len(overlapping) + 1 > capacity:
|
||||
return overlapping[0].source
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def on_doctype_update():
|
||||
frappe.db.add_index("Production Plan Schedule", ["production_plan"])
|
||||
|
||||
@@ -55,10 +55,11 @@ def get_workstation_calendar(row, settings):
|
||||
return ResourceCalendar(daily_windows=daily_windows, holidays=holidays)
|
||||
|
||||
|
||||
def get_booked_load(resource_names, from_date):
|
||||
def get_booked_load(resource_names, from_date, exclude_plan=None):
|
||||
load = defaultdict(list)
|
||||
add_booked_intervals(load, "Job Card Scheduled Time", resource_names, from_date, drafts_only=True)
|
||||
add_booked_intervals(load, "Job Card Time Log", resource_names, from_date, drafts_only=False)
|
||||
add_plan_schedule_intervals(load, resource_names, from_date, exclude_plan)
|
||||
return load
|
||||
|
||||
|
||||
@@ -85,6 +86,118 @@ def add_booked_intervals(load, doctype, resource_names, from_date, drafts_only):
|
||||
load[row.workstation].append(Interval(get_datetime(row.from_time), get_datetime(row.to_time)))
|
||||
|
||||
|
||||
def add_plan_schedule_intervals(load, resource_names, from_date, exclude_plan):
|
||||
schedule = frappe.qb.DocType("Production Plan Schedule")
|
||||
plan = frappe.qb.DocType("Production Plan")
|
||||
|
||||
query = (
|
||||
frappe.qb.from_(schedule)
|
||||
.join(plan)
|
||||
.on(schedule.production_plan == plan.name)
|
||||
.select(
|
||||
schedule.workstation,
|
||||
schedule.from_time,
|
||||
schedule.to_time,
|
||||
schedule.production_plan,
|
||||
schedule.plan_row,
|
||||
schedule.operation,
|
||||
)
|
||||
.where(
|
||||
schedule.workstation.isin(resource_names)
|
||||
& (schedule.to_time > from_date)
|
||||
& (plan.docstatus < 2)
|
||||
& (plan.status != "Closed")
|
||||
)
|
||||
)
|
||||
|
||||
if exclude_plan:
|
||||
query = query.where(schedule.production_plan != exclude_plan)
|
||||
|
||||
for row in filter_covered_schedule_rows(query.run(as_dict=True)):
|
||||
load[row.workstation].append(Interval(get_datetime(row.from_time), get_datetime(row.to_time)))
|
||||
|
||||
|
||||
def filter_covered_schedule_rows(rows):
|
||||
"""A schedule block steps aside only when job cards carrying booked load (scheduled
|
||||
time or logged time) cover the full quantity of its plan row and operation. Partial
|
||||
coverage - a batch-split card deleted, capacity planning disabled - keeps the block,
|
||||
trading double-booked load for never silently freeing a reserved interval."""
|
||||
plans = {row.production_plan for row in rows}
|
||||
if not plans:
|
||||
return rows
|
||||
|
||||
required, carried = get_job_card_coverage(plans)
|
||||
return [row for row in rows if not is_operation_covered(row, required, carried)]
|
||||
|
||||
|
||||
def is_operation_covered(row, required, carried):
|
||||
key = (row.production_plan, row.plan_row, row.operation)
|
||||
needed = required.get(key)
|
||||
return bool(needed) and flt(carried.get(key)) >= flt(needed)
|
||||
|
||||
|
||||
def get_job_card_coverage(production_plans):
|
||||
work_orders = frappe.get_all(
|
||||
"Work Order",
|
||||
filters={"production_plan": ("in", list(production_plans)), "docstatus": ("<", 2)},
|
||||
fields=[
|
||||
"name",
|
||||
"qty",
|
||||
"production_plan",
|
||||
"production_plan_item",
|
||||
"production_plan_sub_assembly_item",
|
||||
],
|
||||
)
|
||||
if not work_orders:
|
||||
return {}, {}
|
||||
|
||||
return get_required_operation_qty(work_orders), get_carried_operation_qty(work_orders)
|
||||
|
||||
|
||||
def get_required_operation_qty(work_orders):
|
||||
by_name = {row.name: row for row in work_orders}
|
||||
required = defaultdict(float)
|
||||
for operation_row in frappe.get_all(
|
||||
"Work Order Operation", filters={"parent": ("in", list(by_name))}, fields=["parent", "operation"]
|
||||
):
|
||||
work_order = by_name[operation_row.parent]
|
||||
add_operation_qty(required, work_order, operation_row.operation, flt(work_order.qty))
|
||||
|
||||
return required
|
||||
|
||||
|
||||
def get_carried_operation_qty(work_orders):
|
||||
by_name = {row.name: row for row in work_orders}
|
||||
job_cards = frappe.get_all(
|
||||
"Job Card",
|
||||
filters={"work_order": ("in", list(by_name)), "docstatus": ("<", 2)},
|
||||
fields=["name", "work_order", "operation", "for_quantity", "total_time_in_mins"],
|
||||
)
|
||||
|
||||
with_scheduled_time = get_job_cards_with_scheduled_time(job_cards)
|
||||
carried = defaultdict(float)
|
||||
for job_card in job_cards:
|
||||
if flt(job_card.total_time_in_mins) or job_card.name in with_scheduled_time:
|
||||
work_order = by_name[job_card.work_order]
|
||||
add_operation_qty(carried, work_order, job_card.operation, flt(job_card.for_quantity))
|
||||
|
||||
return carried
|
||||
|
||||
|
||||
def get_job_cards_with_scheduled_time(job_cards):
|
||||
names = [job_card.name for job_card in job_cards]
|
||||
if not names:
|
||||
return set()
|
||||
|
||||
return set(frappe.get_all("Job Card Scheduled Time", filters={"parent": ("in", names)}, pluck="parent"))
|
||||
|
||||
|
||||
def add_operation_qty(bucket, work_order, operation, qty):
|
||||
for plan_row in (work_order.production_plan_item, work_order.production_plan_sub_assembly_item):
|
||||
if plan_row:
|
||||
bucket[(work_order.production_plan, plan_row, operation)] += qty
|
||||
|
||||
|
||||
def build_bom_operation_tasks(bom_no, qty, prefix, earliest_start=None, priority=0):
|
||||
bom_qty = flt(frappe.get_cached_value("BOM", bom_no, "quantity")) or 1
|
||||
rows = frappe.get_all(
|
||||
|
||||
@@ -100,7 +100,11 @@ def run_engine(plan, start_date, use_item_dates=0, item_dates=None):
|
||||
settings = frappe.get_cached_doc("Manufacturing Settings")
|
||||
|
||||
resources = loaders.get_workstation_resources()
|
||||
load = loaders.get_booked_load([r.name for r in resources], start_date) if resources else {}
|
||||
load = (
|
||||
loaders.get_booked_load([r.name for r in resources], start_date, exclude_plan=plan.name)
|
||||
if resources
|
||||
else {}
|
||||
)
|
||||
engine = SchedulingEngine(
|
||||
resources,
|
||||
existing_load=load,
|
||||
@@ -426,6 +430,7 @@ def build_proposal(plan, result, task_info):
|
||||
|
||||
|
||||
def replace_schedule_entries(plan, proposal):
|
||||
lock_booked_workstations(proposal)
|
||||
frappe.db.delete("Production Plan Schedule", {"production_plan": plan.name})
|
||||
|
||||
for row_name, row in proposal["rows"].items():
|
||||
@@ -435,6 +440,21 @@ def replace_schedule_entries(plan, proposal):
|
||||
entry.insert(ignore_permissions=True)
|
||||
|
||||
|
||||
def lock_booked_workstations(proposal):
|
||||
workstations = sorted(
|
||||
{
|
||||
block["workstation"]
|
||||
for row in proposal["rows"].values()
|
||||
for block in row["blocks"]
|
||||
if block.get("workstation")
|
||||
}
|
||||
)
|
||||
if workstations:
|
||||
frappe.db.get_values(
|
||||
"Workstation", {"name": ("in", workstations)}, "name", order_by="name", for_update=True
|
||||
)
|
||||
|
||||
|
||||
def make_schedule_entry(plan, row_name, row, block):
|
||||
return frappe.get_doc(
|
||||
{
|
||||
|
||||
@@ -5,11 +5,13 @@ import frappe
|
||||
from frappe.tests.utils import change_settings
|
||||
from frappe.utils import add_to_date, get_datetime
|
||||
|
||||
from erpnext.manufacturing.doctype.job_card.job_card import OverlapError
|
||||
from erpnext.manufacturing.doctype.production_plan.test_production_plan import (
|
||||
create_production_plan,
|
||||
make_bom,
|
||||
)
|
||||
from erpnext.manufacturing.doctype.work_order.test_work_order import make_operation, make_workstation
|
||||
from erpnext.manufacturing.scheduling import loaders
|
||||
from erpnext.manufacturing.scheduling.plan_adapter import apply_schedule, get_schedule_preview
|
||||
from erpnext.stock.doctype.item.test_item import create_item
|
||||
from erpnext.tests.utils import ERPNextTestSuite
|
||||
@@ -35,20 +37,22 @@ class TestPlanAdapter(ERPNextTestSuite):
|
||||
self.make_bom_with_operation("Test PPS FG", ["Test PPS SA 1", "Test PPS SA 2"], time_in_mins=30)
|
||||
self.make_bom_with_operation("Test PPS FG 2", ["Test PPS RM"], time_in_mins=30)
|
||||
|
||||
def make_bom_with_operation(self, item, raw_materials, time_in_mins):
|
||||
def make_bom_with_operation(self, item, raw_materials, time_in_mins, operations=None, batch_size=0):
|
||||
if frappe.db.exists("BOM", {"item": item, "docstatus": 1}):
|
||||
return
|
||||
|
||||
bom = make_bom(item=item, raw_materials=raw_materials, with_operations=1, do_not_save=True)
|
||||
bom.append(
|
||||
"operations",
|
||||
{
|
||||
"operation": self.operation,
|
||||
"workstation": self.workstation,
|
||||
"time_in_mins": time_in_mins,
|
||||
"hour_rate": 100,
|
||||
},
|
||||
)
|
||||
for operation in operations or [self.operation]:
|
||||
bom.append(
|
||||
"operations",
|
||||
{
|
||||
"operation": operation,
|
||||
"workstation": self.workstation,
|
||||
"time_in_mins": time_in_mins,
|
||||
"hour_rate": 100,
|
||||
"batch_size": batch_size,
|
||||
},
|
||||
)
|
||||
bom.insert(ignore_permissions=True)
|
||||
bom.submit()
|
||||
|
||||
@@ -172,6 +176,257 @@ class TestPlanAdapter(ERPNextTestSuite):
|
||||
# freely from the dialog start date instead of the stale persisted one
|
||||
self.assertEqual(get_datetime(plan.po_items[0].planned_start_date), add_to_date(day_one, minutes=250))
|
||||
|
||||
@change_settings("Manufacturing Settings", {"mins_between_operations": 10, "allow_overtime": 0})
|
||||
def test_other_plan_schedule_blocks_are_treated_as_load(self):
|
||||
start_date = get_datetime("2027-01-04 09:00:00")
|
||||
plan_one = self.make_plan()
|
||||
apply_schedule(plan_one.name, start_date)
|
||||
|
||||
own_preview = get_schedule_preview(plan_one.name, start_date)
|
||||
own_starts = sorted(
|
||||
row["start"] for row in own_preview["rows"].values() if row["row_type"] == "Sub Assembly"
|
||||
)
|
||||
self.assertEqual(own_starts[0], start_date)
|
||||
|
||||
plan_two = self.make_plan()
|
||||
preview = get_schedule_preview(plan_two.name, start_date)
|
||||
self.assertFalse(preview["unscheduled"])
|
||||
self.assert_no_block_overlap(plan_one.name, preview)
|
||||
|
||||
plan_one.reload()
|
||||
plan_one.cancel()
|
||||
preview_after_cancel = get_schedule_preview(plan_two.name, start_date)
|
||||
two_starts = sorted(
|
||||
row["start"] for row in preview_after_cancel["rows"].values() if row["row_type"] == "Sub Assembly"
|
||||
)
|
||||
self.assertEqual(two_starts[0], start_date)
|
||||
|
||||
def assert_no_block_overlap(self, plan_name, preview):
|
||||
entries = frappe.get_all(
|
||||
"Production Plan Schedule",
|
||||
filters={"production_plan": plan_name},
|
||||
fields=["workstation", "from_time", "to_time"],
|
||||
)
|
||||
blocks = [
|
||||
block for row in preview["rows"].values() for block in row["blocks"] if block.get("workstation")
|
||||
]
|
||||
self.assertTrue(entries)
|
||||
self.assertTrue(blocks)
|
||||
|
||||
for entry in entries:
|
||||
for block in blocks:
|
||||
overlaps = get_datetime(entry.from_time) < block["to_time"] and block[
|
||||
"from_time"
|
||||
] < get_datetime(entry.to_time)
|
||||
self.assertFalse(overlaps, f"{block['task_key']} overlaps a block of {plan_name}")
|
||||
|
||||
@change_settings(
|
||||
"Manufacturing Settings",
|
||||
{"mins_between_operations": 10, "allow_overtime": 0, "disable_capacity_planning": 0},
|
||||
)
|
||||
def test_partially_submitted_plan_keeps_schedule_load(self):
|
||||
start_date = get_datetime("2027-03-01 09:00:00")
|
||||
plan_one = self.make_plan()
|
||||
apply_schedule(plan_one.name, start_date)
|
||||
|
||||
plan_one.reload()
|
||||
plan_one.make_work_order()
|
||||
work_order_name = frappe.get_all(
|
||||
"Work Order",
|
||||
filters={"production_plan": plan_one.name, "production_plan_sub_assembly_item": ("is", "set")},
|
||||
pluck="name",
|
||||
)[0]
|
||||
|
||||
work_order = frappe.get_doc("Work Order", work_order_name)
|
||||
work_order.wip_warehouse = "Work In Progress - _TC"
|
||||
work_order.fg_warehouse = work_order.fg_warehouse or "Finished Goods - _TC"
|
||||
work_order.submit()
|
||||
|
||||
plan_two = self.make_plan()
|
||||
preview = get_schedule_preview(plan_two.name, start_date)
|
||||
self.assertFalse(preview["unscheduled"])
|
||||
self.assert_no_block_overlap(plan_one.name, preview)
|
||||
|
||||
def test_schedule_entry_overlap_validation(self):
|
||||
workstation = "Test PPS WS Cap2"
|
||||
if not frappe.db.exists("Workstation", workstation):
|
||||
make_workstation(workstation=workstation, production_capacity=2)
|
||||
|
||||
plan = self.make_plan()
|
||||
|
||||
def make_entry(from_time, to_time):
|
||||
entry = frappe.get_doc(
|
||||
{
|
||||
"doctype": "Production Plan Schedule",
|
||||
"production_plan": plan.name,
|
||||
"row_type": "Finished Good",
|
||||
"item_code": "Test PPS FG",
|
||||
"workstation": workstation,
|
||||
"from_time": from_time,
|
||||
"to_time": to_time,
|
||||
}
|
||||
)
|
||||
entry.flags.from_scheduler = True
|
||||
return entry
|
||||
|
||||
make_entry("2027-02-01 09:00:00", "2027-02-01 10:00:00").insert()
|
||||
make_entry("2027-02-01 09:30:00", "2027-02-01 10:30:00").insert()
|
||||
|
||||
self.assertRaises(OverlapError, make_entry("2027-02-01 09:45:00", "2027-02-01 10:15:00").insert)
|
||||
self.assertRaises(
|
||||
frappe.ValidationError, make_entry("2027-02-01 12:00:00", "2027-02-01 11:00:00").insert
|
||||
)
|
||||
|
||||
make_entry("2027-02-01 10:30:00", "2027-02-01 11:30:00").insert()
|
||||
|
||||
@change_settings(
|
||||
"Manufacturing Settings",
|
||||
{"mins_between_operations": 10, "allow_overtime": 0, "disable_capacity_planning": 0},
|
||||
)
|
||||
def test_partial_job_cards_keep_remaining_schedule_load(self):
|
||||
operation_two = "Test PPS Op 2"
|
||||
if not frappe.db.exists("Operation", operation_two):
|
||||
make_operation(operation=operation_two, workstation=self.workstation)
|
||||
|
||||
item = "Test PPS FG 3"
|
||||
create_item(item, valuation_rate=100)
|
||||
self.make_bom_with_operation(
|
||||
item, ["Test PPS RM"], time_in_mins=60, operations=[self.operation, operation_two]
|
||||
)
|
||||
|
||||
plan = create_production_plan(
|
||||
item_code=item, planned_qty=1, do_not_submit=True, skip_getting_mr_items=True
|
||||
)
|
||||
plan.submit()
|
||||
|
||||
start_date = get_datetime("2027-04-05 09:00:00")
|
||||
apply_schedule(plan.name, start_date)
|
||||
|
||||
plan.reload()
|
||||
plan.make_work_order()
|
||||
work_order = frappe.get_doc("Work Order", {"production_plan": plan.name})
|
||||
work_order.wip_warehouse = "Work In Progress - _TC"
|
||||
work_order.fg_warehouse = work_order.fg_warehouse or "Finished Goods - _TC"
|
||||
work_order.submit()
|
||||
|
||||
job_card = frappe.db.get_value(
|
||||
"Job Card", {"work_order": work_order.name, "operation": operation_two}
|
||||
)
|
||||
frappe.delete_doc("Job Card", job_card)
|
||||
|
||||
entries = frappe.get_all(
|
||||
"Production Plan Schedule",
|
||||
filters={"production_plan": plan.name, "workstation": self.workstation},
|
||||
fields=["from_time", "to_time"],
|
||||
)
|
||||
expected = sorted((get_datetime(row.from_time), get_datetime(row.to_time)) for row in entries)
|
||||
self.assertEqual(len(expected), 2)
|
||||
|
||||
booked = sorted(
|
||||
(interval.start, interval.end)
|
||||
for interval in loaders.get_booked_load([self.workstation], start_date)[self.workstation]
|
||||
)
|
||||
self.assertEqual(booked, expected)
|
||||
|
||||
@change_settings(
|
||||
"Manufacturing Settings",
|
||||
{"mins_between_operations": 10, "allow_overtime": 0, "disable_capacity_planning": 1},
|
||||
)
|
||||
def test_batch_split_job_cards_keep_schedule_load(self):
|
||||
operation = "Test PPS Op Batch"
|
||||
if not frappe.db.exists("Operation", operation):
|
||||
make_operation(operation=operation, workstation=self.workstation)
|
||||
frappe.db.set_value("Operation", operation, "create_job_card_based_on_batch_size", 1)
|
||||
|
||||
item = "Test PPS FG 4"
|
||||
create_item(item, valuation_rate=100)
|
||||
self.make_bom_with_operation(
|
||||
item, ["Test PPS RM"], time_in_mins=30, operations=[operation], batch_size=1
|
||||
)
|
||||
|
||||
plan = create_production_plan(
|
||||
item_code=item, planned_qty=2, do_not_submit=True, skip_getting_mr_items=True
|
||||
)
|
||||
plan.submit()
|
||||
|
||||
start_date = get_datetime("2027-05-03 09:00:00")
|
||||
apply_schedule(plan.name, start_date)
|
||||
|
||||
plan.reload()
|
||||
plan.make_work_order()
|
||||
work_order = frappe.get_doc("Work Order", {"production_plan": plan.name})
|
||||
work_order.wip_warehouse = "Work In Progress - _TC"
|
||||
work_order.fg_warehouse = work_order.fg_warehouse or "Finished Goods - _TC"
|
||||
work_order.submit()
|
||||
|
||||
job_cards = frappe.get_all("Job Card", filters={"work_order": work_order.name}, pluck="name")
|
||||
self.assertEqual(len(job_cards), 2)
|
||||
self.assertFalse(
|
||||
frappe.db.exists("Job Card Scheduled Time", {"parent": ("in", job_cards)}),
|
||||
)
|
||||
|
||||
entries = frappe.get_all(
|
||||
"Production Plan Schedule",
|
||||
filters={"production_plan": plan.name, "workstation": self.workstation},
|
||||
fields=["from_time", "to_time"],
|
||||
)
|
||||
self.assertTrue(entries)
|
||||
|
||||
booked = {
|
||||
(interval.start, interval.end)
|
||||
for interval in loaders.get_booked_load([self.workstation], start_date)[self.workstation]
|
||||
}
|
||||
for row in entries:
|
||||
self.assertIn((get_datetime(row.from_time), get_datetime(row.to_time)), booked)
|
||||
|
||||
@change_settings(
|
||||
"Manufacturing Settings",
|
||||
{"mins_between_operations": 10, "allow_overtime": 0, "disable_capacity_planning": 0},
|
||||
)
|
||||
def test_partially_covered_batch_split_restores_schedule_load(self):
|
||||
operation = "Test PPS Op Batch"
|
||||
if not frappe.db.exists("Operation", operation):
|
||||
make_operation(operation=operation, workstation=self.workstation)
|
||||
frappe.db.set_value("Operation", operation, "create_job_card_based_on_batch_size", 1)
|
||||
|
||||
item = "Test PPS FG 5"
|
||||
create_item(item, valuation_rate=100)
|
||||
self.make_bom_with_operation(
|
||||
item, ["Test PPS RM"], time_in_mins=30, operations=[operation], batch_size=1
|
||||
)
|
||||
|
||||
plan = create_production_plan(
|
||||
item_code=item, planned_qty=2, do_not_submit=True, skip_getting_mr_items=True
|
||||
)
|
||||
plan.submit()
|
||||
|
||||
start_date = get_datetime("2027-06-07 09:00:00")
|
||||
apply_schedule(plan.name, start_date)
|
||||
|
||||
plan.reload()
|
||||
plan.make_work_order()
|
||||
work_order = frappe.get_doc("Work Order", {"production_plan": plan.name})
|
||||
work_order.wip_warehouse = "Work In Progress - _TC"
|
||||
work_order.fg_warehouse = work_order.fg_warehouse or "Finished Goods - _TC"
|
||||
work_order.submit()
|
||||
|
||||
job_cards = frappe.get_all("Job Card", filters={"work_order": work_order.name}, pluck="name")
|
||||
self.assertEqual(len(job_cards), 2)
|
||||
scheduled_rows = frappe.db.count("Job Card Scheduled Time", {"parent": ("in", job_cards)})
|
||||
self.assertTrue(scheduled_rows)
|
||||
|
||||
booked = loaders.get_booked_load([self.workstation], start_date)[self.workstation]
|
||||
self.assertEqual(len(booked), scheduled_rows)
|
||||
|
||||
frappe.delete_doc("Job Card", job_cards[1])
|
||||
|
||||
entry_count = frappe.db.count(
|
||||
"Production Plan Schedule", {"production_plan": plan.name, "workstation": self.workstation}
|
||||
)
|
||||
remaining_rows = frappe.db.count("Job Card Scheduled Time", {"parent": ("in", job_cards)})
|
||||
booked = loaders.get_booked_load([self.workstation], start_date)[self.workstation]
|
||||
self.assertEqual(len(booked), remaining_rows + entry_count)
|
||||
|
||||
def test_manual_schedule_entry_creation_is_blocked(self):
|
||||
plan = self.make_plan()
|
||||
entry = frappe.get_doc(
|
||||
|
||||
Reference in New Issue
Block a user