fix: consumed operation cost calculation (backport #54858) (#55132)

Co-authored-by: Mihir Kandoi <kandoimihir@gmail.com>
fix: consumed operation cost calculation (#54858)
This commit is contained in:
mergify[bot]
2026-05-26 10:14:19 +00:00
committed by GitHub
parent 937eb87932
commit 46d5395148
11 changed files with 555 additions and 54 deletions

View File

@@ -240,10 +240,10 @@ class StatusUpdater(Document):
# get unique transactions to update
for d in self.get_all_children():
if hasattr(d, "qty") and d.qty < 0 and not self.get("is_return"):
if hasattr(d, "qty") and flt(d.qty) < 0 and not self.get("is_return"):
frappe.throw(_("For an item {0}, quantity must be positive number").format(d.item_code))
if hasattr(d, "qty") and d.qty > 0 and self.get("is_return"):
if hasattr(d, "qty") and flt(d.qty) > 0 and self.get("is_return"):
frappe.throw(_("For an item {0}, quantity must be negative number").format(d.item_code))
if not frappe.db.get_single_value("Selling Settings", "allow_negative_rates_for_items"):

View File

@@ -1367,18 +1367,71 @@ def add_non_stock_items_cost(stock_entry, work_order, expense_account):
def add_operations_cost(stock_entry, work_order=None, expense_account=None):
from erpnext.stock.doctype.stock_entry.stock_entry import get_operating_cost_per_unit
from erpnext.stock.doctype.stock_entry.stock_entry import (
get_consumed_operating_cost,
get_operating_cost_per_unit,
)
operating_cost_per_unit = get_operating_cost_per_unit(work_order, stock_entry.bom_no)
if operating_cost_per_unit:
stock_entry.append(
"additional_costs",
{
def append_operating_cost(amount, operation=None, qty=None):
if amount:
row = {
"expense_account": expense_account,
"description": _("Operating Cost as per Work Order / BOM"),
"amount": operating_cost_per_unit * flt(stock_entry.fg_completed_qty),
},
"amount": flt(
amount,
frappe.get_precision("Landed Cost Taxes and Charges", "amount"),
),
"has_operating_cost": 1,
}
if operation:
row["operation_id"] = operation.name
if qty is not None:
row["qty"] = qty
stock_entry.append(
"additional_costs",
row,
)
if (
work_order
and stock_entry.bom_no
and frappe.db.get_single_value("Manufacturing Settings", "set_op_cost_and_scrap_from_sub_assemblies")
and work_order.get("use_multi_level_bom")
):
operating_cost_per_unit = get_operating_cost_per_unit(work_order, stock_entry.bom_no)
append_operating_cost(
operating_cost_per_unit * flt(stock_entry.fg_completed_qty),
qty=flt(stock_entry.fg_completed_qty),
)
elif work_order and work_order.get("operations"):
for operation in work_order.get("operations"):
qty = flt(stock_entry.fg_completed_qty)
amount = 0
if flt(operation.completed_qty):
consumed_cost = get_consumed_operating_cost(
work_order.name, stock_entry.bom_no, operation.name
)
remaining_cost = flt(
flt(operation.actual_operating_cost) - flt(consumed_cost.get("consumed_cost")),
operation.precision("actual_operating_cost"),
)
remaining_qty = flt(operation.completed_qty) - flt(consumed_cost.get("consumed_qty"))
if remaining_cost <= 0 or remaining_qty <= 0:
continue
qty = min(remaining_qty, flt(stock_entry.fg_completed_qty))
amount = remaining_cost / remaining_qty * qty
elif work_order.qty:
amount = flt(operation.planned_operating_cost) / flt(work_order.qty) * qty
append_operating_cost(amount, operation=operation, qty=qty)
else:
operating_cost_per_unit = get_operating_cost_per_unit(work_order, stock_entry.bom_no)
append_operating_cost(
operating_cost_per_unit * flt(stock_entry.fg_completed_qty),
qty=flt(stock_entry.fg_completed_qty),
)
if work_order and work_order.additional_operating_cost and work_order.qty:

View File

@@ -1,5 +1,6 @@
{
"actions": [],
"allow_bulk_edit": 1,
"autoname": "naming_series:",
"creation": "2018-07-09 17:23:29.518745",
"doctype": "DocType",
@@ -135,6 +136,7 @@
"fieldname": "wip_warehouse",
"fieldtype": "Link",
"label": "WIP Warehouse",
"link_filters": "[[\"Warehouse\",\"disabled\",\"=\",0],[\"Warehouse\",\"is_group\",\"=\",0]]",
"options": "Warehouse",
"reqd": 1
},
@@ -511,7 +513,7 @@
],
"is_submittable": 1,
"links": [],
"modified": "2025-08-04 15:47:54.514290",
"modified": "2026-05-12 12:17:17.750857",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "Job Card",

View File

@@ -7,7 +7,7 @@ from typing import Literal
import frappe
from frappe.test_runner import make_test_records
from frappe.tests.utils import FrappeTestCase, change_settings
from frappe.utils import random_string
from frappe.utils import flt, random_string
from frappe.utils.data import add_to_date, now, today
from erpnext.manufacturing.doctype.job_card.job_card import (
@@ -697,6 +697,403 @@ class TestJobCard(FrappeTestCase):
self.assertEqual(wo_doc.process_loss_qty, 2)
self.assertEqual(wo_doc.status, "Completed")
def test_op_cost_calculation(self):
from erpnext.manufacturing.doctype.routing.test_routing import (
create_routing,
setup_bom,
setup_operations,
)
from erpnext.manufacturing.doctype.work_order.work_order import make_job_card
from erpnext.manufacturing.doctype.work_order.work_order import (
make_stock_entry as make_stock_entry_for_wo,
)
from erpnext.stock.doctype.item.test_item import make_item
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
suffix = random_string(5)
workstation = make_workstation(
workstation_name=f"Test Workstation Z {suffix}", hour_rate_rent=240, hour_rate_labour=0
)
workstation.update(
{
"hour_rate_rent": 240,
"hour_rate_labour": 0,
"hour_rate_electricity": 0,
"hour_rate_consumable": 0,
}
)
workstation.save()
operations = [
{
"operation": f"Test Operation A1 {suffix}",
"workstation": workstation.name,
"time_in_mins": 30,
},
]
warehouse = create_warehouse(f"Test Warehouse 123 for Job Card {suffix}")
setup_operations(operations)
item_code = f"Test Job Card Process Qty Item {suffix}"
for item in [item_code, item_code + "RM 1", item_code + "RM 2"]:
if not frappe.db.exists("Item", item):
make_item(
item,
{
"item_name": item,
"stock_uom": "Nos",
"is_stock_item": 1,
},
)
routing_doc = create_routing(routing_name="Testing Route", operations=operations)
bom_doc = setup_bom(
item_code=item_code,
routing=routing_doc.name,
raw_materials=[item_code + "RM 1", item_code + "RM 2"],
source_warehouse=warehouse,
)
for row in bom_doc.items:
make_stock_entry(
item_code=row.item_code,
target=row.source_warehouse,
qty=10,
basic_rate=100,
)
wo_doc = make_wo_order_test_record(
production_item=item_code,
bom_no=bom_doc.name,
qty=10,
skip_transfer=1,
wip_warehouse=warehouse,
source_warehouse=warehouse,
)
first_job_card = frappe.get_all(
"Job Card",
filters={"work_order": wo_doc.name, "sequence_id": 1},
fields=["name"],
order_by="sequence_id",
limit=1,
)[0].name
jc = frappe.get_doc("Job Card", first_job_card)
from_time = "2025-01-01 09:00:00"
for _ in jc.scheduled_time_logs:
jc.append(
"time_logs",
{
"from_time": from_time,
"to_time": add_to_date(from_time, minutes=1),
"completed_qty": 4,
},
)
jc.for_quantity = 4
jc.save()
jc.submit()
s1 = frappe.get_doc(make_stock_entry_for_wo(wo_doc.name, "Manufacture", 4))
s1.submit()
wo_doc.reload()
precision = s1.additional_costs[0].precision("amount")
self.assertEqual(
flt(s1.additional_costs[0].amount, precision),
flt(wo_doc.operations[0].actual_operating_cost, precision),
)
make_job_card(
wo_doc.name,
[
{
"name": wo_doc.operations[0].name,
"operation": operations[0]["operation"],
"workstation": wo_doc.operations[0].workstation,
"qty": 6,
"pending_qty": 6,
}
],
)
job_card = frappe.get_last_doc("Job Card", {"work_order": wo_doc.name})
from_time = "2025-01-01 10:00:00"
job_card.append(
"time_logs",
{
"from_time": from_time,
"to_time": add_to_date(from_time, minutes=2),
"completed_qty": 6,
},
)
job_card.for_quantity = 6
job_card.save()
job_card.submit()
s2 = frappe.get_doc(make_stock_entry_for_wo(wo_doc.name, "Manufacture", 6))
wo_doc.reload()
precision = s2.additional_costs[0].precision("amount")
self.assertEqual(
flt(s2.additional_costs[0].amount, precision),
flt(wo_doc.operations[0].actual_operating_cost - s1.additional_costs[0].amount, precision),
)
@change_settings("Manufacturing Settings", {"overproduction_percentage_for_work_order": 100})
def test_operating_cost_with_overproduction(self):
from erpnext.manufacturing.doctype.routing.test_routing import (
create_routing,
setup_bom,
setup_operations,
)
from erpnext.manufacturing.doctype.work_order.work_order import make_job_card
from erpnext.manufacturing.doctype.work_order.work_order import (
make_stock_entry as make_stock_entry_for_wo,
)
from erpnext.stock.doctype.item.test_item import make_item
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
suffix = random_string(5)
workstation = make_workstation(
workstation_name=f"Test Workstation for Overproduction {suffix}",
hour_rate_rent=10,
hour_rate_labour=10,
)
workstation.update(
{
"hour_rate_rent": 10,
"hour_rate_labour": 10,
"hour_rate_electricity": 0,
"hour_rate_consumable": 0,
}
)
workstation.save()
operations = [
{"operation": f"Test Operation 1 {suffix}", "workstation": workstation.name, "time_in_mins": 30},
{"operation": f"Test Operation 2 {suffix}", "workstation": workstation.name, "time_in_mins": 30},
]
warehouse = create_warehouse(f"Test Warehouse for Overproduction {suffix}")
setup_operations(operations)
fg = make_item(f"Test FG for Overproduction {suffix}", {"stock_uom": "Nos", "is_stock_item": 1})
rm = make_item(f"Test RM for Overproduction {suffix}", {"stock_uom": "Nos", "is_stock_item": 1})
routing_doc = create_routing(routing_name=f"Testing Route {suffix}", operations=operations)
bom_doc = setup_bom(
item_code=fg.name,
routing=routing_doc.name,
raw_materials=[rm.name],
source_warehouse=warehouse,
)
for row in bom_doc.items:
make_stock_entry(
item_code=row.item_code,
target=row.source_warehouse,
qty=100,
basic_rate=100,
)
wo_doc = make_wo_order_test_record(
production_item=fg.name,
bom_no=bom_doc.name,
qty=10,
skip_transfer=1,
source_warehouse=warehouse,
)
first_operation = frappe.get_all(
"Job Card",
filters={"work_order": wo_doc.name, "sequence_id": 1},
fields=["name"],
order_by="sequence_id",
limit=1,
)[0].name
jc = frappe.get_doc("Job Card", first_operation)
from_time = "2025-01-02 09:00:00"
for _ in jc.scheduled_time_logs:
jc.append(
"time_logs",
{
"from_time": from_time,
"to_time": add_to_date(from_time, days=1),
"completed_qty": 4,
},
)
jc.for_quantity = 4
jc.save()
jc.submit()
second_operation = frappe.get_all(
"Job Card",
filters={"work_order": wo_doc.name, "sequence_id": 2},
fields=["name"],
order_by="sequence_id",
limit=1,
)[0].name
jc = frappe.get_doc("Job Card", second_operation)
from_time = "2025-01-05 09:00:00"
for _ in jc.scheduled_time_logs:
jc.append(
"time_logs",
{
"from_time": from_time,
"to_time": add_to_date(from_time, days=2),
"completed_qty": 4,
},
)
jc.for_quantity = 4
jc.save()
jc.submit()
s = frappe.get_doc(make_stock_entry_for_wo(wo_doc.name, "Manufacture", 6)) # overproduction
s.submit()
def assert_operating_costs(stock_entry, qty, previous_entries):
wo_doc.reload()
for idx, operation in enumerate(wo_doc.operations):
consumed_cost = sum(
entry.additional_costs[idx].amount for entry in previous_entries if entry.docstatus == 1
)
consumed_qty = sum(
entry.additional_costs[idx].qty for entry in previous_entries if entry.docstatus == 1
)
remaining_cost = operation.actual_operating_cost - consumed_cost
remaining_qty = operation.completed_qty - consumed_qty
precision = stock_entry.additional_costs[idx].precision("amount")
expected_cost = flt(remaining_cost / remaining_qty * min(remaining_qty, qty), precision)
self.assertEqual(flt(stock_entry.additional_costs[idx].amount, precision), expected_cost)
assert_operating_costs(s, 6, [])
make_job_card(
wo_doc.name,
[
{
"name": wo_doc.operations[0].name,
"operation": operations[0]["operation"],
"workstation": wo_doc.operations[0].workstation,
"qty": 2,
"pending_qty": 2,
}
],
)
job_card = frappe.get_last_doc("Job Card", {"work_order": wo_doc.name})
from_time = "2025-01-09 09:00:00"
job_card.append(
"time_logs",
{
"from_time": from_time,
"to_time": add_to_date(from_time, days=1),
"completed_qty": 2,
},
)
job_card.for_quantity = 2
job_card.save()
job_card.submit()
make_job_card(
wo_doc.name,
[
{
"name": wo_doc.operations[1].name,
"operation": operations[1]["operation"],
"workstation": wo_doc.operations[1].workstation,
"qty": 2,
"pending_qty": 2,
}
],
)
job_card = frappe.get_last_doc("Job Card", {"work_order": wo_doc.name})
from_time = "2025-01-12 09:00:00"
job_card.append(
"time_logs",
{
"from_time": from_time,
"to_time": add_to_date(from_time, days=2),
"completed_qty": 2,
},
)
job_card.for_quantity = 2
job_card.save()
job_card.submit()
s2 = frappe.get_doc(make_stock_entry_for_wo(wo_doc.name, "Manufacture", 1))
s2.submit()
assert_operating_costs(s2, 1, [s])
make_job_card(
wo_doc.name,
[
{
"name": wo_doc.operations[0].name,
"operation": operations[0]["operation"],
"workstation": wo_doc.operations[0].workstation,
"qty": 2,
"pending_qty": 2,
}
],
)
job_card = frappe.get_last_doc("Job Card", {"work_order": wo_doc.name})
from_time = "2025-01-16 09:00:00"
job_card.append(
"time_logs",
{
"from_time": from_time,
"to_time": add_to_date(from_time, days=1),
"completed_qty": 2,
},
)
job_card.for_quantity = 2
job_card.save()
job_card.submit()
make_job_card(
wo_doc.name,
[
{
"name": wo_doc.operations[1].name,
"operation": operations[1]["operation"],
"workstation": wo_doc.operations[1].workstation,
"qty": 2,
"pending_qty": 2,
}
],
)
job_card = frappe.get_last_doc("Job Card", {"work_order": wo_doc.name})
from_time = "2025-01-19 09:00:00"
job_card.append(
"time_logs",
{
"from_time": from_time,
"to_time": add_to_date(from_time, days=2),
"completed_qty": 2,
},
)
job_card.for_quantity = 2
job_card.save()
job_card.submit()
s3 = frappe.get_doc(make_stock_entry_for_wo(wo_doc.name, "Manufacture", 2))
s3.submit()
assert_operating_costs(s3, 2, [s, s2])
s2.cancel()
s4 = frappe.get_doc(make_stock_entry_for_wo(wo_doc.name, "Manufacture", 3))
s4.submit()
assert_operating_costs(s4, 3, [s, s3])
def create_bom_with_multiple_operations():
"Create a BOM with multiple operations and Material Transfer against Job Card"

View File

@@ -1,5 +1,6 @@
{
"actions": [],
"allow_bulk_edit": 1,
"creation": "2018-07-09 17:20:44.737289",
"doctype": "DocType",
"editable_grid": 1,
@@ -33,6 +34,7 @@
"ignore_user_permissions": 1,
"in_list_view": 1,
"label": "Source Warehouse",
"link_filters": "[[\"Warehouse\",\"disabled\",\"=\",0],[\"Warehouse\",\"is_group\",\"=\",0]]",
"options": "Warehouse"
},
{
@@ -105,7 +107,7 @@
"index_web_pages_for_search": 1,
"istable": 1,
"links": [],
"modified": "2021-04-22 18:50:00.003444",
"modified": "2026-05-12 12:22:18.506904",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "Job Card Item",
@@ -115,4 +117,4 @@
"sort_field": "modified",
"sort_order": "DESC",
"track_changes": 1
}
}

View File

@@ -12,21 +12,10 @@ frappe.ui.form.on("Work Order", {
frm.ignore_doctypes_on_cancel_all = ["Serial and Batch Bundle"];
// Set query for warehouses
frm.set_query("wip_warehouse", function () {
return {
filters: {
company: frm.doc.company,
},
};
});
frm.set_query("source_warehouse", function () {
return {
filters: {
company: frm.doc.company,
},
};
});
frm.events.set_company_filters(frm, "wip_warehouse");
frm.events.set_company_filters(frm, "source_warehouse");
frm.events.set_company_filters(frm, "fg_warehouse");
frm.events.set_company_filters(frm, "scrap_warehouse");
frm.set_query("source_warehouse", "required_items", function () {
return {
@@ -44,24 +33,6 @@ frappe.ui.form.on("Work Order", {
};
});
frm.set_query("fg_warehouse", function () {
return {
filters: {
company: frm.doc.company,
is_group: 0,
},
};
});
frm.set_query("scrap_warehouse", function () {
return {
filters: {
company: frm.doc.company,
is_group: 0,
},
};
});
// Set query for BOM
frm.set_query("bom_no", function () {
if (frm.doc.production_item) {
@@ -118,6 +89,16 @@ frappe.ui.form.on("Work Order", {
});
},
set_company_filters(frm, fieldname) {
frm.set_query(fieldname, () => {
return {
filters: {
company: frm.doc.company,
},
};
});
},
onload: function (frm) {
if (!frm.doc.status) frm.doc.status = "Draft";
@@ -315,7 +296,7 @@ frappe.ui.form.on("Work Order", {
{
fieldtype: "Data",
fieldname: "name",
label: __("Operation Id"),
label: __("Operation ID"),
},
{
fieldtype: "Float",
@@ -385,6 +366,7 @@ frappe.ui.form.on("Work Order", {
if (pending_qty) {
dialog.fields_dict.operations.df.data.push({
__checked: 1,
name: data.name,
operation: data.operation,
workstation: data.workstation,

View File

@@ -1,5 +1,6 @@
{
"actions": [],
"allow_bulk_edit": 1,
"allow_import": 1,
"autoname": "naming_series:",
"creation": "2025-04-09 12:09:40.634472",
@@ -249,6 +250,7 @@
"fieldname": "wip_warehouse",
"fieldtype": "Link",
"label": "Work-in-Progress Warehouse",
"link_filters": "[[\"Warehouse\",\"disabled\",\"=\",0],[\"Warehouse\",\"is_group\",\"=\",0]]",
"mandatory_depends_on": "eval:!doc.skip_transfer || doc.from_wip_warehouse",
"options": "Warehouse"
},
@@ -257,6 +259,7 @@
"fieldname": "fg_warehouse",
"fieldtype": "Link",
"label": "Target Warehouse",
"link_filters": "[[\"Warehouse\",\"disabled\",\"=\",0],[\"Warehouse\",\"is_group\",\"=\",0]]",
"options": "Warehouse",
"reqd": 1
},
@@ -269,6 +272,7 @@
"fieldname": "scrap_warehouse",
"fieldtype": "Link",
"label": "Scrap Warehouse",
"link_filters": "[[\"Warehouse\",\"disabled\",\"=\",0],[\"Warehouse\",\"is_group\",\"=\",0]]",
"options": "Warehouse"
},
{
@@ -498,6 +502,7 @@
"fieldname": "source_warehouse",
"fieldtype": "Link",
"label": "Source Warehouse",
"link_filters": "[[\"Warehouse\",\"disabled\",\"=\",0],[\"Warehouse\",\"is_group\",\"=\",0]]",
"options": "Warehouse"
},
{
@@ -602,7 +607,7 @@
"image_field": "image",
"is_submittable": 1,
"links": [],
"modified": "2025-06-21 00:55:45.916224",
"modified": "2026-05-19 12:20:38.102403",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "Work Order",

View File

@@ -1,5 +1,6 @@
{
"actions": [],
"allow_bulk_edit": 1,
"creation": "2016-04-18 07:38:26.314642",
"doctype": "DocType",
"editable_grid": 1,
@@ -46,6 +47,7 @@
"ignore_user_permissions": 1,
"in_list_view": 1,
"label": "Source Warehouse",
"link_filters": "[[\"Warehouse\",\"disabled\",\"=\",0],[\"Warehouse\",\"is_group\",\"=\",0]]",
"options": "Warehouse"
},
{
@@ -151,7 +153,7 @@
],
"istable": 1,
"links": [],
"modified": "2025-12-02 11:16:05.081613",
"modified": "2026-05-12 12:05:16.687866",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "Work Order Item",

View File

@@ -1,5 +1,6 @@
{
"actions": [],
"allow_bulk_edit": 1,
"creation": "2014-07-11 11:51:00.453717",
"doctype": "DocType",
"editable_grid": 1,
@@ -12,7 +13,10 @@
"col_break3",
"amount",
"base_amount",
"has_corrective_cost"
"has_corrective_cost",
"has_operating_cost",
"operation_id",
"qty"
],
"fields": [
{
@@ -70,12 +74,36 @@
"fieldtype": "Check",
"label": "Has Corrective Cost",
"read_only": 1
},
{
"default": "0",
"fieldname": "has_operating_cost",
"fieldtype": "Check",
"label": "Has Operating Cost",
"read_only": 1
},
{
"fieldname": "operation_id",
"fieldtype": "Data",
"hidden": 1,
"label": "Operation ID",
"no_copy": 1,
"read_only": 1
},
{
"fieldname": "qty",
"fieldtype": "Float",
"hidden": 1,
"label": "Qty",
"no_copy": 1,
"non_negative": 1,
"read_only": 1
}
],
"index_web_pages_for_search": 1,
"istable": 1,
"links": [],
"modified": "2025-01-20 12:22:03.455762",
"modified": "2026-05-19 12:21:07.953801",
"modified_by": "Administrator",
"module": "Stock",
"name": "Landed Cost Taxes and Charges",
@@ -83,4 +111,4 @@
"permissions": [],
"sort_field": "modified",
"sort_order": "DESC"
}
}

View File

@@ -21,9 +21,12 @@ class LandedCostTaxesandCharges(Document):
exchange_rate: DF.Float
expense_account: DF.Link | None
has_corrective_cost: DF.Check
has_operating_cost: DF.Check
operation_id: DF.Data | None
parent: DF.Data
parentfield: DF.Data
parenttype: DF.Data
qty: DF.Float
# end: auto-generated types
pass

View File

@@ -3370,6 +3370,33 @@ def get_work_order_details(work_order, company):
}
def get_consumed_operating_cost(work_order, bom_no, operation_id=None):
table = frappe.qb.DocType("Stock Entry")
child_table = frappe.qb.DocType("Landed Cost Taxes and Charges")
query = (
frappe.qb.from_(child_table)
.join(table)
.on(child_table.parent == table.name)
.select(
Sum(child_table.amount).as_("consumed_cost"),
Sum(child_table.qty).as_("consumed_qty"),
)
.where(
(table.docstatus == 1)
& (table.work_order == work_order)
& (table.purpose == "Manufacture")
& (table.bom_no == bom_no)
& (child_table.has_operating_cost == 1)
)
)
if operation_id:
query = query.where(child_table.operation_id == operation_id)
data = query.run(as_dict=True)
return data[0] if data else frappe._dict()
def get_operating_cost_per_unit(work_order=None, bom_no=None):
operating_cost_per_unit = 0
if work_order: