From 4f652be10dbbacae455007596fc47b41395a4034 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Mon, 7 Sep 2026 19:36:44 +0530 Subject: [PATCH] fix(manufacturing): share transfer stock across Production Plan rows (v15) --- .../production_plan/production_plan.py | 96 +++++++++----- .../production_plan/test_production_plan.py | 119 ++++++++++++++++++ 2 files changed, 182 insertions(+), 33 deletions(-) diff --git a/erpnext/manufacturing/doctype/production_plan/production_plan.py b/erpnext/manufacturing/doctype/production_plan/production_plan.py index 90e791fc9ba..7ab49dacac7 100644 --- a/erpnext/manufacturing/doctype/production_plan/production_plan.py +++ b/erpnext/manufacturing/doctype/production_plan/production_plan.py @@ -1543,9 +1543,10 @@ def get_warehouse_list(warehouses): @frappe.whitelist() -def get_items_for_material_requests(doc, warehouses=None, get_parent_warehouse_data=None): - if isinstance(doc, str): - doc = frappe._dict(json.loads(doc)) +def get_items_for_material_requests( + doc: str | dict, warehouses: str | list[dict] | None = None, get_parent_warehouse_data: bool | None = None +): + doc = frappe._dict(json.loads(doc) if isinstance(doc, str) else doc) if warehouses: warehouses = list(set(get_warehouse_list(warehouses))) @@ -1720,6 +1721,7 @@ def get_items_for_material_requests(doc, warehouses=None, get_parent_warehouse_d if (not ignore_existing_ordered_qty or get_parent_warehouse_data) and warehouses: new_mr_items = [] + locations_by_item = _get_transfer_locations(mr_items, warehouses, company) for item in mr_items: get_materials_from_other_locations( item, @@ -1727,6 +1729,7 @@ def get_items_for_material_requests(doc, warehouses=None, get_parent_warehouse_d new_mr_items, company, consider_minimum_order_qty=doc.get("consider_minimum_order_qty"), + locations=locations_by_item[item.get("item_code")], ) mr_items = new_mr_items @@ -1748,47 +1751,21 @@ def get_items_for_material_requests(doc, warehouses=None, get_parent_warehouse_d def get_materials_from_other_locations( - item, warehouses, new_mr_items, company, consider_minimum_order_qty=False + item, warehouses, new_mr_items, company, consider_minimum_order_qty=False, locations=None ): - from erpnext.stock.doctype.pick_list.pick_list import get_available_item_locations - stock_uom, purchase_uom = frappe.db.get_value( "Item", item.get("item_code"), ["stock_uom", "purchase_uom"] ) - locations = get_available_item_locations( - item.get("item_code"), - warehouses, - item.get("quantity") * item.get("conversion_factor"), - company, - ignore_validation=True, - ) + if locations is None: + locations = _get_transfer_locations([item], warehouses, company)[item.get("item_code")] required_qty = item.get("quantity") if item.get("conversion_factor") and item.get("purchase_uom") != item.get("stock_uom"): # Convert qty to stock UOM required_qty = required_qty * item.get("conversion_factor") - # get available material by transferring to production warehouse - for d in locations: - if required_qty <= 0: - return - - new_dict = copy.deepcopy(item) - quantity = required_qty if d.get("qty") > required_qty else d.get("qty") - - new_dict.update( - { - "quantity": quantity, - "material_request_type": "Material Transfer", - "uom": new_dict.get("stock_uom"), # internal transfer should be in stock UOM - "from_warehouse": d.get("warehouse"), - "conversion_factor": 1.0, - } - ) - - required_qty -= quantity - new_mr_items.append(new_dict) + required_qty = _transfer_from_locations(item, locations, new_mr_items, required_qty) # raise purchase request for remaining qty @@ -1805,6 +1782,59 @@ def get_materials_from_other_locations( new_mr_items.append(item) +def _get_transfer_locations(mr_items, warehouses, company): + from erpnext.stock.doctype.pick_list.pick_list import get_available_item_locations + + required_qty_by_item = defaultdict(float) + for item in mr_items: + required_qty_by_item[item.get("item_code")] += max( + 0, flt(item.get("quantity")) * flt(item.get("conversion_factor")) + ) + + return { + item_code: get_available_item_locations( + item_code, warehouses, required_qty, company, ignore_validation=True + ) + if required_qty > 0 + else [] + for item_code, required_qty in required_qty_by_item.items() + } + + +def _transfer_from_locations(item, locations, new_mr_items, required_qty): + precision = frappe.get_precision("Material Request Plan Item", "quantity") + transfers_by_warehouse = {} + for d in locations: + if flt(required_qty, precision) <= 0: + return required_qty + + quantity = flt(min(required_qty, d.get("qty")), precision) + if quantity <= 0: + continue + d["qty"] -= quantity + required_qty -= quantity + + warehouse = d.get("warehouse") + if warehouse in transfers_by_warehouse: + transfer = transfers_by_warehouse[warehouse] + transfer["quantity"] = flt(transfer["quantity"] + quantity, precision) + continue + + new_dict = copy.deepcopy(item) + new_dict.update( + { + "quantity": quantity, + "material_request_type": "Material Transfer", + "uom": new_dict.get("stock_uom"), # internal transfer should be in stock UOM + "from_warehouse": warehouse, + "conversion_factor": 1.0, + } + ) + transfers_by_warehouse[warehouse] = new_dict + new_mr_items.append(new_dict) + return required_qty + + @frappe.whitelist() def get_item_data(item_code): item_details = get_item_details(item_code) diff --git a/erpnext/manufacturing/doctype/production_plan/test_production_plan.py b/erpnext/manufacturing/doctype/production_plan/test_production_plan.py index c9c266df0bb..de9fef0f65b 100644 --- a/erpnext/manufacturing/doctype/production_plan/test_production_plan.py +++ b/erpnext/manufacturing/doctype/production_plan/test_production_plan.py @@ -1562,6 +1562,125 @@ class TestProductionPlan(FrappeTestCase): self.assertFalse(items) + def _plan_for_transfer_allocation(self, rm_item, qty_per_order): + fg_item = make_item(properties={"is_stock_item": 1}).name + make_bom(item=fg_item, raw_materials=[rm_item], source_warehouse="_Test Warehouse - _TC") + + pln = create_production_plan( + item_code=fg_item, + ignore_existing_ordered_qty=0, + do_not_save=1, + skip_getting_mr_items=1, + ) + pln.get_items_from = "Sales Order" + for _ in range(2): + so = make_sales_order(item_code=fg_item, qty=qty_per_order) + pln.append( + "sales_orders", + { + "sales_order": so.name, + "sales_order_date": so.transaction_date, + "customer": so.customer, + "grand_total": so.grand_total, + }, + ) + pln.get_items() + return pln + + def test_transfer_batches_share_stock_across_requirements(self): + rm_item = make_item(properties={"is_stock_item": 1, "has_batch_no": 1, "create_new_batch": 1}).name + source_warehouse = "_Test Warehouse 1 - _TC" + for qty in (1, 1, 5, 3, 3, 4, 100): + make_stock_entry(item_code=rm_item, qty=qty, rate=100, target=source_warehouse) + pln = self._plan_for_transfer_allocation(rm_item, qty_per_order=250) + pln.po_items[1].planned_qty = 1000 + pln.for_warehouse = "_Test Warehouse - _TC" + warehouses = [{"warehouse": source_warehouse}] + + items = get_items_for_material_requests(pln.as_dict(), warehouses=warehouses) + self.assertEqual( + [row["material_request_type"] for row in items], ["Material Transfer", "Purchase", "Purchase"] + ) + self.assertEqual([row["quantity"] for row in items], [117, 133, 1000]) + self.assertEqual(items[0]["from_warehouse"], source_warehouse) + self.assertEqual([row["warehouse"] for row in items], [pln.for_warehouse] * 3) + self.assertEqual([row["required_bom_qty"] for row in items], [250, 250, 1000]) + self.assertEqual( + [row["sales_order"] for row in items], + [pln.po_items[0].sales_order] * 2 + [pln.po_items[1].sales_order], + ) + self.assertEqual(items, get_items_for_material_requests(pln.as_dict(), warehouses=warehouses)) + + def test_transfer_batches_keep_source_warehouses_and_requirements_separate(self): + rm_item = make_item(properties={"is_stock_item": 1, "has_batch_no": 1, "create_new_batch": 1}).name + source_warehouses = ["_Test Warehouse 1 - _TC", "_Test Warehouse 2 - _TC"] + for warehouse, quantities in zip(source_warehouses, ((10, 20), (50, 60)), strict=True): + for qty in quantities: + make_stock_entry(item_code=rm_item, qty=qty, rate=100, target=warehouse) + pln = self._plan_for_transfer_allocation(rm_item, qty_per_order=50) + pln.po_items[1].planned_qty = 100 + pln.for_warehouse = "_Test Warehouse - _TC" + + items = get_items_for_material_requests( + pln.as_dict(), warehouses=[{"warehouse": warehouse} for warehouse in source_warehouses] + ) + transfers = [row for row in items if row["material_request_type"] == "Material Transfer"] + self.assertEqual(len(transfers), 3) + self.assertEqual( + {(row["sales_order"], row["from_warehouse"]): row["quantity"] for row in transfers}, + { + (pln.po_items[0].sales_order, source_warehouses[0]): 30, + (pln.po_items[0].sales_order, source_warehouses[1]): 20, + (pln.po_items[1].sales_order, source_warehouses[1]): 90, + }, + ) + purchases = [row for row in items if row["material_request_type"] == "Purchase"] + self.assertEqual(len(purchases), 1) + self.assertEqual(purchases[0]["quantity"], 10) + self.assertEqual(purchases[0]["sales_order"], pln.po_items[1].sales_order) + + def test_transfer_shared_stock_uses_stock_uom(self): + rm_item = make_item( + properties={"is_stock_item": 1, "stock_uom": "Nos", "purchase_uom": "_Test UOM 1"}, + uoms=[{"uom": "_Test UOM 1", "conversion_factor": 10}], + ).name + source_warehouse = "_Test Warehouse 1 - _TC" + make_stock_entry(item_code=rm_item, qty=60, rate=100, target=source_warehouse) + pln = self._plan_for_transfer_allocation(rm_item, qty_per_order=50) + pln.for_warehouse = "_Test Warehouse - _TC" + + items = get_items_for_material_requests(pln.as_dict(), warehouses=[{"warehouse": source_warehouse}]) + self.assertEqual( + [row["material_request_type"] for row in items], + ["Material Transfer", "Material Transfer", "Purchase"], + ) + self.assertEqual([row["quantity"] for row in items], [50, 10, 4]) + self.assertEqual([row["uom"] for row in items], ["Nos", "Nos", "_Test UOM 1"]) + self.assertEqual([row["conversion_factor"] for row in items], [1, 1, 10]) + self.assertEqual( + [row["sales_order"] for row in items], + [pln.po_items[0].sales_order] + [pln.po_items[1].sales_order] * 2, + ) + + def test_transfer_shared_stock_rounds_away_float_residue(self): + from erpnext.manufacturing.doctype.production_plan.production_plan import ( + _transfer_from_locations, + ) + + locations = [ + frappe._dict(qty=0.7, warehouse="_Test Warehouse 1 - _TC"), + frappe._dict(qty=1, warehouse="_Test Warehouse 2 - _TC"), + ] + transfers = [] + for quantity in (0.1, 0.2, 0.4): + item = {"item_code": "Raw Material Item 1", "quantity": quantity, "conversion_factor": 1} + self.assertEqual(_transfer_from_locations(item, locations, transfers, quantity), 0) + + self.assertEqual( + [(row["from_warehouse"], row["quantity"]) for row in transfers], + [("_Test Warehouse 1 - _TC", quantity) for quantity in (0.1, 0.2, 0.4)], + ) + def test_transfer_and_purchase_mrp_for_purchase_uom(self): from erpnext.manufacturing.doctype.bom.test_bom import create_nested_bom from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse