diff --git a/erpnext/manufacturing/doctype/production_plan/production_plan.js b/erpnext/manufacturing/doctype/production_plan/production_plan.js
index 5cba4e50eb4..292a36866a4 100644
--- a/erpnext/manufacturing/doctype/production_plan/production_plan.js
+++ b/erpnext/manufacturing/doctype/production_plan/production_plan.js
@@ -142,6 +142,21 @@ frappe.ui.form.on("Production Plan", {
__("View")
);
+ frm.add_custom_button(
+ __("Production Schedule"),
+ () => {
+ frappe.route_options = { production_plan: frm.doc.name };
+ frappe.set_route("List", "Production Plan Schedule", "Calendar");
+ },
+ __("View")
+ );
+
+ if (!["Completed", "Closed"].includes(frm.doc.status)) {
+ frm.add_custom_button(__("Schedule Items"), () => {
+ frm.events.show_schedule_dialog(frm);
+ });
+ }
+
let has_create_buttons = false;
if (frm.doc.status !== "Completed") {
@@ -248,6 +263,324 @@ frappe.ui.form.on("Production Plan", {
set_field_options("projected_qty_formula", projected_qty_formula);
},
+ show_schedule_dialog(frm) {
+ let items_data = frm.doc.po_items.map((row) => ({
+ plan_row: row.name,
+ item_code: row.item_code,
+ planned_qty: row.planned_qty,
+ start_date: row.planned_start_date,
+ }));
+
+ let dialog = new frappe.ui.Dialog({
+ title: __("Schedule Production Plan"),
+ size: "large",
+ fields: [
+ {
+ label: __("Start Date"),
+ fieldname: "start_date",
+ fieldtype: "Datetime",
+ reqd: 1,
+ default: frappe.datetime.now_datetime(),
+ },
+ {
+ label: __("Use Item Wise Start Dates"),
+ fieldname: "use_item_dates",
+ fieldtype: "Check",
+ default: 0,
+ description: __(
+ "Set a start date per assembly item below; its sub-assemblies are scheduled from the same date. The Start Date above is the earliest limit. Rows with a date here keep it as entered; clear a date to let the system schedule that item freely and write back the computed start."
+ ),
+ },
+ {
+ label: __("Item Wise Start Dates"),
+ fieldname: "items",
+ fieldtype: "Table",
+ depends_on: "eval:doc.use_item_dates",
+ cannot_add_rows: true,
+ cannot_delete_rows: true,
+ in_place_edit: true,
+ data: items_data,
+ get_data: () => items_data,
+ fields: [
+ {
+ fieldname: "plan_row",
+ fieldtype: "Data",
+ hidden: 1,
+ },
+ {
+ label: __("Item"),
+ fieldname: "item_code",
+ fieldtype: "Link",
+ options: "Item",
+ in_list_view: 1,
+ read_only: 1,
+ columns: 3,
+ },
+ {
+ label: __("Planned Qty"),
+ fieldname: "planned_qty",
+ fieldtype: "Float",
+ in_list_view: 1,
+ read_only: 1,
+ columns: 2,
+ },
+ {
+ label: __("Start Date"),
+ fieldname: "start_date",
+ fieldtype: "Datetime",
+ in_list_view: 1,
+ columns: 4,
+ },
+ ],
+ },
+ ],
+ primary_action_label: __("Preview"),
+ primary_action: (values) => {
+ dialog.hide();
+ frm.events.fetch_schedule_preview(frm, frm.events.get_schedule_args(frm, values));
+ },
+ });
+
+ dialog.show();
+ },
+
+ get_schedule_args(frm, values) {
+ let item_dates = {};
+ if (values.use_item_dates) {
+ (values.items || []).forEach((row) => {
+ if (row.plan_row && row.start_date) {
+ item_dates[row.plan_row] = row.start_date;
+ }
+ });
+ }
+
+ return {
+ production_plan: frm.doc.name,
+ start_date: values.start_date,
+ use_item_dates: values.use_item_dates,
+ item_dates: item_dates,
+ };
+ },
+
+ fetch_schedule_preview(frm, args) {
+ frappe.call({
+ method: "erpnext.manufacturing.scheduling.plan_adapter.get_schedule_preview",
+ type: "GET",
+ args: args,
+ freeze: true,
+ freeze_message: __("Calculating Schedule..."),
+ callback: (r) => {
+ if (!r.exc) {
+ frm.events.show_schedule_preview(frm, args, r.message);
+ }
+ },
+ });
+ },
+
+ show_schedule_preview(frm, values, proposal) {
+ let ordered = frm.events.get_ordered_preview_rows(proposal);
+ let rows_html = ordered.map((row) => frm.events.get_preview_row_html(row)).join("");
+
+ let unscheduled = Object.entries(proposal.unscheduled || {});
+ let warning = unscheduled.length
+ ? `${__(
+ "Could not schedule {0} task(s), so this proposal cannot be applied",
+ [unscheduled.length]
+ )}: ${unscheduled
+ .map(([key, reason]) => `${frappe.utils.escape_html(key)} (${reason})`)
+ .join(", ")}
`
+ : "";
+
+ let locked_note = proposal.orders_exist
+ ? `${__(
+ "Work Orders / Purchase Orders already exist against this plan, so the schedule is locked. Cancel them to re-schedule."
+ )}
`
+ : "";
+
+ let dialog_options = {
+ title: __("Schedule Preview"),
+ size: "extra-large",
+ };
+
+ if (!unscheduled.length && !proposal.orders_exist) {
+ dialog_options.primary_action_label = __("Apply Schedule");
+ dialog_options.primary_action = () => {
+ dialog.hide();
+ frm.events.apply_schedule(frm, values);
+ };
+ }
+
+ let dialog = new frappe.ui.Dialog(dialog_options);
+
+ dialog.$body.html(`
+ ${frm.events.get_preview_styles()}
+ ${frm.events.get_preview_summary_html(proposal, ordered)}
+ ${locked_note}
+ ${warning}
+
+
+
+ | ${__("Item")} |
+ ${__("Workstations")} |
+ ${__("Start")} |
+ ${__("End")} |
+ ${__("Starts In")} |
+
+ ${rows_html}
+
+
+ | ${item} |
+ ${detail} |
+ ${format_schedule_date(row.start)} |
+ ${format_schedule_date(row.end)} |
+
+ ${starts_in.label}
+ |
+
`;
+ },
+
+ get_preview_summary_html(proposal, ordered) {
+ let starts = ordered.map((row) => row.start).sort();
+ let total = starts.length ? schedule_duration_label(starts[0], proposal.completion_date) : "-";
+
+ return `
+
+
${__("Expected Completion")}
+
${format_schedule_date(proposal.completion_date)}
+
+
+
${__("Total Duration")}
+
${total}
+
+
+
${__("Items")}
+
${Object.keys(proposal.rows || {}).length}
+
+
".join(f"{key}: {reason}" for key, reason in unscheduled.items())
+ frappe.throw(
+ _("Cannot apply an incomplete schedule. {0} task(s) could not be placed:
{1}").format(
+ len(unscheduled), reasons
+ )
+ )
+
+
+def validate_plan_for_scheduling(plan):
+ if plan.docstatus == 2:
+ frappe.throw(_("Cannot schedule a cancelled Production Plan"))
+
+ if plan.status in ("Completed", "Closed"):
+ frappe.throw(_("Cannot schedule a Production Plan with status {0}").format(_(plan.status)))
+
+ if has_orders_against_plan(plan):
+ frappe.throw(
+ _(
+ "Work Orders / Purchase Orders have already been created against this Production Plan. Cancel them before re-scheduling."
+ )
+ )
+
+
+def has_orders_against_plan(plan):
+ if frappe.db.exists("Work Order", {"production_plan": plan.name, "docstatus": ("<", 2)}):
+ return True
+
+ return bool(
+ frappe.db.exists("Purchase Order Item", {"production_plan": plan.name, "docstatus": ("<", 2)})
+ )
+
+
+def run_engine(plan, start_date, use_item_dates=0, item_dates=None):
+ tasks, task_info = build_plan_tasks(plan, use_item_dates, item_dates)
+ 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 {}
+ engine = SchedulingEngine(
+ resources,
+ existing_load=load,
+ gap_mins=cint(settings.mins_between_operations) or 10,
+ horizon_days=cint(settings.capacity_planning_for_days) or 365,
+ )
+
+ result = engine.schedule(tasks, anchor=start_date)
+ return build_proposal(plan, result, task_info)
+
+
+def build_plan_tasks(plan, use_item_dates=0, item_dates=None):
+ ctx = get_build_context(plan)
+ tasks, task_info = ctx.tasks, ctx.task_info
+
+ for fg_row in plan.po_items:
+ sub_rows = [row for row in plan.sub_assembly_items if row.production_plan_item == fg_row.name]
+ row_bounds = build_sub_assembly_tasks(plan, sub_rows, ctx)
+ fg_deps = get_finished_good_dependencies(fg_row, sub_rows, row_bounds)
+
+ fg_tasks, first_keys, _terminal = build_row_tasks(
+ plan, fg_row.bom_no, fg_row.item_code, flt(fg_row.planned_qty), False, fg_row.name, ctx.lead_times
+ )
+ wire_dependencies(fg_tasks, fg_deps)
+ wire_material_dependencies(
+ fg_row.bom_no, get_first_tasks((first_keys, None, fg_tasks)), ctx, consumer_row=fg_row.name
+ )
+ register_tasks(fg_tasks, fg_row.name, "Finished Good", fg_row.item_code, tasks, task_info)
+
+ if use_item_dates:
+ row_date = (item_dates or {}).get(fg_row.name)
+ if row_date:
+ set_chain_earliest_start(row_date, sub_rows, row_bounds, fg_tasks)
+
+ return tasks, task_info
+
+
+def get_build_context(plan):
+ bom_materials = get_bom_materials(plan)
+ produced_items = {row.item_code for row in plan.po_items}
+ produced_items.update(row.production_item for row in plan.sub_assembly_items)
+
+ material_items = {
+ item for items in bom_materials.values() for item in items if item not in produced_items
+ }
+ lead_times = get_lead_time_details(plan, material_items)
+
+ return frappe._dict(
+ bom_materials={
+ bom: [item for item in items if item not in produced_items]
+ for bom, items in bom_materials.items()
+ },
+ lead_times=lead_times,
+ material_lead_days=get_material_lead_days(material_items, lead_times),
+ material_tasks={},
+ tasks=[],
+ task_info={},
+ )
+
+
+def get_bom_materials(plan):
+ boms = {row.bom_no for row in plan.po_items if row.bom_no}
+ boms.update(row.bom_no for row in plan.sub_assembly_items if row.bom_no)
+ if not boms:
+ return {}
+
+ materials = defaultdict(list)
+ for row in frappe.get_all(
+ "BOM Item",
+ filters={"parent": ("in", list(boms)), "parenttype": "BOM"},
+ fields=["parent", "item_code"],
+ ):
+ materials[row.parent].append(row.item_code)
+
+ return materials
+
+
+def get_material_lead_days(material_items, lead_times):
+ missing = [item for item in material_items if item not in lead_times]
+ item_master_days = {}
+ if missing:
+ item_master_days = dict(
+ frappe.get_all(
+ "Item", filters={"name": ("in", missing)}, fields=["name", "lead_time_days"], as_list=True
+ )
+ )
+
+ lead_days = {}
+ for item in material_items:
+ lead_time = lead_times.get(item)
+ if lead_time:
+ lead_days[item] = cint(lead_time.purchase_time) + cint(lead_time.buffer_time)
+ else:
+ lead_days[item] = cint(item_master_days.get(item))
+
+ return lead_days
+
+
+def wire_material_dependencies(bom_no, first_tasks, ctx, consumer_row=None):
+ if not bom_no or not first_tasks:
+ return
+
+ dependency_keys = []
+ for item_code in ctx.bom_materials.get(bom_no, []):
+ lead_days = ctx.material_lead_days.get(item_code)
+ if lead_days:
+ dependency_keys.append(get_material_task(item_code, lead_days, ctx, consumer_row))
+
+ if dependency_keys:
+ wire_dependencies(first_tasks, dependency_keys)
+
+
+def get_material_task(item_code, lead_days, ctx, consumer_row=None):
+ key = f"material:{item_code}"
+ if key not in ctx.material_tasks:
+ task = Task(key=key, duration_mins=lead_days * 1440.0)
+ ctx.material_tasks[key] = task
+ ctx.tasks.append(task)
+ ctx.task_info[key] = {
+ "plan_row": key,
+ "row_type": "Raw Material",
+ "item_code": item_code,
+ "operation": None,
+ "parent_row": None,
+ "consumers": [],
+ }
+
+ if consumer_row and consumer_row not in ctx.task_info[key]["consumers"]:
+ ctx.task_info[key]["consumers"].append(consumer_row)
+
+ return key
+
+
+def set_chain_earliest_start(row_date, sub_rows, row_bounds, fg_tasks):
+ earliest_start = get_datetime(row_date)
+ chain_tasks = list(fg_tasks)
+ for row in sub_rows:
+ chain_tasks.extend(row_bounds[row.name][2])
+
+ for task in chain_tasks:
+ task.earliest_start = earliest_start
+
+
+def build_sub_assembly_tasks(plan, sub_rows, ctx):
+ row_bounds = {}
+ for row in sub_rows:
+ subcontracted = row.type_of_manufacturing == "Subcontract"
+ row_tasks, first_keys, terminal_keys = build_row_tasks(
+ plan, row.bom_no, row.production_item, flt(row.qty), subcontracted, row.name, ctx.lead_times
+ )
+ row_bounds[row.name] = (first_keys, terminal_keys, row_tasks)
+ if not subcontracted:
+ wire_material_dependencies(
+ row.bom_no, get_first_tasks(row_bounds[row.name]), ctx, consumer_row=row.name
+ )
+
+ rows_by_item = defaultdict(list)
+ for row in sub_rows:
+ rows_by_item[row.production_item].append(row)
+
+ for row in sub_rows:
+ parents = rows_by_item.get(row.parent_item_code) or []
+ if parents:
+ parent_first_tasks = get_first_tasks(row_bounds[parents[0].name])
+ wire_dependencies(parent_first_tasks, row_bounds[row.name][1])
+
+ for row in sub_rows:
+ register_tasks(
+ row_bounds[row.name][2],
+ row.name,
+ "Sub Assembly",
+ row.production_item,
+ ctx.tasks,
+ ctx.task_info,
+ parent_row=row.production_plan_item,
+ )
+
+ return row_bounds
+
+
+def get_first_tasks(bounds):
+ first_keys, _terminal_keys, row_tasks = bounds
+ return [task for task in row_tasks if task.key in first_keys]
+
+
+def get_finished_good_dependencies(fg_row, sub_rows, row_bounds):
+ produced_items = {row.production_item for row in sub_rows}
+ dependencies = []
+ for row in sub_rows:
+ if row.parent_item_code == fg_row.item_code or row.parent_item_code not in produced_items:
+ dependencies.extend(row_bounds[row.name][1])
+
+ return dependencies
+
+
+def wire_dependencies(first_tasks, dependency_keys):
+ for task in first_tasks:
+ task.depends_on = list(dict.fromkeys([*task.depends_on, *dependency_keys]))
+
+
+def register_tasks(row_tasks, row_name, row_type, item_code, tasks, task_info, parent_row=None):
+ for task in row_tasks:
+ tasks.append(task)
+ task_info[task.key] = {
+ "plan_row": row_name,
+ "row_type": row_type,
+ "item_code": item_code,
+ "operation": task.label,
+ "parent_row": parent_row,
+ }
+
+
+def build_row_tasks(plan, bom_no, item_code, qty, subcontracted, prefix, lead_times):
+ if not subcontracted and bom_no and frappe.get_cached_value("BOM", bom_no, "with_operations"):
+ row_tasks, terminal_keys = loaders.build_bom_operation_tasks(bom_no, qty, prefix)
+ if row_tasks:
+ first_keys = [task.key for task in row_tasks if not task.depends_on]
+ return row_tasks, first_keys, terminal_keys
+
+ duration = get_lead_time_duration_mins(
+ lead_times.get(item_code), qty, subcontracted, cint(plan.get("no_of_shifts"))
+ )
+ task = Task(key=prefix, duration_mins=duration)
+ return [task], [task.key], [task.key]
+
+
+def get_lead_time_duration_mins(lead_time, qty, subcontracted, no_of_shifts):
+ if not lead_time:
+ return 1440.0
+
+ if subcontracted:
+ return max(cint(lead_time.purchase_time) + cint(lead_time.buffer_time), 1) * 1440.0
+
+ days = 0
+ capacity = get_daily_capacity(lead_time, no_of_shifts)
+ if capacity:
+ days = math.ceil(qty / capacity)
+ elif lead_time.manufacturing_time_in_mins:
+ minutes_per_day = (
+ (no_of_shifts or cint(lead_time.no_of_shift) or 1)
+ * (cint(lead_time.shift_time_in_hours) or 8)
+ * 60
+ * (cint(lead_time.no_of_workstations) or 1)
+ )
+ days = math.ceil(cint(lead_time.manufacturing_time_in_mins) * qty / minutes_per_day)
+
+ return max(days + cint(lead_time.buffer_time), 1) * 1440.0
+
+
+def get_daily_capacity(lead_time, no_of_shifts):
+ capacity = flt(lead_time.capacity_per_day)
+ if capacity and lead_time.daily_yield:
+ capacity = capacity * flt(lead_time.daily_yield) / 100
+
+ if capacity and no_of_shifts:
+ capacity = capacity * no_of_shifts / (cint(lead_time.no_of_shift) or 1)
+
+ return capacity
+
+
+def get_lead_time_details(plan, extra_items=None):
+ item_codes = {row.item_code for row in plan.po_items}
+ item_codes.update(row.production_item for row in plan.sub_assembly_items)
+ item_codes.update(extra_items or [])
+
+ return {
+ row.item_code: row
+ for row in frappe.get_all(
+ "Item Lead Time",
+ filters={"item_code": ("in", list(item_codes))},
+ fields=[
+ "item_code",
+ "capacity_per_day",
+ "daily_yield",
+ "manufacturing_time_in_mins",
+ "no_of_shift",
+ "shift_time_in_hours",
+ "no_of_workstations",
+ "purchase_time",
+ "buffer_time",
+ ],
+ )
+ }
+
+
+def build_proposal(plan, result, task_info):
+ rows = defaultdict(lambda: {"blocks": []})
+ for key, assignment in result.assignments.items():
+ info = task_info[key]
+ row = rows[info["plan_row"]]
+ row.update(
+ {
+ "row_type": info["row_type"],
+ "item_code": info["item_code"],
+ "parent_row": info.get("parent_row"),
+ "consumers": info.get("consumers") or [],
+ }
+ )
+ for block in assignment.blocks:
+ row["blocks"].append(
+ {
+ "task_key": key,
+ "operation": info["operation"],
+ "workstation": assignment.resource,
+ "from_time": block.start,
+ "to_time": block.end,
+ "duration_mins": block.duration_mins(),
+ }
+ )
+
+ for row in rows.values():
+ row["blocks"].sort(key=lambda block: block["from_time"])
+ row["start"] = row["blocks"][0]["from_time"]
+ row["end"] = row["blocks"][-1]["to_time"]
+
+ return {
+ "production_plan": plan.name,
+ "direction_used": result.direction_used,
+ "completion_date": result.end_date,
+ "rows": dict(rows),
+ "unscheduled": {key: reason for key, reason in result.unscheduled.items()},
+ }
+
+
+def replace_schedule_entries(plan, proposal):
+ frappe.db.delete("Production Plan Schedule", {"production_plan": plan.name})
+
+ for row_name, row in proposal["rows"].items():
+ for block in row["blocks"]:
+ entry = make_schedule_entry(plan, row_name, row, block)
+ entry.flags.from_scheduler = True
+ entry.insert(ignore_permissions=True)
+
+
+def make_schedule_entry(plan, row_name, row, block):
+ return frappe.get_doc(
+ {
+ "doctype": "Production Plan Schedule",
+ "production_plan": plan.name,
+ "company": plan.company,
+ "plan_row": row_name,
+ "row_type": row["row_type"],
+ "item_code": row["item_code"],
+ "operation": block.get("operation") if block.get("workstation") else None,
+ "workstation": block.get("workstation"),
+ "from_time": block["from_time"],
+ "to_time": block["to_time"],
+ "duration_mins": block["duration_mins"],
+ "task_key": block["task_key"],
+ "subject": get_entry_subject(row, block),
+ }
+ )
+
+
+def get_entry_subject(row, block):
+ item_name = frappe.get_cached_value("Item", row["item_code"], "item_name") or row["item_code"]
+
+ if row["row_type"] == "Raw Material":
+ activity = _("Procurement")
+ elif block.get("workstation") and block.get("operation"):
+ activity = block["operation"]
+ else:
+ activity = _("Production")
+
+ return f"{item_name} · {activity}"
+
+
+def update_plan_row_dates(plan, proposal, use_item_dates=0, item_dates=None):
+ rows = proposal["rows"]
+ for fg_row in plan.po_items:
+ if fg_row.name in rows:
+ set_finished_good_start_date(fg_row, rows, use_item_dates, item_dates)
+ fg_row.db_set("planned_end_date", rows[fg_row.name]["end"], update_modified=False)
+
+ for row in plan.sub_assembly_items:
+ if row.name in rows:
+ row.db_set("schedule_date", rows[row.name]["start"], update_modified=False)
+ row.db_set("schedule_end_date", rows[row.name]["end"], update_modified=False)
+
+
+def set_finished_good_start_date(fg_row, rows, use_item_dates, item_dates):
+ if use_item_dates and (item_dates or {}).get(fg_row.name):
+ fg_row.db_set("planned_start_date", item_dates[fg_row.name], update_modified=False)
+ else:
+ fg_row.db_set("planned_start_date", rows[fg_row.name]["start"], update_modified=False)
diff --git a/erpnext/manufacturing/scheduling/test_plan_adapter.py b/erpnext/manufacturing/scheduling/test_plan_adapter.py
new file mode 100644
index 00000000000..6c18d1637c3
--- /dev/null
+++ b/erpnext/manufacturing/scheduling/test_plan_adapter.py
@@ -0,0 +1,264 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# See license.txt
+
+import frappe
+from frappe.tests.utils import change_settings
+from frappe.utils import add_to_date, get_datetime
+
+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.plan_adapter import apply_schedule, get_schedule_preview
+from erpnext.stock.doctype.item.test_item import create_item
+from erpnext.tests.utils import ERPNextTestSuite
+
+
+class TestPlanAdapter(ERPNextTestSuite):
+ def setUp(self):
+ super().setUp()
+ self.workstation = "Test PPS WS"
+ if not frappe.db.exists("Workstation", self.workstation):
+ make_workstation(workstation=self.workstation, production_capacity=1)
+
+ self.operation = "Test PPS Op"
+ if not frappe.db.exists("Operation", self.operation):
+ make_operation(operation=self.operation, workstation=self.workstation)
+
+ for item in ["Test PPS FG", "Test PPS FG 2", "Test PPS SA 1", "Test PPS SA 2", "Test PPS RM"]:
+ create_item(item, valuation_rate=100)
+
+ for sub_assembly in ["Test PPS SA 1", "Test PPS SA 2"]:
+ self.make_bom_with_operation(sub_assembly, ["Test PPS RM"], time_in_mins=60)
+
+ 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):
+ 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,
+ },
+ )
+ bom.insert(ignore_permissions=True)
+ bom.submit()
+
+ def make_plan(self):
+ plan = create_production_plan(
+ item_code="Test PPS FG",
+ planned_qty=2,
+ use_multi_level_bom=1,
+ do_not_submit=True,
+ skip_getting_mr_items=True,
+ )
+ plan.get_sub_assembly_items()
+ plan.submit()
+ return plan
+
+ @change_settings(
+ "Manufacturing Settings",
+ {"mins_between_operations": 10, "allow_overtime": 0, "disable_capacity_planning": 0},
+ )
+ def test_job_cards_match_plan_schedule(self):
+ plan = self.make_plan()
+ start_date = get_datetime("2026-10-01 09:00:00")
+ apply_schedule(plan.name, start_date)
+
+ plan.reload()
+ plan.make_work_order()
+ work_orders = frappe.get_all("Work Order", filters={"production_plan": plan.name}, pluck="name")
+ self.assertEqual(len(work_orders), 3)
+
+ for name in work_orders:
+ work_order = frappe.get_doc("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()
+ self.assert_job_card_matches_schedule(plan, work_order)
+
+ self.assertRaises(frappe.ValidationError, apply_schedule, plan.name, start_date)
+
+ def assert_job_card_matches_schedule(self, plan, work_order):
+ plan_row = work_order.production_plan_item or work_order.production_plan_sub_assembly_item
+ entries = frappe.get_all(
+ "Production Plan Schedule",
+ filters={"production_plan": plan.name, "plan_row": plan_row},
+ fields=["from_time", "to_time", "workstation"],
+ order_by="from_time",
+ )
+ self.assertTrue(entries)
+
+ job_card = frappe.get_doc("Job Card", {"work_order": work_order.name})
+ self.assertEqual(len(job_card.scheduled_time_logs), len(entries))
+ for log, entry in zip(job_card.scheduled_time_logs, entries, strict=True):
+ self.assertEqual(get_datetime(log.from_time), get_datetime(entry.from_time))
+ self.assertEqual(get_datetime(log.to_time), get_datetime(entry.to_time))
+
+ self.assertEqual(job_card.workstation, entries[0].workstation)
+ self.assertEqual(get_datetime(job_card.expected_start_date), get_datetime(entries[0].from_time))
+ self.assertEqual(get_datetime(job_card.expected_end_date), get_datetime(entries[-1].to_time))
+
+ operation_row = work_order.operations[0]
+ self.assertEqual(get_datetime(operation_row.planned_start_time), get_datetime(entries[0].from_time))
+ self.assertEqual(get_datetime(operation_row.planned_end_time), get_datetime(entries[-1].to_time))
+
+ @change_settings("Manufacturing Settings", {"mins_between_operations": 10, "allow_overtime": 0})
+ def test_item_wise_start_dates(self):
+ day_one = get_datetime("2026-09-01 09:00:00")
+ day_two = get_datetime("2026-09-02 09:00:00")
+
+ plan = create_production_plan(
+ item_code="Test PPS FG",
+ planned_qty=2,
+ planned_start_date=day_one,
+ use_multi_level_bom=1,
+ do_not_submit=True,
+ skip_getting_mr_items=True,
+ )
+ plan.append(
+ "po_items",
+ {
+ "use_multi_level_bom": 1,
+ "item_code": "Test PPS FG 2",
+ "bom_no": frappe.db.get_value("Item", "Test PPS FG 2", "default_bom"),
+ "planned_qty": 2,
+ "planned_start_date": day_one,
+ "stock_uom": "Nos",
+ "warehouse": plan.po_items[0].warehouse,
+ },
+ )
+ plan.get_sub_assembly_items()
+ plan.submit()
+
+ apply_schedule(plan.name, day_one, use_item_dates=1, item_dates={plan.po_items[1].name: str(day_two)})
+ plan.reload()
+
+ fg_one, fg_two = plan.po_items
+ self.assertEqual(get_datetime(fg_two.planned_start_date), day_two)
+ self.assertEqual(get_datetime(fg_two.planned_end_date), add_to_date(day_two, minutes=60))
+ # no dialog date for the first row, so its computed start (after both
+ # sub-assemblies and the operation gap) is persisted to match the calendar
+ self.assertEqual(get_datetime(fg_one.planned_start_date), add_to_date(day_one, minutes=250))
+ self.assertEqual(get_datetime(fg_one.planned_end_date), add_to_date(day_one, minutes=310))
+
+ for row in plan.sub_assembly_items:
+ self.assertGreaterEqual(get_datetime(row.schedule_date), day_one)
+
+ @change_settings("Manufacturing Settings", {"mins_between_operations": 10, "allow_overtime": 0})
+ def test_cleared_item_date_frees_the_chain(self):
+ day_one = get_datetime("2026-12-01 09:00:00")
+ day_two = get_datetime("2026-12-02 09:00:00")
+
+ plan = self.make_plan()
+ fg_row = plan.po_items[0].name
+
+ apply_schedule(plan.name, day_one, use_item_dates=1, item_dates={fg_row: str(day_two)})
+ plan.reload()
+ self.assertEqual(get_datetime(plan.po_items[0].planned_start_date), day_two)
+
+ apply_schedule(plan.name, day_one, use_item_dates=1, item_dates={})
+ plan.reload()
+
+ # the cleared date no longer constrains the chain, so the item schedules
+ # 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))
+
+ def test_manual_schedule_entry_creation_is_blocked(self):
+ plan = self.make_plan()
+ entry = frappe.get_doc(
+ {
+ "doctype": "Production Plan Schedule",
+ "production_plan": plan.name,
+ "item_code": "Test PPS FG",
+ "from_time": "2026-11-02 09:00:00",
+ "to_time": "2026-11-02 10:00:00",
+ }
+ )
+
+ self.assertRaises(frappe.ValidationError, entry.insert)
+
+ def test_incomplete_proposal_is_not_applied(self):
+ from unittest.mock import patch
+
+ from erpnext.manufacturing.scheduling import plan_adapter
+
+ plan = self.make_plan()
+ incomplete = {
+ "rows": {},
+ "unscheduled": {"task": "no capacity within horizon"},
+ "completion_date": None,
+ }
+
+ with patch.object(plan_adapter, "run_engine", return_value=incomplete):
+ self.assertRaises(frappe.ValidationError, apply_schedule, plan.name, "2026-11-02 09:00:00")
+
+ @change_settings("Manufacturing Settings", {"mins_between_operations": 10, "allow_overtime": 0})
+ def test_schedule_considers_raw_material_lead_time(self):
+ if frappe.db.exists("Item Lead Time", "Test PPS RM"):
+ frappe.db.set_value("Item Lead Time", "Test PPS RM", {"purchase_time": 2, "buffer_time": 0})
+ else:
+ frappe.get_doc(
+ {"doctype": "Item Lead Time", "item_code": "Test PPS RM", "purchase_time": 2}
+ ).insert()
+ self.addCleanup(frappe.delete_doc, "Item Lead Time", "Test PPS RM", force=True)
+
+ plan = self.make_plan()
+ start_date = get_datetime("2026-11-02 09:00:00")
+ preview = get_schedule_preview(plan.name, start_date)
+
+ material_row = preview["rows"].get("material:Test PPS RM")
+ self.assertIsNotNone(material_row)
+ self.assertEqual(material_row["start"], start_date)
+ self.assertEqual(material_row["end"], add_to_date(start_date, days=2))
+
+ material_arrival = add_to_date(start_date, days=2, minutes=10)
+ sub_starts = sorted(
+ row["start"] for row in preview["rows"].values() if row["row_type"] == "Sub Assembly"
+ )
+ self.assertEqual(sub_starts, [material_arrival, add_to_date(material_arrival, minutes=120)])
+
+ @change_settings("Manufacturing Settings", {"mins_between_operations": 10, "allow_overtime": 0})
+ def test_preview_and_apply_schedule(self):
+ plan = self.make_plan()
+ start_date = get_datetime("2026-08-13 09:00:00")
+
+ preview = get_schedule_preview(plan.name, start_date)
+ self.assertEqual(len(preview["rows"]), 3)
+ self.assertFalse(preview["unscheduled"])
+
+ sub_starts = sorted(
+ row["start"] for row in preview["rows"].values() if row["row_type"] == "Sub Assembly"
+ )
+ self.assertEqual(sub_starts, [start_date, add_to_date(start_date, minutes=120)])
+
+ fg_row = next(row for row in preview["rows"].values() if row["row_type"] == "Finished Good")
+ self.assertEqual(fg_row["start"], add_to_date(start_date, minutes=250))
+ self.assertEqual(preview["completion_date"], add_to_date(start_date, minutes=310))
+
+ self.assertFalse(frappe.db.exists("Production Plan Schedule", {"production_plan": plan.name}))
+
+ apply_schedule(plan.name, start_date)
+
+ entries = frappe.get_all(
+ "Production Plan Schedule",
+ filters={"production_plan": plan.name},
+ fields=["workstation", "from_time", "to_time", "row_type", "operation"],
+ )
+ self.assertEqual(len(entries), 3)
+ self.assertTrue(all(entry.workstation == self.workstation for entry in entries))
+
+ plan.reload()
+ self.assertEqual(get_datetime(plan.po_items[0].planned_start_date), fg_row["start"])
+ self.assertEqual(get_datetime(plan.po_items[0].planned_end_date), preview["completion_date"])
+ for row in plan.sub_assembly_items:
+ self.assertIn(get_datetime(row.schedule_date), sub_starts)
+ self.assertIsNotNone(row.schedule_end_date)
diff --git a/erpnext/manufacturing/scheduling/test_scheduling_engine.py b/erpnext/manufacturing/scheduling/test_scheduling_engine.py
new file mode 100644
index 00000000000..63cc805fa0e
--- /dev/null
+++ b/erpnext/manufacturing/scheduling/test_scheduling_engine.py
@@ -0,0 +1,184 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# See license.txt
+
+import datetime
+import unittest
+
+from erpnext.manufacturing.scheduling.engine import SchedulingEngine
+from erpnext.manufacturing.scheduling.models import (
+ BACKWARD,
+ FORWARD,
+ INFINITE,
+ Interval,
+ Resource,
+ ResourceCalendar,
+ Task,
+)
+
+
+def dt(day, hour, minute=0):
+ return datetime.datetime(2026, 8, day, hour, minute)
+
+
+def day_shift_calendar():
+ return ResourceCalendar(daily_windows=[(datetime.time(9, 0), datetime.time(17, 0))])
+
+
+class TestSchedulingEngine(unittest.TestCase):
+ def test_forward_places_task_within_working_window(self):
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar())])
+ result = engine.schedule([Task("t1", duration_mins=120, resource="WS-A")], anchor=dt(12, 10))
+
+ assignment = result.assignments["t1"]
+ self.assertEqual(assignment.start, dt(12, 10))
+ self.assertEqual(assignment.end, dt(12, 12))
+
+ def test_task_splits_across_days_and_skips_holiday(self):
+ calendar = day_shift_calendar()
+ calendar.holidays.add(datetime.date(2026, 8, 13))
+ engine = SchedulingEngine([Resource("WS-A", calendar=calendar)])
+
+ result = engine.schedule([Task("t1", duration_mins=600, resource="WS-A")], anchor=dt(12, 15))
+
+ blocks = result.assignments["t1"].blocks
+ self.assertEqual(blocks[0], Interval(dt(12, 15), dt(12, 17)))
+ self.assertEqual(blocks[1], Interval(dt(14, 9), dt(14, 17)))
+ self.assertEqual(result.assignments["t1"].end, dt(14, 17))
+
+ def test_finite_capacity_serializes_and_capacity_two_runs_parallel(self):
+ tasks = [
+ Task("t1", duration_mins=120, resource="WS-A"),
+ Task("t2", duration_mins=120, resource="WS-A"),
+ ]
+
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar(), capacity=1)])
+ result = engine.schedule([*tasks], anchor=dt(12, 9))
+ self.assertEqual(result.assignments["t1"].start, dt(12, 9))
+ self.assertEqual(result.assignments["t2"].start, dt(12, 11))
+
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar(), capacity=2)])
+ result = engine.schedule([*tasks], anchor=dt(12, 9))
+ self.assertEqual(result.assignments["t2"].start, dt(12, 9))
+
+ def test_existing_load_pushes_task_and_infinite_mode_ignores_it(self):
+ booked = {"WS-A": [Interval(dt(12, 9), dt(12, 12))]}
+
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar())], existing_load=booked)
+ result = engine.schedule([Task("t1", duration_mins=60, resource="WS-A")], anchor=dt(12, 9))
+ self.assertEqual(result.assignments["t1"].start, dt(12, 12))
+
+ engine = SchedulingEngine(
+ [Resource("WS-A", calendar=day_shift_calendar())], existing_load=booked, mode=INFINITE
+ )
+ result = engine.schedule([Task("t1", duration_mins=60, resource="WS-A")], anchor=dt(12, 9))
+ self.assertEqual(result.assignments["t1"].start, dt(12, 9))
+
+ def test_dependency_chain_applies_gap(self):
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar())], gap_mins=10)
+ tasks = [
+ Task("t1", duration_mins=60, resource="WS-A"),
+ Task("t2", duration_mins=60, resource="WS-A", depends_on=["t1"]),
+ ]
+
+ result = engine.schedule(tasks, anchor=dt(12, 9))
+ self.assertEqual(result.assignments["t2"].start, dt(12, 10, 10))
+
+ def test_capability_selection_picks_free_machine(self):
+ resources = [
+ Resource("WS-A", calendar=day_shift_calendar(), resource_type="CNC"),
+ Resource("WS-B", calendar=day_shift_calendar(), resource_type="CNC"),
+ ]
+ engine = SchedulingEngine(resources, existing_load={"WS-A": [Interval(dt(12, 9), dt(12, 13))]})
+
+ result = engine.schedule([Task("t1", duration_mins=60, resource_type="CNC")], anchor=dt(12, 9))
+ self.assertEqual(result.assignments["t1"].resource, "WS-B")
+ self.assertEqual(result.assignments["t1"].start, dt(12, 9))
+
+ def test_four_jobs_run_concurrently_on_two_capacity_two_machines(self):
+ resources = [
+ Resource("Mold-A", calendar=day_shift_calendar(), capacity=2, resource_type="Molding"),
+ Resource("Mold-B", calendar=day_shift_calendar(), capacity=2, resource_type="Molding"),
+ ]
+ engine = SchedulingEngine(resources)
+ tasks = [Task(f"t{i}", duration_mins=120, resource_type="Molding") for i in range(4)]
+
+ result = engine.schedule(tasks, anchor=dt(12, 9))
+
+ self.assertEqual([a.start for a in result.assignments.values()], [dt(12, 9)] * 4)
+ machines = sorted(a.resource for a in result.assignments.values())
+ self.assertEqual(machines, ["Mold-A", "Mold-A", "Mold-B", "Mold-B"])
+
+ overflow = engine.schedule([Task("t5", duration_mins=120, resource_type="Molding")], anchor=dt(12, 9))
+ self.assertEqual(overflow.assignments["t5"].start, dt(12, 11))
+
+ def test_priority_wins_contention(self):
+ tasks = [
+ Task("low", duration_mins=120, resource="WS-A", priority=1),
+ Task("high", duration_mins=120, resource="WS-A", priority=10),
+ ]
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar())])
+
+ result = engine.schedule(tasks, anchor=dt(12, 9))
+ self.assertEqual(result.assignments["high"].start, dt(12, 9))
+ self.assertEqual(result.assignments["low"].start, dt(12, 11))
+
+ def test_backward_scheduling_meets_due_date(self):
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar())], gap_mins=10)
+ tasks = [
+ Task("t1", duration_mins=60, resource="WS-A"),
+ Task("t2", duration_mins=120, resource="WS-A", depends_on=["t1"]),
+ ]
+
+ result = engine.schedule(tasks, anchor=dt(14, 17), direction=BACKWARD, not_before=dt(12, 9))
+
+ self.assertEqual(result.direction_used, BACKWARD)
+ self.assertEqual(result.assignments["t2"].end, dt(14, 17))
+ self.assertEqual(result.assignments["t2"].start, dt(14, 15))
+ self.assertEqual(result.assignments["t1"].end, dt(14, 14, 50))
+ self.assertEqual(result.assignments["t1"].start, dt(14, 13, 50))
+
+ def test_backward_falls_back_forward_when_due_date_infeasible(self):
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar())])
+ tasks = [Task("t1", duration_mins=480, resource="WS-A")]
+
+ result = engine.schedule(tasks, anchor=dt(12, 11), direction=BACKWARD, not_before=dt(12, 9))
+
+ self.assertEqual(result.direction_used, FORWARD)
+ self.assertEqual(result.assignments["t1"].start, dt(12, 9))
+ self.assertEqual(result.assignments["t1"].end, dt(12, 17))
+
+ def test_calendarless_task_runs_continuously(self):
+ engine = SchedulingEngine([])
+ result = engine.schedule([Task("buy", duration_mins=2880)], anchor=dt(12, 8))
+
+ self.assertIsNone(result.assignments["buy"].resource)
+ self.assertEqual(result.assignments["buy"].end, dt(14, 8))
+
+ def test_cycle_and_missing_capacity_reported_not_raised(self):
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar())], horizon_days=1)
+ tasks = [
+ Task("a", duration_mins=60, resource="WS-A", depends_on=["b"]),
+ Task("b", duration_mins=60, resource="WS-A", depends_on=["a"]),
+ Task("c", duration_mins=600, resource="WS-A"),
+ ]
+
+ result = engine.schedule(tasks, anchor=dt(12, 9))
+
+ self.assertIn("a", result.unscheduled)
+ self.assertIn("b", result.unscheduled)
+ self.assertEqual(result.unscheduled["c"], "no capacity within horizon")
+
+ def test_run_holds_multiple_documents_without_blind_spots(self):
+ engine = SchedulingEngine([Resource("WS-A", calendar=day_shift_calendar())])
+ plan_a = [Task("planA:op", duration_mins=240, resource="WS-A")]
+ plan_b = [Task("planB:op", duration_mins=240, resource="WS-A")]
+
+ result_a = engine.schedule(plan_a, anchor=dt(12, 9))
+ result_b = engine.schedule(plan_b, anchor=dt(12, 9))
+
+ self.assertEqual(result_a.assignments["planA:op"].start, dt(12, 9))
+ self.assertEqual(result_b.assignments["planB:op"].start, dt(12, 13))
+
+
+if __name__ == "__main__":
+ unittest.main()