diff --git a/erpnext/manufacturing/doctype/production_plan/services/material_request.py b/erpnext/manufacturing/doctype/production_plan/services/material_request.py new file mode 100644 index 00000000000..c9073d5e648 --- /dev/null +++ b/erpnext/manufacturing/doctype/production_plan/services/material_request.py @@ -0,0 +1,731 @@ +# Copyright (c) 2017, Frappe Technologies Pvt. Ltd. and contributors +# For license information, please see license.txt + +"""Material Request planning and creation for a Production Plan. + +Consolidates the former ``material_planning``, ``material_request_items`` and +``material_request_helpers`` modules. Also re-exports the planning helpers so +existing imports of ``...services.material_planning`` keep working through here. +""" + +import copy +import json +from collections import defaultdict + +import frappe +from frappe import _, msgprint +from frappe.model.document import Document +from frappe.utils import add_days, ceil, cint, comma_and, flt, get_link_to_form, nowdate +from frappe.utils.csvutils import build_csv_response + +from erpnext.manufacturing.doctype.production_plan.services.bom_explosion import ( + get_exploded_items, + get_subitems, +) +from erpnext.manufacturing.doctype.production_plan.services.planning_queries import ( + get_bin_details, + get_item_data, + get_sales_orders, + get_uom_conversion_factor, + get_warehouse_list, + set_default_warehouses, +) +from erpnext.manufacturing.doctype.production_plan.services.sub_assembly_queries import ( + get_raw_materials_of_sub_assembly_items, + get_sub_assembly_items, +) +from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults +from erpnext.stock.get_item_details import get_conversion_factor + + +class MaterialRequestService: + def __init__(self, doc): + self.doc = doc + + def validate_mr_subcontracted(self): + for row in self.doc.mr_items: + if row.material_request_type != "Subcontracting": + continue + if not frappe.db.get_value("Item", row.item_code, "is_sub_contracted_item"): + frappe.throw( + _("Item {0} is not a subcontracted item").format(row.item_code), + title=_("Invalid Item"), + ) + + def make_material_request(self): + """Create Material Requests grouped by Sales Order and Material Request Type""" + self.validate_mr_subcontracted() + + if all(item.requested_qty == item.quantity for item in self.doc.mr_items): + msgprint(_("All items are already requested")) + return + + material_request_map = {} + material_request_list = [] + for item in self.doc.mr_items: + if item.quantity == item.requested_qty: + continue + self._add_item_to_material_request(item, material_request_map, material_request_list) + + self._submit_material_requests(material_request_list) + + def _add_item_to_material_request(self, item, material_request_map, material_request_list): + item_doc = frappe.get_cached_doc("Item", item.item_code) + material_request_type = item.material_request_type or item_doc.default_material_request_type + + # key for Sales Order:Material Request Type:Customer + key = "{}:{}:{}".format(item.sales_order, material_request_type, "") + if key not in material_request_map: + material_request_map[key] = self._new_material_request(material_request_type) + material_request_list.append(material_request_map[key]) + + schedule_date = item.schedule_date or add_days(nowdate(), cint(item_doc.lead_time_days)) + row = self._material_request_item(item, material_request_type, schedule_date) + material_request_map[key].append("items", row) + + def _new_material_request(self, material_request_type): + mr = frappe.new_doc("Material Request") + mr.update( + { + "transaction_date": nowdate(), + "status": "Draft", + "company": self.doc.company, + "material_request_type": material_request_type, + } + ) + return mr + + def _material_request_item(self, item, material_request_type, schedule_date): + from_warehouse = item.from_warehouse if material_request_type == "Material Transfer" else None + # a group warehouse cannot receive stock; it must never reach a Material Request line + if item.warehouse and frappe.get_cached_value("Warehouse", item.warehouse, "is_group"): + frappe.throw( + _("Cannot create Material Request for item {0} in group warehouse {1}.").format( + frappe.bold(item.item_code), frappe.bold(item.warehouse) + ) + ) + project = ( + frappe.db.get_value("Sales Order", item.sales_order, "project") if item.sales_order else None + ) + return { + "item_code": item.item_code, + "from_warehouse": from_warehouse, + "qty": item.quantity - item.requested_qty, + "uom": item.uom, + "schedule_date": schedule_date, + "warehouse": item.warehouse, + "sales_order": item.sales_order, + "production_plan": self.doc.name, + "material_request_plan_item": item.name, + "project": project, + } + + def _submit_material_requests(self, material_request_list): + for material_request in material_request_list: + material_request.flags.ignore_permissions = 1 + material_request.run_method("set_missing_values") + material_request.save() + if self.doc.get("submit_material_request"): + material_request.submit() + + frappe.flags.mute_messages = False + if not material_request_list: + msgprint(_("No material request created")) + return + + links = [get_link_to_form("Material Request", m.name) for m in material_request_list] + msgprint(_("{0} created").format(comma_and(links))) + + +@frappe.whitelist() +def get_items_for_material_requests( + doc: str | dict | Document, + warehouses: str | list | None = None, + get_parent_warehouse_data: bool | int | None = None, +): + frappe.has_permission("Production Plan", "read", throw=True) + + doc = _normalize_mr_doc(doc) + _validate_group_warehouse_target(doc) + warehouses = _filter_warehouses(doc, warehouses, get_parent_warehouse_data) + doc["mr_items"] = [] + + po_items = _collect_po_items(doc) + _validate_po_items(po_items) + + ignore_ordered_qty = _effective_ignore_ordered_qty(doc, po_items) + so_item_details = _collect_item_details(doc, po_items) + + mr_items = _build_mr_items(doc, so_item_details, ignore_ordered_qty) + mr_items = _apply_other_locations( + doc, mr_items, warehouses, ignore_ordered_qty, get_parent_warehouse_data + ) + + if not mr_items: + _warn_no_mr_items(doc) + return mr_items + + +def _normalize_mr_doc(doc): + doc = frappe._dict(frappe.parse_json(doc)) + return doc + + +def _validate_group_warehouse_target(doc): + # the group only scopes availability; raw materials still need a concrete + # receiving warehouse, so for_warehouse is required once we generate items. + if doc.get("raw_material_group_warehouse") and not doc.get("for_warehouse"): + frappe.throw( + _("{0} is required to get raw materials when {1} is set.").format( + frappe.bold(_("For Warehouse")), frappe.bold(_("Raw Material Group Warehouse")) + ) + ) + + +def _filter_warehouses(doc, warehouses, get_parent_warehouse_data): + if not warehouses: + return warehouses + + warehouses = list(set(get_warehouse_list(warehouses))) + for_warehouse = doc.get("for_warehouse") + if for_warehouse and not get_parent_warehouse_data and for_warehouse in warehouses: + warehouses.remove(for_warehouse) + return warehouses + + +def _collect_po_items(doc): + po_items = doc.get("po_items") if doc.get("po_items") else doc.get("items") + for sa_row in doc.get("sub_assembly_items") or []: + sa_row = frappe._dict(sa_row) + if sa_row.type_of_manufacturing != "Material Request": + continue + po_items.append( + frappe._dict( + { + "item_code": sa_row.production_item, + "required_qty": sa_row.qty, + "include_exploded_items": 0, + } + ) + ) + return po_items + + +def _validate_po_items(po_items): + if not po_items or not [row.get("item_code") for row in po_items if row.get("item_code")]: + frappe.throw( + _("Items to Manufacture are required to pull the Raw Materials associated with it."), + title=_("Items Required"), + ) + + +def _effective_ignore_ordered_qty(doc, po_items): + if doc.get("ignore_existing_ordered_qty"): + return doc.get("ignore_existing_ordered_qty") + return any(data.get("ignore_existing_ordered_qty") for data in po_items) + + +def _build_sub_assembly_map(doc): + if not (doc.get("skip_available_sub_assembly_item") and doc.get("sub_assembly_items")): + return {} + + sub_assembly_items = defaultdict(int) + for d in doc.get("sub_assembly_items"): + key = (d.get("production_item"), d.get("bom_no"), d.get("type_of_manufacturing")) + sub_assembly_items[key] += d.get("qty") + return {k[:2]: v for k, v in sub_assembly_items.items()} + + +def _collect_item_details(doc, po_items): + company = doc.get("company") + sub_assembly_items = _build_sub_assembly_map(doc) + existing_sub_assembly_items = set() + so_item_details = frappe._dict() + qty_precision = frappe.get_precision("Material Request Plan Item", "quantity") + + for data in po_items: + if not data.get("include_exploded_items") and doc.get("sub_assembly_items"): + data["include_exploded_items"] = 1 + item_details = _item_details_for_row( + doc, data, company, sub_assembly_items, existing_sub_assembly_items + ) + _accumulate_so_items(so_item_details, data.get("sales_order"), item_details, qty_precision) + return so_item_details + + +def _item_details_for_row(doc, data, company, sub_assembly_items, existing_sub_assembly_items): + planned_qty = data.get("required_qty") or data.get("planned_qty") + if data.get("bom") or data.get("bom_no"): + return _bom_item_details( + doc, data, company, planned_qty, sub_assembly_items, existing_sub_assembly_items + ) + if data.get("item_code"): + return _plain_item_details(doc, data, planned_qty) + return {} + + +def _bom_item_details(doc, data, company, planned_qty, sub_assembly_items, existing_sub_assembly_items): + bom_no, include_non_stock_items, include_subcontracted_items = _bom_explosion_flags(doc, data) + if not planned_qty: + frappe.throw(_("For row {0}: Enter Planned Qty").format(data.get("idx"))) + if not bom_no: + return {} + return _explode_bom_items( + doc, + data, + company, + bom_no, + planned_qty, + include_non_stock_items, + include_subcontracted_items, + sub_assembly_items, + existing_sub_assembly_items, + ) + + +def _bom_explosion_flags(doc, data): + if data.get("required_qty"): + include_subcontracted_items = 1 if data.get("include_exploded_items") else 0 + return data.get("bom"), 1, include_subcontracted_items + return data.get("bom_no"), doc.get("include_non_stock_items"), doc.get("include_subcontracted_items") + + +def _explode_bom_items( + doc, + data, + company, + bom_no, + planned_qty, + include_non_stock_items, + include_subcontracted_items, + sub_assembly_items, + existing_sub_assembly_items, +): + item_details = {} + if ( + data.get("include_exploded_items") + and doc.get("skip_available_sub_assembly_item") + and doc.get("sub_assembly_items") + ): + return get_raw_materials_of_sub_assembly_items( + existing_sub_assembly_items, + item_details, + company, + bom_no, + include_non_stock_items, + sub_assembly_items, + planned_qty=planned_qty, + ) + if data.get("include_exploded_items") and include_subcontracted_items: + return get_exploded_items( + item_details, company, bom_no, include_non_stock_items, planned_qty=planned_qty, doc=doc + ) + return get_subitems( + doc, + data, + item_details, + bom_no, + company, + include_non_stock_items, + include_subcontracted_items, + 1, + planned_qty=planned_qty, + ) + + +def _plain_item_details(doc, data, planned_qty): + item_master = frappe.get_doc("Item", data["item_code"]).as_dict() + purchase_uom = item_master.purchase_uom or item_master.stock_uom + conversion_factor = ( + get_uom_conversion_factor(item_master.name, purchase_uom) if item_master.purchase_uom else 1.0 + ) + return { + item_master.item_code: frappe._dict( + { + "item_name": item_master.item_name, + "default_bom": doc.bom, + "purchase_uom": purchase_uom, + "default_warehouse": item_master.default_warehouse, + "min_order_qty": item_master.min_order_qty, + "default_material_request_type": item_master.default_material_request_type, + "qty": planned_qty or 1, + "is_sub_contracted": item_master.is_sub_contracted_item, + "item_code": item_master.name, + "description": item_master.description, + "stock_uom": item_master.stock_uom, + "conversion_factor": conversion_factor, + "safety_stock": item_master.safety_stock, + } + ) + } + + +def _accumulate_so_items(so_item_details, sales_order, item_details, qty_precision): + for key, details in item_details.items(): + details.qty = flt(details.qty, qty_precision) + so_item_details.setdefault(sales_order, frappe._dict()) + if key in so_item_details[sales_order]: + existing = so_item_details[sales_order][key] + existing["qty"] = existing.get("qty", 0) + flt(details.qty) + else: + so_item_details[sales_order][key] = details + + +def _build_mr_items(doc, so_item_details, ignore_ordered_qty): + mr_items = [] + consumed_qty = defaultdict(float) + # raw_material_group_warehouse (optional, group) only widens the availability + # scope to its child warehouses; material is still received into for_warehouse. + target_warehouse = doc.get("for_warehouse") + scope_warehouse = doc.get("raw_material_group_warehouse") or target_warehouse + company = doc.get("company") + include_safety_stock = doc.get("include_safety_stock") + + for sales_order, item_dict in so_item_details.items(): + for details in item_dict.values(): + fallback = details.get("source_warehouse") or details.get("default_warehouse") + scope_warehouse = scope_warehouse or fallback + target_warehouse = target_warehouse or fallback + row = _mr_item_for_details( + doc, + details, + sales_order, + company, + ignore_ordered_qty, + include_safety_stock, + scope_warehouse, + target_warehouse, + consumed_qty, + ) + if row: + mr_items.append(row) + return mr_items + + +def _mr_item_for_details( + doc, + details, + sales_order, + company, + ignore_ordered_qty, + include_safety_stock, + warehouse, + target_warehouse, + consumed_qty, +): + # get_bin_details scopes to the warehouse's descendants, returning one row per + # child warehouse; sum them so a group warehouse reflects combined child stock. + bin_dict = _aggregate_bin_details(get_bin_details(details, doc.company, warehouse)) + if details.qty <= 0: + return None + return get_material_request_items( + doc, + details, + sales_order, + company, + ignore_ordered_qty, + include_safety_stock, + warehouse, + target_warehouse, + bin_dict, + consumed_qty, + ) + + +def _aggregate_bin_details(bin_list): + qty_fields = ( + "projected_qty", + "actual_qty", + "ordered_qty", + "reserved_qty_for_production", + "planned_qty", + ) + aggregated = {field: 0 for field in qty_fields} + for row in bin_list or []: + for field in qty_fields: + aggregated[field] += flt(row.get(field)) + return aggregated + + +def _apply_other_locations(doc, mr_items, warehouses, ignore_ordered_qty, get_parent_warehouse_data): + if not ((ignore_ordered_qty or get_parent_warehouse_data) and warehouses): + return mr_items + + new_mr_items = [] + for item in mr_items: + get_materials_from_other_locations( + item, + warehouses, + new_mr_items, + doc.get("company"), + consider_minimum_order_qty=doc.get("consider_minimum_order_qty"), + ) + return new_mr_items + + +def _warn_no_mr_items(doc): + to_enable = frappe.bold(frappe.get_meta("Production Plan").get_field("ignore_existing_ordered_qty").label) + warehouse = frappe.bold(doc.get("for_warehouse")) + message = ( + _( + "As there are sufficient raw materials, Material Request is not required for Warehouse {0}." + ).format(warehouse) + + "

" + ) + message += _("If you still want to proceed, please enable {0}.").format(to_enable) + frappe.msgprint(message, title=_("Note")) + + +def get_material_request_items( + doc, + row, + sales_order, + company, + ignore_existing_ordered_qty, + include_safety_stock, + warehouse, + target_warehouse, + bin_dict, + consumed_qty, +): + required_qty = _required_qty_for_mr( + doc, row, ignore_existing_ordered_qty, warehouse, bin_dict, consumed_qty, include_safety_stock + ) + item_group_defaults = get_item_group_defaults(row.item_code, company) + conversion_factor = _mr_purchase_conversion_factor(row) + return _material_request_item_row( + row, sales_order, target_warehouse, bin_dict, required_qty, conversion_factor, item_group_defaults + ) + + +def _required_qty_for_mr( + doc, 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")) + + if not ignore_existing_ordered_qty or bin_dict.get("projected_qty", 0) < 0: + required_qty = _apply_minimum_order_qty(doc, row, qty + safety_stock) + return _adjust_required_qty_for_uom(row, required_qty) + + key = (row.get("item_code"), warehouse) + available_qty = flt(bin_dict.get("projected_qty", 0)) - consumed_qty[key] + required_qty = _apply_minimum_order_qty(doc, row, max(0, qty - (available_qty - safety_stock))) + required_qty = _adjust_required_qty_for_uom(row, required_qty) + consumed_qty[key] += qty - required_qty + return required_qty + + +def _apply_minimum_order_qty(doc, row, required_qty): + if doc.get("consider_minimum_order_qty") and 0 < required_qty < row["min_order_qty"]: + return row["min_order_qty"] + return 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 + + +def _mr_purchase_conversion_factor(row): + item_details = frappe.get_cached_value("Item", row.item_code, ["purchase_uom", "stock_uom"], as_dict=1) + if ( + row.get("default_material_request_type") == "Purchase" + and item_details.purchase_uom + and item_details.purchase_uom != item_details.stock_uom + ): + return get_conversion_factor(row.item_code, item_details.purchase_uom).get("conversion_factor") or 1.0 + return 1.0 + + +def _material_request_item_row( + row, sales_order, warehouse, bin_dict, required_qty, conversion_factor, item_group_defaults +): + warehouse = ( + warehouse + or row.get("source_warehouse") + or row.get("default_warehouse") + or item_group_defaults.get("default_warehouse") + ) + return { + "item_code": row.item_code, + "item_name": row.item_name, + "quantity": required_qty / conversion_factor, + "conversion_factor": conversion_factor, + "required_bom_qty": row.get("qty"), + "stock_uom": row.get("stock_uom"), + "warehouse": warehouse, + "safety_stock": row.safety_stock, + "actual_qty": bin_dict.get("actual_qty", 0), + "projected_qty": bin_dict.get("projected_qty", 0), + "ordered_qty": bin_dict.get("ordered_qty", 0), + "reserved_qty_for_production": bin_dict.get("reserved_qty_for_production", 0), + "min_order_qty": row["min_order_qty"], + "material_request_type": row.get("default_material_request_type"), + "sales_order": sales_order, + "description": row.get("description"), + "uom": row.get("purchase_uom") or row.get("stock_uom"), + "main_item_code": row.get("main_bom_item"), + "from_bom": row.get("main_bom"), + } + + +def get_materials_from_other_locations( + item, warehouses, new_mr_items, company, consider_minimum_order_qty=False +): + from erpnext.stock.doctype.pick_list.pick_list import get_available_item_locations + + locations = get_available_item_locations( + item.get("item_code"), + warehouses, + item.get("quantity") * item.get("conversion_factor"), + company, + ignore_validation=True, + ) + + 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") + + required_qty = _transfer_from_locations(item, locations, new_mr_items, required_qty) + _add_remaining_purchase_request(item, new_mr_items, required_qty, consider_minimum_order_qty) + + +def _transfer_from_locations(item, locations, new_mr_items, required_qty): + # get available material by transferring to production warehouse + for d in locations: + if required_qty <= 0: + return required_qty + + 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) + return required_qty + + +def _add_remaining_purchase_request(item, new_mr_items, required_qty, consider_minimum_order_qty=False): + # raise purchase request for remaining qty + precision = frappe.get_precision("Material Request Plan Item", "quantity") + if flt(required_qty, precision) <= 0: + return + + if consider_minimum_order_qty: + required_qty = max(required_qty, flt(item.get("min_order_qty"))) + + purchase_uom = frappe.db.get_value("Item", item.get("item_code"), "purchase_uom") + 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") + new_mr_items.append(item) + + +@frappe.whitelist() +def download_raw_materials(doc: str | dict | Document, warehouses: str | list | None = None): + frappe.has_permission("Production Plan", "read", throw=True) + + doc = _normalize_mr_doc(doc) + item_list = [_raw_materials_header()] + + doc.warehouse = None + frappe.flags.show_qty_in_stock_uom = 1 + items = get_items_for_material_requests(doc, warehouses=warehouses, get_parent_warehouse_data=True) + + _build_download_rows(doc, items, item_list) + build_csv_response(item_list, doc.name) + + +def _raw_materials_header(): + return [ + "Item Code", + "Item Name", + "Description", + "Stock UOM", + "Warehouse", + "Required Qty as per BOM", + "Projected Qty", + "Available Qty In Hand", + "Ordered Qty", + "Planned Qty", + "Reserved Qty for Production", + "Safety Stock", + "Required Qty", + ] + + +def _build_download_rows(doc, items, item_list): + duplicate_item_wh_list = frappe._dict() + for d in items: + key = (d.get("item_code"), d.get("warehouse")) + if key in duplicate_item_wh_list: + duplicate_item_wh_list[key][12] += d.get("quantity") + continue + + rm_data = _raw_material_row(d) + duplicate_item_wh_list[key] = rm_data + item_list.append(rm_data) + + if not doc.get("for_warehouse"): + _append_other_warehouse_bins(item_list, d, doc) + + +def _raw_material_row(d): + return [ + d.get("item_code"), + d.get("item_name"), + d.get("description"), + d.get("stock_uom"), + d.get("warehouse"), + d.get("required_bom_qty"), + d.get("projected_qty"), + d.get("actual_qty"), + d.get("ordered_qty"), + d.get("planned_qty"), + d.get("reserved_qty_for_production"), + d.get("safety_stock"), + d.get("quantity"), + ] + + +def _append_other_warehouse_bins(item_list, d, doc): + row = {"item_code": d.get("item_code")} + for bin_dict in get_bin_details(row, doc.company, all_warehouse=True): + if d.get("warehouse") == bin_dict.get("warehouse"): + continue + + item_list.append( + [ + "", + "", + "", + bin_dict.get("warehouse"), + "", + bin_dict.get("projected_qty", 0), + bin_dict.get("actual_qty", 0), + bin_dict.get("ordered_qty", 0), + bin_dict.get("reserved_qty_for_production", 0), + ] + )