nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-09-08 23:09:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(manufacturing): share transfer stock across Production Plan rows (v16) (#58834)

Browse Source
This commit is contained in:
Mihir Kandoi
2026-09-08 09:29:46 +05:30
committed by GitHub
parent a7c5ab89e8
commit e59fb396e5
2 changed files with 182 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
erpnext/manufacturing/doctype/production_plan/production_plan.py
View File

@@ -1665,9 +1665,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)))
@@ -1843,6 +1844,7 @@ def get_items_for_material_requests(doc, warehouses=None, get_parent_warehouse_d
if (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,
@@ -1850,6 +1852,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
@@ -1873,45 +1876,19 @@ 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
purchase_uom = frappe.db.get_value("Item", item.get("item_code"), "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
@@ -1931,6 +1908,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)

119
erpnext/manufacturing/doctype/production_plan/test_production_plan.py
View File

@@ -1790,6 +1790,125 @@ class TestProductionPlan(ERPNextTestSuite):
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=1,
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