diff --git a/erpnext/manufacturing/doctype/production_plan_schedule/production_plan_schedule.json b/erpnext/manufacturing/doctype/production_plan_schedule/production_plan_schedule.json index 8e0ce97a1bb..26a8518a240 100644 --- a/erpnext/manufacturing/doctype/production_plan_schedule/production_plan_schedule.json +++ b/erpnext/manufacturing/doctype/production_plan_schedule/production_plan_schedule.json @@ -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, diff --git a/erpnext/manufacturing/doctype/production_plan_schedule/production_plan_schedule.py b/erpnext/manufacturing/doctype/production_plan_schedule/production_plan_schedule.py index da9e6e0efbc..f49c5abe171 100644 --- a/erpnext/manufacturing/doctype/production_plan_schedule/production_plan_schedule.py +++ b/erpnext/manufacturing/doctype/production_plan_schedule/production_plan_schedule.py @@ -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"]) diff --git a/erpnext/manufacturing/scheduling/loaders.py b/erpnext/manufacturing/scheduling/loaders.py index 2063879b2a1..6c0c1ece861 100644 --- a/erpnext/manufacturing/scheduling/loaders.py +++ b/erpnext/manufacturing/scheduling/loaders.py @@ -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( diff --git a/erpnext/manufacturing/scheduling/plan_adapter.py b/erpnext/manufacturing/scheduling/plan_adapter.py index 5f200bc071e..7c8c06ece41 100644 --- a/erpnext/manufacturing/scheduling/plan_adapter.py +++ b/erpnext/manufacturing/scheduling/plan_adapter.py @@ -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( { diff --git a/erpnext/manufacturing/scheduling/test_plan_adapter.py b/erpnext/manufacturing/scheduling/test_plan_adapter.py index 6c18d1637c3..1db3912e9ab 100644 --- a/erpnext/manufacturing/scheduling/test_plan_adapter.py +++ b/erpnext/manufacturing/scheduling/test_plan_adapter.py @@ -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(