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

fix(manufacturing): apply safety stock once across Production Plan rows (v16) (#58832)

Browse Source
This commit is contained in:
Mihir Kandoi
2026-09-08 09:41:24 +05:30
committed by GitHub
parent e59fb396e5
commit 4719ad9b91
2 changed files with 197 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1450,42 +1450,18 @@ def get_material_request_items(
bin_dict,
consumed_qty,
):
required_qty = 0
item_code = row.get("item_code")
if not ignore_existing_ordered_qty or bin_dict.get("projected_qty", 0) < 0:
required_qty = flt(row.get("qty"))
else:
key = (item_code, warehouse)
available_qty = flt(bin_dict.get("projected_qty", 0)) - consumed_qty[key]
if available_qty > 0:
required_qty = max(0, flt(row.get("qty")) - available_qty)
consumed_qty[key] += min(flt(row.get("qty")), available_qty)
else:
required_qty = flt(row.get("qty"))
required_qty = _required_qty_for_mr(
row, ignore_existing_ordered_qty, warehouse, bin_dict, consumed_qty, include_safety_stock
)
if doc.get("consider_minimum_order_qty") and required_qty > 0 and required_qty < row["min_order_qty"]:
required_qty = row["min_order_qty"]
item_group_defaults = get_item_group_defaults(row.item_code, company)
if not row["purchase_uom"]:
row["purchase_uom"] = row["stock_uom"]
if row["purchase_uom"] != row["stock_uom"]:
if not (row["conversion_factor"] or frappe.flags.show_qty_in_stock_uom):
frappe.throw(
_("UOM Conversion factor ({0} -> {1}) not found for item: {2}").format(
row["purchase_uom"], row["stock_uom"], row.item_code
)
)
if frappe.db.get_value("UOM", row["purchase_uom"], "must_be_whole_number"):
required_qty = ceil(required_qty)
if include_safety_stock:
required_qty += flt(row["safety_stock"])
item_details = frappe.get_cached_value("Item", row.item_code, ["purchase_uom", "stock_uom"], as_dict=1)
conversion_factor = 1.0
@@ -1606,6 +1582,37 @@ def get_sales_orders(self):
return open_so
def _required_qty_for_mr(
row, ignore_existing_ordered_qty, warehouse, bin_dict, consumed_qty, include_safety_stock
):
safety_stock = flt(row["safety_stock"]) if include_safety_stock else 0
qty = flt(row.get("qty"))
projected_qty = max(0, flt(bin_dict.get("projected_qty"))) if ignore_existing_ordered_qty else 0
key = (row.get("item_code"), warehouse)
available_qty = projected_qty - consumed_qty[key]
required_qty = max(0, qty - (available_qty - safety_stock))
consumed_qty[key] += qty - required_qty
return _adjust_required_qty_for_uom(row, required_qty)
def _adjust_required_qty_for_uom(row, required_qty):
if not row["purchase_uom"]:
row["purchase_uom"] = row["stock_uom"]
if row["purchase_uom"] != row["stock_uom"]:
if not (row["conversion_factor"] or frappe.flags.show_qty_in_stock_uom):
frappe.throw(
_("UOM Conversion factor ({0} -> {1}) not found for item: {2}").format(
row["purchase_uom"], row["stock_uom"], row.item_code
)
)
if frappe.db.get_value("UOM", row["purchase_uom"], "must_be_whole_number"):
required_qty = ceil(required_qty)
return required_qty
@frappe.whitelist()
def get_bin_details(row, company, for_warehouse=None, all_warehouse=False):
if isinstance(row, str):

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

@@ -110,6 +110,169 @@ class TestProductionPlan(ERPNextTestSuite):
pln = frappe.get_doc("Production Plan", pln.name)
pln.cancel()
def _plan_for_safety_stock(self, rm_item, qty_per_order, bom_quantity=1):
fg_item = make_item(properties={"is_stock_item": 1}).name
make_bom(
item=fg_item,
raw_materials=[rm_item],
source_warehouse="_Test Warehouse - _TC",
quantity=bom_quantity,
)
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_safety_stock_added_once_for_repeated_raw_material(self):
rm_item = make_item(properties={"is_stock_item": 1, "safety_stock": 10, "valuation_rate": 100}).name
make_stock_entry(item_code=rm_item, qty=100, rate=100, target="_Test Warehouse - _TC")
pln = self._plan_for_safety_stock(rm_item, qty_per_order=50)
pln.include_safety_stock = 1
items = get_items_for_material_requests(pln.as_dict())
quantities = sorted(flt(d.get("quantity")) for d in items if d.get("item_code") == rm_item)
self.assertEqual(quantities, [0, 10])
def test_safety_stock_added_once_with_negative_or_ignored_projected_qty(self):
from erpnext.stock.utils import get_or_make_bin
rm_item = make_item(properties={"is_stock_item": 1, "safety_stock": 100}).name
bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC")
pln = self._plan_for_safety_stock(rm_item, qty_per_order=250)
pln.po_items[1].planned_qty = 1000
pln.include_safety_stock = 1
for projected_qty in (-5, 0, 200, 1500):
frappe.db.set_value("Bin", bin_name, "projected_qty", projected_qty)
for consider_projected_qty in (0, 1):
with self.subTest(projected_qty=projected_qty, consider_projected_qty=consider_projected_qty):
pln.ignore_existing_ordered_qty = consider_projected_qty
items = get_items_for_material_requests(pln.as_dict())
expected_qty = [350, 1000]
if consider_projected_qty and projected_qty > 0:
expected_qty = [150, 1000] if projected_qty == 200 else [0, 0]
self.assertEqual([row["quantity"] for row in items], expected_qty)
self.assertEqual([row["required_bom_qty"] for row in items], [250, 1000])
self.assertEqual([row["safety_stock"] for row in items], [100, 100])
self.assertEqual(
[row["sales_order"] for row in items], [row.sales_order for row in pln.po_items]
)
def test_safety_stock_disabled_with_negative_projected_qty(self):
from erpnext.stock.utils import get_or_make_bin
rm_item = make_item(properties={"is_stock_item": 1, "safety_stock": 100}).name
bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC")
frappe.db.set_value("Bin", bin_name, "projected_qty", -5)
pln = self._plan_for_safety_stock(rm_item, qty_per_order=250)
pln.po_items[1].planned_qty = 1000
items = get_items_for_material_requests(pln.as_dict())
self.assertEqual([row["quantity"] for row in items], [250, 1000])
def test_safety_stock_is_separate_for_each_item_and_warehouse(self):
from collections import defaultdict
from erpnext.manufacturing.doctype.production_plan.production_plan import (
_required_qty_for_mr,
)
row = frappe._dict(qty=250, safety_stock=100, purchase_uom="Nos", stock_uom="Nos")
items_and_warehouses = [
("Raw Material Item 1", "_Test Warehouse - _TC"),
("Raw Material Item 1", "_Test Warehouse 1 - _TC"),
("Raw Material Item 2", "_Test Warehouse - _TC"),
]
for consider_projected_qty in (0, 1):
with self.subTest(consider_projected_qty=consider_projected_qty):
consumed_qty = defaultdict(float)
quantities = []
for item_code, warehouse in items_and_warehouses * 2:
row.item_code = item_code
quantities.append(
_required_qty_for_mr(
row, consider_projected_qty, warehouse, {"projected_qty": -5}, consumed_qty, True
)
)
self.assertEqual(quantities, [350, 350, 350, 250, 250, 250])
def test_safety_stock_added_once_before_transferring_materials(self):
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
from erpnext.stock.utils import get_or_make_bin
rm_item = make_item(properties={"is_stock_item": 1, "safety_stock": 100, "min_order_qty": 1234}).name
source_warehouse = create_warehouse("Safety Stock Source Warehouse", company="_Test Company")
make_stock_entry(item_code=rm_item, qty=2000, rate=100, target=source_warehouse)
bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC")
frappe.db.set_value("Bin", bin_name, "projected_qty", -5)
pln = self._plan_for_safety_stock(rm_item, qty_per_order=250)
pln.po_items[1].planned_qty = 1000
pln.for_warehouse = "_Test Warehouse - _TC"
pln.include_safety_stock = 1
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"] * 2)
self.assertEqual([row["quantity"] for row in items], [350, 1000])
def test_safety_stock_does_not_share_purchase_rounding_between_rows(self):
from erpnext.stock.utils import get_or_make_bin
rm_item = make_item(properties={"is_stock_item": 1, "stock_uom": "Nos", "safety_stock": 1}).name
pln = self._plan_for_safety_stock(rm_item, qty_per_order=1, bom_quantity=2)
bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC")
for projected_qty in (0, 0.25, 0.75):
frappe.db.set_value("Bin", bin_name, "projected_qty", projected_qty)
for include_safety_stock in (0, 1):
for consider_projected_qty in (0, 1):
with self.subTest(
projected_qty=projected_qty,
include_safety_stock=include_safety_stock,
consider_projected_qty=consider_projected_qty,
):
pln.include_safety_stock = include_safety_stock
pln.ignore_existing_ordered_qty = consider_projected_qty
items = get_items_for_material_requests(pln.as_dict())
expected_qty = [2, 1] if include_safety_stock else [1, 1]
if consider_projected_qty and projected_qty == 0.75:
expected_qty = [1, 1] if include_safety_stock else [0, 1]
self.assertEqual([row["quantity"] for row in items], expected_qty)
self.assertEqual([row["required_bom_qty"] for row in items], [0.5, 0.5])
self.assertEqual(
[row["sales_order"] for row in items], [row.sales_order for row in pln.po_items]
)
def test_safety_stock_with_fractional_minimum_uses_whole_purchase_uom(self):
rm_item = make_item(
properties={"is_stock_item": 1, "stock_uom": "Nos", "safety_stock": 0.5, "min_order_qty": 2.5}
).name
pln = self._plan_for_safety_stock(rm_item, qty_per_order=1)
pln.set("po_items", [pln.po_items[0]])
pln.include_safety_stock = 1
pln.consider_minimum_order_qty = 1
items = get_items_for_material_requests(pln.as_dict())
self.assertEqual(len(items), 1)
self.assertEqual(items[0]["quantity"], 3)
def test_production_plan_start_date(self):
"Test if Work Order has same Planned Start Date as Prod Plan."
planned_date = add_to_date(date=None, days=3)