mirror of
https://github.com/frappe/erpnext.git
synced 2026-09-09 15:29:30 +00:00
fix(manufacturing): apply MOQ once across Production Plan rows (v15)
This commit is contained in:
Mihir Kandoi
@@ -5,6 +5,7 @@
|
||||
import copy
|
||||
import json
|
||||
from collections import defaultdict
|
||||
from decimal import ROUND_CEILING, Decimal
|
||||
|
||||
import frappe
|
||||
from frappe import _, msgprint
|
||||
@@ -1353,9 +1354,6 @@ def get_material_request_items(
|
||||
else:
|
||||
required_qty = flt(row.get("qty"))
|
||||
|
||||
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"]:
|
||||
@@ -1389,11 +1387,12 @@ def get_material_request_items(
|
||||
get_conversion_factor(row.item_code, item_details.purchase_uom).get("conversion_factor") or 1.0
|
||||
)
|
||||
|
||||
min_order_qty = flt(row.get("min_order_qty")) if doc.get("consider_minimum_order_qty") else 0
|
||||
if flt(row.get("qty")) > 0:
|
||||
return {
|
||||
"item_code": row.item_code,
|
||||
"item_name": row.item_name,
|
||||
"quantity": required_qty / conversion_factor,
|
||||
"quantity": _quantity_in_purchase_uom(required_qty, conversion_factor, min_order_qty),
|
||||
"conversion_factor": conversion_factor,
|
||||
"required_bom_qty": row.get("qty"),
|
||||
"stock_uom": row.get("stock_uom"),
|
||||
@@ -1415,6 +1414,91 @@ def get_material_request_items(
|
||||
}
|
||||
|
||||
|
||||
def _apply_minimum_order_qty(mr_items):
|
||||
for rows in _purchase_rows_by_item(mr_items).values():
|
||||
surplus_qty = 0.0
|
||||
for order_rows in _rows_by_sales_order(rows):
|
||||
surplus_qty = _apply_minimum_order_qty_to_order(order_rows, surplus_qty)
|
||||
|
||||
|
||||
def _purchase_rows_by_item(mr_items):
|
||||
rows_by_item = defaultdict(list)
|
||||
for row in mr_items:
|
||||
if row.get("material_request_type") not in ("Purchase", "Subcontracting"):
|
||||
continue
|
||||
if flt(row.get("quantity")) <= 0:
|
||||
continue
|
||||
key = (
|
||||
row.get("item_code"),
|
||||
row.get("warehouse"),
|
||||
row.get("material_request_type"),
|
||||
row.get("supplier"),
|
||||
)
|
||||
rows_by_item[key].append(row)
|
||||
return rows_by_item
|
||||
|
||||
|
||||
def _rows_by_sales_order(rows):
|
||||
rows_by_order = defaultdict(list)
|
||||
for row in rows:
|
||||
rows_by_order[row.get("sales_order")].append(row)
|
||||
return rows_by_order.values()
|
||||
|
||||
|
||||
def _apply_minimum_order_qty_to_order(rows, surplus_qty):
|
||||
"""Cover the order from an earlier order's surplus, then raise the rest to the minimum.
|
||||
|
||||
Material Requests and Purchase Orders are raised per Sales Order and a Purchase
|
||||
Order rejects an item below its minimum, so each order either buys at least the
|
||||
minimum or is covered by what an earlier order over-purchased."""
|
||||
demand_qty = sum(_stock_quantity(row) for row in rows)
|
||||
_cover_from_surplus(rows, surplus_qty)
|
||||
|
||||
min_order_qty = max(flt(row.get("min_order_qty")) for row in rows)
|
||||
total_qty = sum(_stock_quantity(row) for row in rows)
|
||||
if 0 < total_qty < min_order_qty:
|
||||
row = next(row for row in rows if _stock_quantity(row) > 0)
|
||||
_set_stock_quantity(row, _stock_quantity(row) + min_order_qty - total_qty)
|
||||
|
||||
purchased_qty = sum(_stock_quantity(row) for row in rows)
|
||||
return surplus_qty + purchased_qty - demand_qty
|
||||
|
||||
|
||||
def _cover_from_surplus(rows, surplus_qty):
|
||||
for row in rows:
|
||||
covered_qty = min(surplus_qty, _stock_quantity(row))
|
||||
if covered_qty <= 0:
|
||||
break
|
||||
_set_stock_quantity(row, _stock_quantity(row) - covered_qty)
|
||||
surplus_qty -= covered_qty
|
||||
|
||||
|
||||
def _stock_quantity(row):
|
||||
return flt(row.get("quantity")) * (flt(row.get("conversion_factor")) or 1)
|
||||
|
||||
|
||||
def _set_stock_quantity(row, stock_qty):
|
||||
conversion_factor = flt(row.get("conversion_factor")) or 1
|
||||
quantity = _quantity_in_purchase_uom(stock_qty, conversion_factor, stock_qty)
|
||||
if frappe.get_cached_value("UOM", row.get("uom"), "must_be_whole_number"):
|
||||
quantity = ceil(quantity)
|
||||
row["quantity"] = quantity
|
||||
|
||||
|
||||
def _quantity_in_purchase_uom(required_qty, conversion_factor, min_order_qty=0):
|
||||
"""Convert to purchase UOM; a binding minimum order qty takes the smallest
|
||||
representable quantity whose stock equivalent still meets it. The minimum is
|
||||
capped at the requirement so a small shortage never rounds down to zero."""
|
||||
min_order_qty = min(min_order_qty, required_qty)
|
||||
precision = frappe.get_precision("Material Request Plan Item", "quantity")
|
||||
quantity = flt(required_qty / conversion_factor, precision)
|
||||
if min_order_qty and quantity * conversion_factor < min_order_qty <= required_qty:
|
||||
grid = Decimal(10) ** -precision
|
||||
exact = Decimal(str(min_order_qty)) / Decimal(str(conversion_factor))
|
||||
quantity = flt(exact.quantize(grid, rounding=ROUND_CEILING))
|
||||
return quantity
|
||||
|
||||
|
||||
def get_sales_orders(self):
|
||||
bom = frappe.qb.DocType("BOM")
|
||||
pi = frappe.qb.DocType("Packed Item")
|
||||
@@ -1543,9 +1627,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)))
|
||||
@@ -1731,6 +1816,9 @@ def get_items_for_material_requests(doc, warehouses=None, get_parent_warehouse_d
|
||||
|
||||
mr_items = new_mr_items
|
||||
|
||||
if doc.get("consider_minimum_order_qty"):
|
||||
_apply_minimum_order_qty(mr_items)
|
||||
|
||||
if not mr_items:
|
||||
to_enable = frappe.bold(_("Ignore Existing Projected Quantity"))
|
||||
warehouse = frappe.bold(doc.get("for_warehouse"))
|
||||
@@ -1794,13 +1882,13 @@ def get_materials_from_other_locations(
|
||||
|
||||
precision = frappe.get_precision("Material Request Plan Item", "quantity")
|
||||
if flt(required_qty, precision) > 0:
|
||||
if consider_minimum_order_qty:
|
||||
required_qty = max(required_qty, flt(item.get("min_order_qty")))
|
||||
|
||||
if frappe.db.get_value("UOM", purchase_uom, "must_be_whole_number"):
|
||||
required_qty = ceil(required_qty)
|
||||
|
||||
item["quantity"] = required_qty / item.get("conversion_factor")
|
||||
min_order_qty = flt(item.get("min_order_qty")) if consider_minimum_order_qty else 0
|
||||
item["quantity"] = _quantity_in_purchase_uom(
|
||||
required_qty, item.get("conversion_factor"), min_order_qty
|
||||
)
|
||||
|
||||
new_mr_items.append(item)
|
||||
|
||||
|
||||
@@ -54,6 +54,196 @@ class TestProductionPlan(FrappeTestCase):
|
||||
def tearDown(self) -> None:
|
||||
frappe.db.rollback()
|
||||
|
||||
def _plan_with_shared_raw_material(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_minimum_order_qty_surplus_covers_later_rows(self):
|
||||
rm_item = make_item(properties={"is_stock_item": 1, "min_order_qty": 100, "valuation_rate": 100}).name
|
||||
make_stock_entry(item_code=rm_item, qty=40, rate=100, target="_Test Warehouse - _TC")
|
||||
|
||||
pln = self._plan_with_shared_raw_material(rm_item, qty_per_order=50)
|
||||
pln.consider_minimum_order_qty = 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, 100])
|
||||
|
||||
def test_minimum_order_qty_surplus_carries_across_sales_orders(self):
|
||||
from erpnext.stock.utils import get_or_make_bin
|
||||
|
||||
rm_item = make_item(properties={"is_stock_item": 1, "min_order_qty": 1234}).name
|
||||
bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC")
|
||||
pln = self._plan_with_shared_raw_material(rm_item, qty_per_order=250)
|
||||
pln.consider_minimum_order_qty = 1
|
||||
|
||||
for projected_qty in (0, -5):
|
||||
frappe.db.set_value("Bin", bin_name, "projected_qty", projected_qty)
|
||||
for consider_projected_qty in (0, 1):
|
||||
pln.ignore_existing_ordered_qty = not consider_projected_qty
|
||||
for second_qty, expected_qty in ((500, [1234, 0]), (984, [1234, 0]), (1000, [1234, 1234])):
|
||||
with self.subTest(
|
||||
projected_qty=projected_qty,
|
||||
consider_projected_qty=consider_projected_qty,
|
||||
second_qty=second_qty,
|
||||
):
|
||||
pln.po_items[1].planned_qty = second_qty
|
||||
items = get_items_for_material_requests(pln.as_dict())
|
||||
self.assertEqual([row["quantity"] for row in items], expected_qty)
|
||||
self.assertEqual([row["required_bom_qty"] for row in items], [250, second_qty])
|
||||
self.assertEqual(
|
||||
[row["sales_order"] for row in items], [row.sales_order for row in pln.po_items]
|
||||
)
|
||||
|
||||
def test_minimum_order_qty_does_not_purchase_when_stock_covers_demand(self):
|
||||
rm_item = make_item(properties={"is_stock_item": 1, "min_order_qty": 1234}).name
|
||||
make_stock_entry(item_code=rm_item, qty=1000, rate=100, target="_Test Warehouse - _TC")
|
||||
pln = self._plan_with_shared_raw_material(rm_item, qty_per_order=250)
|
||||
pln.consider_minimum_order_qty = 1
|
||||
|
||||
items = get_items_for_material_requests(pln.as_dict())
|
||||
self.assertEqual([row["quantity"] for row in items], [0, 0])
|
||||
|
||||
def test_minimum_order_qty_disabled_for_repeated_raw_material(self):
|
||||
rm_item = make_item(properties={"is_stock_item": 1, "min_order_qty": 1234}).name
|
||||
pln = self._plan_with_shared_raw_material(rm_item, qty_per_order=250)
|
||||
pln.po_items[1].planned_qty = 500
|
||||
|
||||
items = get_items_for_material_requests(pln.as_dict())
|
||||
self.assertEqual([row["quantity"] for row in items], [250, 500])
|
||||
|
||||
def test_minimum_order_qty_does_not_transfer_surplus_stock(self):
|
||||
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
|
||||
|
||||
rm_item = make_item(properties={"is_stock_item": 1, "min_order_qty": 1234}).name
|
||||
source_warehouse = create_warehouse("MOQ Sufficient Source Warehouse", company="_Test Company")
|
||||
make_stock_entry(item_code=rm_item, qty=1500, rate=100, target=source_warehouse)
|
||||
pln = self._plan_with_shared_raw_material(rm_item, qty_per_order=250)
|
||||
pln.consider_minimum_order_qty = 1
|
||||
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"] * 2)
|
||||
self.assertEqual([row["quantity"] for row in items], [250, 250])
|
||||
|
||||
def test_minimum_order_qty_respects_purchase_groups_and_sales_orders(self):
|
||||
from erpnext.manufacturing.doctype.production_plan.production_plan import (
|
||||
_apply_minimum_order_qty,
|
||||
)
|
||||
|
||||
base_row = {
|
||||
"item_code": "Raw Material Item 1",
|
||||
"warehouse": "_Test Warehouse - _TC",
|
||||
"supplier": "_Test Supplier",
|
||||
"sales_order": "SO-1",
|
||||
"material_request_type": "Purchase",
|
||||
"uom": "Nos",
|
||||
"conversion_factor": 1,
|
||||
"min_order_qty": 1234,
|
||||
"quantity": 250,
|
||||
}
|
||||
rows = [
|
||||
base_row.copy(),
|
||||
base_row | {"quantity": 500},
|
||||
base_row | {"warehouse": "_Test Warehouse 1 - _TC"},
|
||||
base_row | {"supplier": "_Test Supplier 1"},
|
||||
base_row | {"item_code": "Raw Material Item 2"},
|
||||
base_row | {"material_request_type": "Subcontracting"},
|
||||
base_row | {"material_request_type": "Material Transfer", "quantity": 1000},
|
||||
base_row | {"material_request_type": "Manufacture"},
|
||||
base_row | {"sales_order": "SO-2", "quantity": 400},
|
||||
base_row | {"sales_order": "SO-3", "quantity": 100},
|
||||
]
|
||||
expected_rows = [row.copy() for row in rows]
|
||||
quantities = [734, 500, 1234, 1234, 1234, 1234, 1000, 250, 0, 1234]
|
||||
for row, quantity in zip(expected_rows, quantities, strict=True):
|
||||
row["quantity"] = quantity
|
||||
|
||||
_apply_minimum_order_qty(rows)
|
||||
self.assertEqual(rows, expected_rows)
|
||||
|
||||
def test_minimum_order_qty_shortfall_uses_stock_uom(self):
|
||||
from erpnext.manufacturing.doctype.production_plan.production_plan import (
|
||||
_apply_minimum_order_qty,
|
||||
)
|
||||
|
||||
frappe.db.set_default("float_precision", "3")
|
||||
rows = [
|
||||
{
|
||||
"item_code": "Raw Material Item 1",
|
||||
"warehouse": "_Test Warehouse - _TC",
|
||||
"material_request_type": "Purchase",
|
||||
"uom": uom,
|
||||
"conversion_factor": conversion_factor,
|
||||
"min_order_qty": 1234,
|
||||
"quantity": quantity,
|
||||
}
|
||||
for uom, conversion_factor, quantity in (("_Test UOM 1", 7, 10), ("Nos", 1, 500))
|
||||
]
|
||||
_apply_minimum_order_qty(rows)
|
||||
self.assertEqual([row["quantity"] for row in rows], [104.858, 500])
|
||||
self.assertGreaterEqual(sum(row["quantity"] * row["conversion_factor"] for row in rows), 1234)
|
||||
|
||||
rows[0].update(uom="Nos", quantity=10)
|
||||
_apply_minimum_order_qty(rows)
|
||||
self.assertEqual([row["quantity"] for row in rows], [105, 500])
|
||||
|
||||
def test_minimum_order_qty_surplus_includes_rounding(self):
|
||||
from erpnext.manufacturing.doctype.production_plan.production_plan import (
|
||||
_apply_minimum_order_qty,
|
||||
)
|
||||
|
||||
rows = [
|
||||
{
|
||||
"item_code": "Raw Material Item 1",
|
||||
"warehouse": "_Test Warehouse - _TC",
|
||||
"sales_order": sales_order,
|
||||
"material_request_type": "Purchase",
|
||||
"uom": "Nos",
|
||||
"conversion_factor": 2,
|
||||
"min_order_qty": 5,
|
||||
"quantity": 1.5,
|
||||
}
|
||||
for sales_order in ("SO-1", "SO-2")
|
||||
]
|
||||
_apply_minimum_order_qty(rows)
|
||||
self.assertEqual([row["quantity"] for row in rows], [3, 0])
|
||||
|
||||
def test_min_order_qty_keeps_small_shortages_in_purchase_uom(self):
|
||||
frappe.db.set_default("float_precision", "3")
|
||||
conversion_factor = 10000
|
||||
rm_item = make_item(
|
||||
properties={"is_stock_item": 1, "min_order_qty": 50000, "purchase_uom": "_Test UOM 1"},
|
||||
uoms=[{"uom": "_Test UOM 1", "conversion_factor": conversion_factor}],
|
||||
).name
|
||||
pln = self._plan_with_shared_raw_material(rm_item, qty_per_order=1)
|
||||
pln.consider_minimum_order_qty = 1
|
||||
|
||||
items = get_items_for_material_requests(pln.as_dict())
|
||||
self.assertEqual([row["quantity"] for row in items], [5, 0])
|
||||
self.assertEqual(sum(row["quantity"] * row["conversion_factor"] for row in items), 50000)
|
||||
|
||||
def test_production_plan_mr_creation(self):
|
||||
"Test if MRs are created for unavailable raw materials."
|
||||
pln = create_production_plan(item_code="Test Production Item 1")
|
||||
|
||||
Reference in New Issue
Block a user