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

chore: resolve conflict

Browse Source
This commit is contained in:
Mihir Kandoi
2026-08-09 22:56:02 +05:30
parent eb90186d3f 6e4a7fba7e
commit 8310b56f32
15 changed files with 1248 additions and 3040 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
erpnext/manufacturing/doctype/bom/bom.py
View File

@@ -282,6 +282,7 @@ class BOM(WebsiteGenerator):
self.clear_inspection()
self.validate_main_item()
self.validate_currency()
self.set_operation_finished_goods()
self.set_materials_based_on_operation_bom()
self.set_conversion_rate()
self.set_plc_conversion_rate()
@@ -307,15 +308,42 @@ class BOM(WebsiteGenerator):
if self.docstatus == 1:
self.validate_raw_materials_of_operation()
def set_operation_finished_goods(self):
"""Fill each operation's FG item where it is unambiguous: the final operation produces
this BOM's item, an operation with a BOM produces that BOM's item. Runs before
set_materials_based_on_operation_bom so derived rows get their materials expanded."""
if not self.track_semi_finished_goods:
return
for row in self.operations:
if row.is_final_finished_good and not row.finished_good:
row.finished_good = self.item
elif row.bom_no and not row.finished_good:
row.finished_good = frappe.get_cached_value("BOM", row.bom_no, "item")
def validate_semi_finished_goods(self):
if not self.track_semi_finished_goods or not self.operations:
return
fg_items = []
for row in self.operations:
if not row.finished_good:
frappe.throw(
_(
"Row #{0}: FG / Semi FG Item is required for the operation {1} as 'Track Semi Finished Goods' is enabled."
).format(row.idx, bold(row.operation)),
)
if not row.is_final_finished_good:
continue
if row.finished_good != self.item:
frappe.throw(
_(
"Row #{0}: The operation {1} has 'Is Final Finished Good' checked, so its FG / Semi FG Item must be {2}."
).format(row.idx, bold(row.operation), bold(self.item)),
)
fg_items.append(row.finished_good)
if not fg_items:
@@ -826,7 +854,7 @@ class BOM(WebsiteGenerator):
self.add_materials_from_bom(row.finished_good, row.bom_no, row.idx, qty=row.finished_good_qty)
@frappe.whitelist()
def add_raw_materials(self, operation_row_id, items):
def add_raw_materials(self, operation_row_id: str | int, items: str | list[dict]) -> None:
if isinstance(items, str):
items = parse_json(items)
@@ -836,17 +864,10 @@ class BOM(WebsiteGenerator):
row.update(get_item_details(row.get("item_code")))
row.operation_row_id = operation_row_id
item_row = None
if row.name:
item_row = self.get_item_data(row.name)
item_row = self.get_item_data(row.item_code, operation_row_id)
if item_row:
item_row.update(
{
"item_code": row.get("item_code"),
"qty": row.get("qty"),
}
)
item_row.qty = row.get("qty")
else:
row.idx = None
row.name = None
@@ -867,9 +888,9 @@ class BOM(WebsiteGenerator):
return False
def get_item_data(self, name):
def get_item_data(self, item_code, operation_row_id):
for row in self.items:
if row.item_code == name:
if row.item_code == item_code and cint(row.operation_row_id) == cint(operation_row_id):
return row
@frappe.whitelist()

203
erpnext/manufacturing/doctype/bom/test_bom.py
View File

@@ -7,7 +7,7 @@ from functools import partial
import frappe
from frappe.tests import timeout
from frappe.utils import cstr, flt
from frappe.utils import cint, cstr, flt
from erpnext.controllers.tests.test_subcontracting_controller import (
set_backflush_based_on,
@@ -811,6 +811,207 @@ class TestBOM(ERPNextTestSuite):
for row in bom.items:
self.assertEqual(row.stock_uom, "Kg")
@timeout
def test_track_semi_finished_goods_requires_finished_good_on_operations(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
fg_item = make_item(properties={"is_stock_item": 1}).name
sfg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100.0}).name
make_workstation({"workstation": "_Test SFG Workstation"})
for operation in ("_Test SFG Operation", "_Test SFG Final Operation"):
make_operation({"operation": operation, "workstation": "_Test SFG Workstation"})
bom = frappe.new_doc("BOM")
bom.company = "_Test Company"
bom.item = fg_item
bom.quantity = 1
bom.with_operations = 1
bom.track_semi_finished_goods = 1
bom.append(
"operations",
{
"operation": "_Test SFG Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
},
)
bom.append(
"operations",
{
"operation": "_Test SFG Final Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"is_final_finished_good": 1,
},
)
bom.append("items", {"item_code": rm_item, "qty": 1, "operation_row_id": 1})
bom.append("items", {"item_code": sfg_item, "qty": 1, "operation_row_id": 2})
# the first operation produces nothing derivable: no FG item, no BOM to take it from
self.assertRaises(frappe.ValidationError, bom.insert)
bom.operations[0].finished_good = sfg_item
bom.insert()
# the final operation's FG item is derived from the BOM's own item
self.assertEqual(bom.operations[1].finished_good, fg_item)
@timeout
def test_add_raw_materials_when_item_is_used_by_another_operation(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
fg_item = make_item(properties={"is_stock_item": 1}).name
sfg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100.0}).name
make_workstation({"workstation": "_Test SFG Workstation"})
for operation in ("_Test SFG Operation", "_Test SFG Final Operation"):
make_operation({"operation": operation, "workstation": "_Test SFG Workstation"})
bom = frappe.new_doc("BOM")
bom.company = "_Test Company"
bom.item = fg_item
bom.quantity = 1
bom.with_operations = 1
bom.track_semi_finished_goods = 1
bom.append(
"operations",
{
"operation": "_Test SFG Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"finished_good": sfg_item,
},
)
bom.append(
"operations",
{
"operation": "_Test SFG Final Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"is_final_finished_good": 1,
},
)
bom.append("items", {"item_code": rm_item, "qty": 1, "operation_row_id": 1})
bom.append("items", {"item_code": sfg_item, "qty": 1, "operation_row_id": 2})
bom.insert()
def rows_for(item_code, operation_row_id):
return [
row
for row in bom.items
if row.item_code == item_code and cint(row.operation_row_id) == operation_row_id
]
# the item already used by operation 1 gets its own new row under operation 2
bom.add_raw_materials(2, [{"item_code": rm_item, "qty": 3}])
self.assertEqual(len(rows_for(rm_item, 2)), 1)
self.assertEqual(flt(rows_for(rm_item, 2)[0].qty), 3.0)
self.assertEqual(flt(rows_for(rm_item, 1)[0].qty), 1.0)
# adding it again for the same operation updates the row instead of stacking another
bom.add_raw_materials(2, [{"item_code": rm_item, "qty": 5}])
self.assertEqual(len(rows_for(rm_item, 2)), 1)
self.assertEqual(flt(rows_for(rm_item, 2)[0].qty), 5.0)
@timeout
def test_operation_bom_materials_expand_on_single_pass_submit(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
fg_item = make_item(properties={"is_stock_item": 1}).name
sfg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100.0}).name
make_workstation({"workstation": "_Test SFG Workstation"})
for operation in ("_Test SFG Operation", "_Test SFG Final Operation"):
make_operation({"operation": operation, "workstation": "_Test SFG Workstation"})
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg_item, quantity=1)
sfg_bom.append("items", {"item_code": rm_item, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
bom = frappe.new_doc("BOM")
bom.company = "_Test Company"
bom.item = fg_item
bom.quantity = 1
bom.with_operations = 1
bom.track_semi_finished_goods = 1
bom.append(
"operations",
{
"operation": "_Test SFG Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"bom_no": sfg_bom.name,
},
)
bom.append(
"operations",
{
"operation": "_Test SFG Final Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"is_final_finished_good": 1,
},
)
bom.append("items", {"item_code": sfg_item, "qty": 1, "operation_row_id": 2})
bom.submit()
self.assertEqual(bom.docstatus, 1)
self.assertEqual(bom.operations[0].finished_good, sfg_item)
self.assertTrue(
any(row.item_code == rm_item and cint(row.operation_row_id) == 1 for row in bom.items)
)
@timeout
def test_final_operation_must_produce_the_bom_item(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
fg_item = make_item(properties={"is_stock_item": 1}).name
sfg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100.0}).name
make_workstation({"workstation": "_Test SFG Workstation"})
for operation in ("_Test SFG Operation", "_Test SFG Final Operation"):
make_operation({"operation": operation, "workstation": "_Test SFG Workstation"})
bom = frappe.new_doc("BOM")
bom.company = "_Test Company"
bom.item = fg_item
bom.quantity = 1
bom.with_operations = 1
bom.track_semi_finished_goods = 1
bom.append(
"operations",
{
"operation": "_Test SFG Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"finished_good": sfg_item,
},
)
bom.append(
"operations",
{
"operation": "_Test SFG Final Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"is_final_finished_good": 1,
"finished_good": sfg_item,
},
)
bom.append("items", {"item_code": rm_item, "qty": 1, "operation_row_id": 1})
bom.append("items", {"item_code": sfg_item, "qty": 1, "operation_row_id": 2})
# the final operation claims to produce the semi FG, not this BOM's item
self.assertRaises(frappe.ValidationError, bom.insert)
bom.operations[1].finished_good = fg_item
bom.insert()
def get_default_bom(item_code="_Test FG Item 2"):
return frappe.db.get_value("BOM", {"item": item_code, "is_active": 1, "is_default": 1})

3
erpnext/manufacturing/doctype/bom_operation/bom_operation.json
View File

@@ -213,6 +213,7 @@
"fieldtype": "Link",
"in_list_view": 1,
"label": "FG / Semi FG Item",
"mandatory_depends_on": "eval:parent.track_semi_finished_goods === 1",
"options": "Item"
},
{
@@ -307,7 +308,7 @@
"index_web_pages_for_search": 1,
"istable": 1,
"links": [],
"modified": "2026-05-25 17:15:42.044630",
"modified": "2026-08-08 12:00:00.000000",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "BOM Operation",

67
erpnext/manufacturing/doctype/job_card/job_card.js
View File

@@ -103,7 +103,8 @@ frappe.ui.form.on("Job Card", {
doc.docstatus === 1 &&
!doc.is_subcontracted &&
(doc.skip_material_transfer || doc.transferred_qty > 0) &&
flt(doc.manufactured_qty) + flt(doc.process_loss_qty) < flt(doc.for_quantity);
flt(doc.manufactured_qty) + flt(doc.process_loss_qty) <
flt(doc.for_quantity) - flt(doc.pending_qty);
if (!can_make_stock_entry) return;
@@ -247,13 +248,15 @@ frappe.ui.form.on("Job Card", {
const fields = [
{
fieldtype: "Float",
label: __("Qty to Manufacture"),
label: __("Qty to Manufacture in this Cycle"),
fieldname: "for_quantity",
reqd: 1,
default: pending_qty,
description: __("Completed, Pending and Process Loss quantities must add up to this."),
change() {
const dialog = frm.job_completion_dialog;
dialog.set_value("completed_qty", dialog.get_value("for_quantity"));
dialog.set_value("pending_qty", 0);
dialog.set_value("process_loss_qty", 0);
},
},
@@ -265,10 +268,6 @@ frappe.ui.form.on("Job Card", {
default: pending_qty,
change() {
const dialog = frm.job_completion_dialog;
<<<<<<< HEAD
const remaining = dialog.get_value("for_quantity") - dialog.get_value("completed_qty");
if (remaining > 0 && remaining != dialog.get_value("pending_qty")) {
=======
const remaining =
dialog.get_value("for_quantity") -
dialog.get_value("completed_qty") -
@@ -286,7 +285,6 @@ frappe.ui.form.on("Job Card", {
}
if (remaining != dialog.get_value("pending_qty")) {
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
dialog.set_value("pending_qty", remaining);
}
},
@@ -296,15 +294,13 @@ frappe.ui.form.on("Job Card", {
label: __("Pending Quantity"),
fieldname: "pending_qty",
default: 0.0,
description: __("Qty left for a later cycle or for another job card."),
change() {
const dialog = frm.job_completion_dialog;
const process_loss_qty =
dialog.get_value("for_quantity") -
dialog.get_value("completed_qty") -
dialog.get_value("pending_qty");
<<<<<<< HEAD
if (process_loss_qty >= 0 && process_loss_qty != dialog.get_value("process_loss_qty")) {
=======
if (process_loss_qty < 0) {
dialog.set_value("pending_qty", 0);
@@ -320,7 +316,6 @@ frappe.ui.form.on("Job Card", {
}
if (process_loss_qty != dialog.get_value("process_loss_qty")) {
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
dialog.set_value("process_loss_qty", process_loss_qty);
}
},
@@ -329,15 +324,13 @@ frappe.ui.form.on("Job Card", {
fieldtype: "Float",
label: __("Process Loss Quantity"),
fieldname: "process_loss_qty",
description: __("Qty scrapped in this cycle, nobody will produce it."),
onchange() {
const dialog = frm.job_completion_dialog;
const remaining =
dialog.get_value("for_quantity") -
dialog.get_value("completed_qty") -
dialog.get_value("process_loss_qty");
<<<<<<< HEAD
if (remaining >= 0 && remaining != dialog.get_value("pending_qty")) {
=======
if (remaining < 0) {
dialog.set_value("process_loss_qty", 0);
@@ -353,7 +346,6 @@ frappe.ui.form.on("Job Card", {
}
if (remaining != dialog.get_value("pending_qty")) {
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
dialog.set_value("pending_qty", remaining);
}
},
@@ -417,9 +409,8 @@ frappe.ui.form.on("Job Card", {
},
});
},
__("Enter Value"),
__("Update"),
__("Set Finished Good Quantity")
__("Complete Job"),
__("Update")
);
},
@@ -445,46 +436,6 @@ frappe.ui.form.on("Job Card", {
});
},
make_finished_good(frm) {
const fields = [
{
fieldtype: "Float",
label: __("Completed Quantity"),
fieldname: "qty",
reqd: 1,
default: frm.doc.for_quantity - frm.doc.manufactured_qty,
},
{
fieldtype: "Datetime",
label: __("End Time"),
fieldname: "end_time",
default: frappe.datetime.now_datetime(),
},
];
frappe.prompt(
fields,
(data) => {
if (data.qty <= 0) {
frappe.throw(__("Quantity should be greater than 0"));
}
frm.call({
method: "make_finished_good",
doc: frm.doc,
args: { qty: data.qty, end_time: data.end_time },
callback(r) {
const doc = frappe.model.sync(r.message);
frappe.set_route("Form", doc[0].doctype, doc[0].name);
},
});
},
__("Enter Value"),
__("Update"),
__("Set Finished Good Quantity")
);
},
setup_quality_inspection(frm) {
const quality_inspection_field = frm.get_docfield("quality_inspection");
quality_inspection_field.get_route_options_for_new_doc = function (frm) {

4
erpnext/manufacturing/doctype/job_card/job_card.json
View File

@@ -703,11 +703,7 @@
"grid_page_length": 50,
"is_submittable": 1,
"links": [],
<<<<<<< HEAD
"modified": "2026-06-19 17:39:42.293242",
=======
"modified": "2026-08-01 14:22:19.926911",
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "Job Card",

356
erpnext/manufacturing/doctype/job_card/job_card.py
View File

@@ -847,6 +847,9 @@ class JobCard(Document):
)
def validate_transfer_qty(self):
if self.track_semi_finished_goods and self.skip_material_transfer:
return
if (
not self.finished_good
and not self.is_corrective_job_card
@@ -897,24 +900,14 @@ class JobCard(Document):
)
precision = self.precision("total_completed_qty")
total_completed_qty = flt(
accounted_qty = flt(
flt(self.total_completed_qty, precision)
+ flt(self.process_loss_qty, precision)
+ flt(self.pending_qty, precision)
)
if self.for_quantity and flt(total_completed_qty, precision) != flt(self.for_quantity, precision):
total_completed_qty_label = bold(_("Total Completed Qty"))
qty_to_manufacture = bold(_("Qty to Manufacture"))
if self.for_quantity and flt(accounted_qty, precision) != flt(self.for_quantity, precision):
frappe.throw(
<<<<<<< HEAD
_("The {0} ({1}) must be equal to {2} ({3})").format(
total_completed_qty_label,
bold(flt(total_completed_qty, precision)),
qty_to_manufacture,
bold(self.for_quantity),
=======
_(
"Total Completed Qty ({0}), Process Loss Qty ({1}) and Pending Qty ({2}) must add up to the Qty to Manufacture ({3})."
).format(
@@ -922,7 +915,6 @@ class JobCard(Document):
bold(self.get_qty_with_uom(self.process_loss_qty)),
bold(self.get_qty_with_uom(self.pending_qty)),
bold(self.get_qty_with_uom(self.for_quantity)),
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
)
)
@@ -1085,6 +1077,9 @@ class JobCard(Document):
wo.calculate_operating_cost()
wo.set_actual_dates()
if wo.track_semi_finished_goods:
wo.set_process_loss_qty()
if time_data:
wo.status = "In Process"
@@ -1174,7 +1169,10 @@ class JobCard(Document):
_(
"Row #{0}: Cannot transfer more than Required Qty {1} for Item {2} against Job Card {3}"
).format(
row.idx, frappe.bold(required_qty), frappe.bold(row.item_code), ste_doc.job_card
row.idx,
frappe.bold(self.get_qty_with_uom(required_qty, row.item_code)),
frappe.bold(row.item_code),
ste_doc.job_card,
),
title=_("Excess Transfer"),
exc=JobCardOverTransferError,
@@ -1196,51 +1194,6 @@ class JobCard(Document):
self.set_status(update_status=True)
<<<<<<< HEAD
=======
def get_job_card_items_transferred_qty(self, ste_doc):
from frappe.query_builder.functions import Sum
job_card_items = [x.get("job_card_item") for x in ste_doc.get("items") if x.get("job_card_item")]
if not job_card_items:
return {}
se = frappe.qb.DocType("Stock Entry")
sed = frappe.qb.DocType("Stock Entry Detail")
query = (
frappe.qb.from_(sed)
.join(se)
.on(sed.parent == se.name)
.select(sed.job_card_item, Sum(sed.qty))
.where(
(sed.job_card_item.isin(job_card_items))
& (se.docstatus == 1)
& (se.purpose == "Material Transfer for Manufacture")
)
.groupby(sed.job_card_item)
)
return frappe._dict(query.run(as_list=True))
def validate_over_transfer(self, ste_doc, row, transferred_qty):
"Block over transfer of items if not allowed in settings."
required_qty = frappe.db.get_value("Job Card Item", row.job_card_item, "required_qty")
if flt(transferred_qty) > flt(required_qty):
frappe.throw(
_(
"Row #{0}: Cannot transfer more than Required Qty {1} for Item {2} against Job Card {3}"
).format(
row.idx,
frappe.bold(self.get_qty_with_uom(required_qty, row.item_code)),
frappe.bold(row.item_code),
ste_doc.job_card,
),
title=_("Excess Transfer"),
exc=JobCardOverTransferError,
)
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
def set_transferred_qty(self, update_status=False):
from frappe.query_builder.functions import Sum
@@ -1293,7 +1246,7 @@ class JobCard(Document):
def set_status(self, update_status=False):
self.status = {0: "Open", 1: "Submitted", 2: "Cancelled"}[self.docstatus or 0]
if self.finished_good and self.docstatus == 1:
if (self.manufactured_qty + self.process_loss_qty) >= self.for_quantity:
if (self.manufactured_qty + self.process_loss_qty) >= self.get_qty_to_produce():
self.status = "Completed"
elif self.transferred_qty > 0 or self.skip_material_transfer:
self.status = "Work In Progress"
@@ -1324,7 +1277,8 @@ class JobCard(Document):
self.status = "Work In Progress"
if self.docstatus == 1 and (
self.for_quantity <= (self.total_completed_qty + self.process_loss_qty) or not self.items
self.get_qty_to_produce() <= (self.total_completed_qty + self.process_loss_qty)
or not self.items
):
self.status = "Completed"
@@ -1337,8 +1291,6 @@ class JobCard(Document):
if self.workstation:
self.update_workstation_status()
<<<<<<< HEAD
=======
def get_qty_to_produce(self):
"""Qty this job card is expected to produce, the pending qty is left to another job card."""
return flt(self.for_quantity) - flt(self.pending_qty)
@@ -1351,52 +1303,15 @@ class JobCard(Document):
return f"{flt(qty, self.precision('total_completed_qty'))} {uom or ''}".strip()
def set_finished_good_status(self):
# Only reached for a submitted job card (docstatus == 1) with a finished good, see set_status().
qty_to_produce = self.get_qty_to_produce()
if (self.manufactured_qty + self.process_loss_qty) >= qty_to_produce:
self.status = "Completed"
elif (self.total_completed_qty + self.process_loss_qty) >= qty_to_produce:
# Production is done and the card is submitted, but the finished goods have not been
# booked into stock yet (Manufacture Stock Entry pending) — distinct from active WIP.
self.status = "To Manufacture"
elif self.transferred_qty > 0 or self.skip_material_transfer:
self.status = "Work In Progress"
def set_non_semi_fg_status(self):
if self.items:
item_data = frappe.get_all(
"Job Card Item",
filters={"parent": self.name},
fields=["transferred_qty", "required_qty"],
)
all_transferred = item_data and all(
flt(d.transferred_qty) >= flt(d.required_qty) for d in item_data
)
any_transferred = any(flt(d.transferred_qty) > 0 for d in item_data)
if all_transferred:
self.status = "Material Transferred"
elif any_transferred:
self.status = "Partially Transferred"
elif flt(self.for_quantity) <= flt(self.transferred_qty):
self.status = "Material Transferred"
if self.time_logs:
self.status = "Work In Progress"
if self.docstatus == 1 and (
self.get_qty_to_produce() <= (self.total_completed_qty + self.process_loss_qty) or not self.items
):
self.status = "Completed"
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
def set_wip_warehouse(self):
if not self.wip_warehouse:
self.wip_warehouse = frappe.get_cached_value("Company", self.company, "default_wip_warehouse")
<<<<<<< HEAD
def set_stock_uom(self):
item_code = self.finished_good or self.production_item
if item_code:
self.stock_uom = frappe.get_cached_value("Item", item_code, "stock_uom")
def validate_operation_id(self):
if (
self.get("operation_id")
@@ -1407,36 +1322,6 @@ class JobCard(Document):
!= self.operation_id
):
work_order = bold(get_link_to_form("Work Order", self.work_order))
=======
def set_stock_uom(self):
item_code = self.finished_good or self.production_item
if item_code:
self.stock_uom = frappe.get_cached_value("Item", item_code, "stock_uom")
def set_operation_id(self):
if not (self.work_order and self.operation):
return
if self.operation_id and self.docstatus != 0:
return
operation_rows = frappe.get_all(
"Work Order Operation",
filters={"parent": self.work_order, "operation": self.operation},
pluck="name",
)
if self.operation_id:
if operation_rows and self.operation_id not in operation_rows:
frappe.throw(
_("Operation {0} does not belong to the work order {1}").format(
bold(self.operation), get_link_to_form("Work Order", self.work_order)
)
)
elif len(operation_rows) == 1:
self.operation_id = operation_rows[0]
elif operation_rows and self.docstatus == 0:
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
frappe.throw(
_("Operation {0} does not belong to the work order {1}").format(
bold(self.operation), work_order
@@ -1482,17 +1367,13 @@ class JobCard(Document):
if not (self.work_order and self.sequence_id):
return
<<<<<<< HEAD
=======
current_operation_qty = self.get_current_operation_completed_qty()
current_operation_qty = 0.0
data = self.get_current_operation_data()
if data and len(data) > 0:
current_operation_qty = flt(data[0].completed_qty)
for row in self.get_previous_operations():
if self.track_semi_finished_goods:
self.validate_previous_operation_manufactured_qty(row, current_operation_qty)
else:
self.validate_previous_operation(row, current_operation_qty)
current_operation_qty += flt(self.total_completed_qty)
def get_previous_operations(self):
previous_operations = frappe.get_all(
"Work Order Operation",
fields=["name", "operation", "status", "completed_qty", "sequence_id", "finished_good"],
@@ -1500,6 +1381,10 @@ class JobCard(Document):
order_by="sequence_id, idx",
)
message = "Job Card {}: As per the sequence of the operations in the work order {}".format(
bold(self.name), bold(get_link_to_form("Work Order", self.work_order))
)
if self.track_semi_finished_goods and previous_operations:
manufactured_qty = self.get_manufactured_qty_per_operation(
[row.name for row in previous_operations]
@@ -1508,7 +1393,38 @@ class JobCard(Document):
for row in previous_operations:
row.manufactured_qty = flt(manufactured_qty.get(row.name))
return previous_operations
for row in previous_operations:
if self.track_semi_finished_goods:
self.validate_previous_operation_manufactured_qty(row, current_operation_qty)
continue
if not row.completed_qty:
frappe.throw(
_("{0}, complete the operation {1} before the operation {2}.").format(
message, bold(row.operation), bold(self.operation)
),
OperationSequenceError,
)
if row.status != "Completed" and row.completed_qty < current_operation_qty:
frappe.throw(
_("{0}, complete the operation {1} before the operation {2}.").format(
message, bold(row.operation), bold(self.operation)
),
OperationSequenceError,
)
if row.completed_qty < current_operation_qty:
frappe.throw(
_(
"The completed quantity {0} of an operation {1} cannot be greater than the completed quantity {2} of a previous operation {3}."
).format(
bold(self.get_qty_with_uom(current_operation_qty)),
bold(self.operation),
bold(self.get_qty_with_uom(row.completed_qty, row.finished_good)),
bold(row.operation),
)
)
def get_manufactured_qty_per_operation(self, operation_ids):
job_card = frappe.qb.DocType("Job Card")
@@ -1527,68 +1443,9 @@ class JobCard(Document):
return dict(data)
def get_current_operation_completed_qty(self):
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
current_operation_qty = 0.0
data = self.get_current_operation_data()
if data and len(data) > 0:
current_operation_qty = flt(data[0].completed_qty)
def validate_previous_operation_manufactured_qty(self, row, current_operation_qty):
manufactured_qty = flt(row.manufactured_qty)
current_operation_qty += flt(self.total_completed_qty)
data = frappe.get_all(
"Work Order Operation",
fields=["operation", "status", "completed_qty", "sequence_id"],
filters={"docstatus": 1, "parent": self.work_order, "sequence_id": ("<", self.sequence_id)},
order_by="sequence_id, idx",
)
<<<<<<< HEAD
message = "Job Card {}: As per the sequence of the operations in the work order {}".format(
bold(self.name), bold(get_link_to_form("Work Order", self.work_order))
)
for row in data:
if not row.completed_qty:
frappe.throw(
_("{0}, complete the operation {1} before the operation {2}.").format(
message, bold(row.operation), bold(self.operation)
),
OperationSequenceError,
=======
if row.completed_qty < current_operation_qty:
frappe.throw(
_(
"The completed quantity {0} of an operation {1} cannot be greater than the completed quantity {2} of a previous operation {3}."
).format(
bold(self.get_qty_with_uom(current_operation_qty)),
bold(self.operation),
bold(self.get_qty_with_uom(row.completed_qty, row.finished_good)),
bold(row.operation),
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
)
if row.status != "Completed" and row.completed_qty < current_operation_qty:
frappe.throw(
_("{0}, complete the operation {1} before the operation {2}.").format(
message, bold(row.operation), bold(self.operation)
),
OperationSequenceError,
)
<<<<<<< HEAD
if row.completed_qty < current_operation_qty:
frappe.throw(
_(
"The completed quantity {0} of an operation {1} cannot be greater than the completed quantity {2} of a previous operation {3}."
).format(
bold(current_operation_qty),
bold(self.operation),
bold(row.completed_qty),
bold(row.operation),
)
)
=======
if not manufactured_qty:
frappe.throw(
_(
@@ -1614,7 +1471,6 @@ class JobCard(Document):
),
OperationSequenceError,
)
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
def validate_work_order(self):
if self.is_work_order_closed():
@@ -1744,8 +1600,17 @@ class JobCard(Document):
if isinstance(kwargs, dict):
kwargs = frappe._dict(kwargs)
self.set_for_quantity(kwargs)
self.validate_complete_job_card_qty(kwargs)
def set_for_quantity(self, kwargs):
"""Qty to Manufacture of the completion dialog covers the current cycle only,
so the qty completed by the earlier cycles of this job card is kept."""
if not flt(kwargs.for_quantity):
return
self.for_quantity = flt(self.total_completed_qty) + flt(kwargs.for_quantity)
def validate_docstatus(self):
if self.docstatus == 2:
frappe.throw(_("Cancelled Job Card cannot be processed."))
@@ -1763,34 +1628,11 @@ class JobCard(Document):
if flt(kwargs.pending_qty) and flt(kwargs.pending_qty) > self.for_quantity:
frappe.throw(_("Pending quantity cannot be greater than the for quantity."))
self.validate_completion_qty_split(kwargs)
self.pending_qty = flt(kwargs.pending_qty)
self.process_loss_qty = flt(kwargs.process_loss_qty)
<<<<<<< HEAD
=======
def validate_completion_qty_split(self, kwargs):
if not flt(kwargs.for_quantity):
return
precision = self.precision("total_completed_qty")
accounted_qty = flt(kwargs.qty) + flt(kwargs.pending_qty) + flt(kwargs.process_loss_qty)
if flt(accounted_qty, precision) == flt(kwargs.for_quantity, precision):
return
frappe.throw(
_(
"Completed Quantity ({0}), Pending Quantity ({1}) and Process Loss Quantity ({2}) must add up to the Qty to Manufacture ({3})."
).format(
bold(self.get_qty_with_uom(kwargs.qty)),
bold(self.get_qty_with_uom(kwargs.pending_qty)),
bold(self.get_qty_with_uom(kwargs.process_loss_qty)),
bold(self.get_qty_with_uom(kwargs.for_quantity)),
)
)
def add_completion_time_logs(self, kwargs):
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
if kwargs.end_time:
self.add_time_logs(
to_time=kwargs.end_time,
@@ -1816,25 +1658,49 @@ class JobCard(Document):
_("Job Card {0} has been completed").format(get_link_to_form("Job Card", self.name))
)
def validate_completion_qty_split(self, kwargs):
if not flt(kwargs.for_quantity):
return
precision = self.precision("total_completed_qty")
accounted_qty = flt(
flt(kwargs.qty, precision)
+ flt(kwargs.pending_qty, precision)
+ flt(kwargs.process_loss_qty, precision)
)
if flt(accounted_qty, precision) == flt(kwargs.for_quantity, precision):
return
frappe.throw(
_(
"Completed Quantity ({0}), Pending Quantity ({1}) and Process Loss Quantity ({2}) must add up to the Qty to Manufacture ({3})."
).format(
bold(self.get_qty_with_uom(kwargs.qty)),
bold(self.get_qty_with_uom(kwargs.pending_qty)),
bold(self.get_qty_with_uom(kwargs.process_loss_qty)),
bold(self.get_qty_with_uom(kwargs.for_quantity)),
)
)
def get_consumed_process_loss(self):
table = frappe.qb.DocType("Stock Entry")
query = (
frappe.qb.from_(table)
.select(Sum(table.process_loss_qty))
.where((table.purpose == "Manufacture") & (table.job_card == self.name) & (table.docstatus == 1))
)
return query.run()[0][0] or 0
@frappe.whitelist()
def make_stock_entry_for_semi_fg_item(self, auto_submit: bool = False):
def get_consumed_process_loss():
table = frappe.qb.DocType("Stock Entry")
query = (
frappe.qb.from_(table)
.select(Sum(table.process_loss_qty))
.where(
(table.purpose == "Manufacture") & (table.job_card == self.name) & (table.docstatus == 1)
)
)
return query.run()[0][0] or 0
from erpnext.stock.doctype.stock_entry_type.stock_entry_type import ManufactureEntry
consumed_process_loss = self.get_consumed_process_loss()
ste = ManufactureEntry(
{
"for_quantity": self.for_quantity - self.manufactured_qty,
"process_loss_qty": max(self.process_loss_qty - get_consumed_process_loss(), 0),
"for_quantity": self.get_qty_to_produce() - self.manufactured_qty - consumed_process_loss,
"process_loss_qty": max(self.process_loss_qty - consumed_process_loss, 0),
"job_card": self.name,
"skip_material_transfer": self.skip_material_transfer,
"backflush_from_wip_warehouse": self.backflush_from_wip_warehouse,

877
erpnext/manufacturing/doctype/job_card/test_job_card.py
View File

@@ -11,6 +11,7 @@ from frappe.utils.data import add_to_date, now, today
from erpnext.manufacturing.doctype.job_card.job_card import (
JobCardOverTransferError,
OperationMismatchError,
OperationSequenceError,
OverlapError,
make_corrective_job_card,
make_material_request,
@@ -888,8 +889,6 @@ class TestJobCard(ERPNextTestSuite):
self.assertEqual(wo_doc.process_loss_qty, 2)
self.assertEqual(wo_doc.status, "Completed")
<<<<<<< HEAD
=======
def get_first_job_card(self, work_order):
return frappe.get_doc(
"Job Card",
@@ -966,7 +965,6 @@ class TestJobCard(ERPNextTestSuite):
self.assertEqual(flt(job_card.total_completed_qty), 5)
self.assertEqual(flt(job_card.process_loss_qty), 0)
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))
def test_op_cost_calculation(self):
from erpnext.manufacturing.doctype.routing.test_routing import (
create_routing,
@@ -1344,6 +1342,440 @@ class TestJobCard(ERPNextTestSuite):
8,
)
def test_semi_fg_process_loss_rolls_up_to_work_order(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm = make_item("Process Loss Rollup RM 1", {"is_stock_item": 1}).name
fg = make_item("Process Loss Rollup FG 1", {"is_stock_item": 1}).name
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
fg_bom.append("items", {"item_code": rm, "qty": 1, "operation_row_id": 1})
operation = {
"operation": "Process Loss Rollup Op A",
"workstation": "_Test Workstation A",
"finished_good": fg,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation)
make_operation(operation)
fg_bom.append("operations", operation)
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg,
qty=10,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
do_not_save=True,
)
work_order.operations[0].time_in_mins = 60
work_order.save()
work_order.submit()
make_stock_entry(item_code=rm, target=warehouse, qty=100, basic_rate=100)
job_card = frappe.get_doc(
"Job Card",
frappe.get_all(
"Job Card",
filters={"work_order": work_order.name},
order_by="sequence_id, creation",
limit=1,
pluck="name",
)[0],
)
job_card.append("time_logs", {"from_time": "2024-05-01 08:00:00"})
job_card.save()
job_card.complete_job_card(
qty=8,
for_quantity=10,
pending_qty=0,
process_loss_qty=2,
end_time="2024-05-01 09:00:00",
)
job_card.reload()
self.assertEqual(flt(job_card.process_loss_qty), 2)
job_card.submit()
frappe.get_doc(job_card.make_stock_entry_for_semi_fg_item()).submit()
self.assertEqual(
flt(
frappe.db.get_value("Work Order Operation", work_order.operations[0].name, "process_loss_qty")
),
2,
)
work_order.reload()
self.assertEqual(flt(work_order.produced_qty), 8)
self.assertEqual(flt(work_order.process_loss_qty), 2)
self.assertEqual(work_order.status, "Completed")
def test_semi_fg_process_loss_of_an_intermediate_operation_rolls_up_to_work_order(self):
"""Loss booked by an earlier operation shrinks what the final operation can produce,
so it has to show up on the work order even though the final operation loses nothing."""
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm = make_item("Intermediate Loss RM 1", {"is_stock_item": 1}).name
sfg = make_item("Intermediate Loss SFG 1", {"is_stock_item": 1}).name
fg = make_item("Intermediate Loss FG 1", {"is_stock_item": 1}).name
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg, quantity=1)
sfg_bom.append("items", {"item_code": rm, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
operations = [
{
"operation": "Intermediate Loss Op A",
"finished_good": sfg,
"bom_no": sfg_bom.name,
"sequence_id": 1,
},
{
"operation": "Intermediate Loss Op B",
"finished_good": fg,
"is_final_finished_good": 1,
"sequence_id": 2,
},
]
for row in operations:
row.update(
{
"workstation": "_Test Workstation A",
"finished_good_qty": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
)
make_workstation(row)
make_operation(row)
fg_bom.append("operations", row)
fg_bom.append("items", {"item_code": sfg, "qty": 1, "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg,
qty=10,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
do_not_save=True,
)
for row in work_order.operations:
row.time_in_mins = 60
work_order.save()
work_order.submit()
make_stock_entry(item_code=rm, target=warehouse, qty=100, basic_rate=100)
def get_job_card(operation):
return frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card",
{"work_order": work_order.name, "operation": operation, "docstatus": 0},
"name",
),
)
jc_a = get_job_card("Intermediate Loss Op A")
jc_a.append("time_logs", {"from_time": "2024-06-01 08:00:00"})
jc_a.save()
jc_a.complete_job_card(
qty=8, for_quantity=10, pending_qty=0, process_loss_qty=2, end_time="2024-06-01 09:00:00"
)
jc_a.reload()
jc_a.submit()
frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item()).submit()
work_order.reload()
self.assertEqual(flt(work_order.process_loss_qty), 2)
jc_b = get_job_card("Intermediate Loss Op B")
jc_b.for_quantity = 8
for row in jc_b.items:
row.required_qty = 8
jc_b.append(
"time_logs",
{"from_time": "2024-06-02 08:00:00", "to_time": "2024-06-02 09:00:00", "completed_qty": 8},
)
jc_b.save()
jc_b.submit()
frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item()).submit()
work_order.reload()
self.assertEqual(flt(work_order.produced_qty), 8)
self.assertEqual(flt(work_order.process_loss_qty), 2)
self.assertEqual(work_order.status, "Completed")
def test_semi_fg_pending_qty_is_left_to_another_job_card(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm = make_item("Pending Qty RM 1", {"is_stock_item": 1}).name
fg = make_item("Pending Qty FG 1", {"is_stock_item": 1}).name
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
fg_bom.append("items", {"item_code": rm, "qty": 1, "operation_row_id": 1})
operation = {
"operation": "Pending Qty Op A",
"workstation": "_Test Workstation A",
"finished_good": fg,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation)
make_operation(operation)
fg_bom.append("operations", operation)
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg,
qty=5,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
do_not_save=True,
)
work_order.operations[0].time_in_mins = 60
work_order.save()
work_order.submit()
make_stock_entry(item_code=rm, target=warehouse, qty=100, basic_rate=100)
job_card = frappe.get_doc(
"Job Card",
frappe.get_all(
"Job Card",
filters={"work_order": work_order.name},
order_by="sequence_id, creation",
limit=1,
pluck="name",
)[0],
)
job_card.append("time_logs", {"from_time": "2024-04-01 08:00:00"})
job_card.save()
job_card.complete_job_card(
qty=3,
for_quantity=5,
pending_qty=2,
process_loss_qty=0,
end_time="2024-04-01 09:00:00",
)
job_card.reload()
self.assertEqual(flt(job_card.for_quantity), 5)
self.assertEqual(flt(job_card.pending_qty), 2)
self.assertEqual(flt(job_card.process_loss_qty), 0)
job_card.submit()
self.assertEqual(job_card.status, "Work In Progress")
manufacturing_entry = frappe.get_doc(job_card.make_stock_entry_for_semi_fg_item())
finished_item = next(row for row in manufacturing_entry.items if row.is_finished_item)
self.assertEqual(flt(finished_item.qty), 3)
manufacturing_entry.submit()
job_card.reload()
self.assertEqual(flt(job_card.manufactured_qty), 3)
self.assertEqual(job_card.status, "Completed")
def test_semi_fg_sequence_needs_previous_operations_manufactured(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm1 = make_item("Sequence Check RM 1", {"is_stock_item": 1}).name
rm2 = make_item("Sequence Check RM 2", {"is_stock_item": 1}).name
sfg1 = make_item("Sequence Check SFG 1", {"is_stock_item": 1}).name
sfg2 = make_item("Sequence Check SFG 2", {"is_stock_item": 1}).name
fg = make_item("Sequence Check FG 1", {"is_stock_item": 1}).name
semi_fg_boms = {}
for semi_fg_item, raw_material in ((sfg1, rm1), (sfg2, rm2)):
bom = frappe.new_doc("BOM", company="_Test Company", item=semi_fg_item, quantity=1)
bom.append("items", {"item_code": raw_material, "qty": 1})
bom.insert()
bom.submit()
semi_fg_boms[semi_fg_item] = bom.name
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
operations = [
{
"operation": "Sequence Check Op A",
"finished_good": sfg1,
"bom_no": semi_fg_boms[sfg1],
"sequence_id": 1,
},
{
"operation": "Sequence Check Op B",
"finished_good": sfg2,
"bom_no": semi_fg_boms[sfg2],
"sequence_id": 1,
},
{
"operation": "Sequence Check Op C",
"finished_good": fg,
"is_final_finished_good": 1,
"sequence_id": 2,
},
]
for row in operations:
row.update(
{
"workstation": "_Test Workstation A",
"finished_good_qty": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
)
make_workstation(row)
make_operation(row)
fg_bom.append("operations", row)
fg_bom.append("items", {"item_code": sfg1, "qty": 1, "operation_row_id": 3})
fg_bom.append("items", {"item_code": sfg2, "qty": 1, "operation_row_id": 3})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg,
qty=5,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
do_not_save=True,
)
for row in work_order.operations:
row.time_in_mins = 60
work_order.save()
work_order.submit()
make_stock_entry(item_code=rm1, target=warehouse, qty=10, basic_rate=100)
make_stock_entry(item_code=rm2, target=warehouse, qty=10, basic_rate=100)
def get_job_card(operation):
return frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card",
{"work_order": work_order.name, "operation": operation, "docstatus": 0},
"name",
),
)
def add_time_log(job_card, day, qty):
job_card.append(
"time_logs",
{
"from_time": f"2024-01-{day} 08:00:00",
"to_time": f"2024-01-{day} 09:00:00",
"completed_qty": qty,
},
)
jc_a = get_job_card("Sequence Check Op A")
jc_a.for_quantity = 3
add_time_log(jc_a, "01", 3)
jc_a.submit()
jc_b = get_job_card("Sequence Check Op B")
add_time_log(jc_b, "02", jc_b.for_quantity)
jc_b.submit()
frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item()).submit()
jc_c = get_job_card("Sequence Check Op C")
jc_c.for_quantity = 3
add_time_log(jc_c, "03", 3)
self.assertRaises(OperationSequenceError, jc_c.save)
frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item()).submit()
jc_c.reload()
jc_c.for_quantity = 4
add_time_log(jc_c, "03", 4)
self.assertRaises(OperationSequenceError, jc_c.save)
jc_c.reload()
jc_c.for_quantity = 3
add_time_log(jc_c, "03", 3)
jc_c.submit()
self.assertEqual(jc_c.docstatus, 1)
def test_semi_fg_batch_auto_pull_on_manufacture(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
@@ -1481,6 +1913,299 @@ class TestJobCard(ERPNextTestSuite):
consumed_batches = get_batches_from_bundle(sfg_consume_row.serial_and_batch_bundle)
self.assertEqual(set(consumed_batches.keys()), set(produced_batches.keys()))
def test_manufacture_entry_process_loss_not_taken_from_previous_operation(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm1 = make_item("PL Scope RM 1", {"is_stock_item": 1}).name
rm2 = make_item("PL Scope RM 2", {"is_stock_item": 1}).name
sfg = make_item("PL Scope SFG 1", {"is_stock_item": 1}).name
fg1 = make_item("PL Scope FG 1", {"is_stock_item": 1}).name
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg, quantity=1)
sfg_bom.append("items", {"item_code": rm1, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg1,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
operation1 = {
"operation": "PL Scope Op A",
"workstation": "_Test Workstation A",
"finished_good": sfg,
"bom_no": sfg_bom.name,
"finished_good_qty": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
operation2 = {
"operation": "PL Scope Op B",
"workstation": "_Test Workstation A",
"finished_good": fg1,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 2,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation1)
make_operation(operation1)
make_operation(operation2)
fg_bom.append("operations", operation1)
fg_bom.append("operations", operation2)
fg_bom.append("items", {"item_code": rm2, "qty": 1})
fg_bom.append("items", {"item_code": sfg, "qty": 1, "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg1,
qty=5,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
)
make_stock_entry(item_code=rm1, target=warehouse, qty=10, basic_rate=100)
make_stock_entry(item_code=rm2, target=warehouse, qty=10, basic_rate=100)
make_stock_entry(item_code=sfg, target=warehouse, qty=10, basic_rate=100)
jc_a = frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card", {"work_order": work_order.name, "operation": "PL Scope Op A"}, "name"
),
)
jc_a.append(
"time_logs",
{"from_time": "2024-01-01 08:00:00", "to_time": "2024-01-01 09:00:00", "completed_qty": 3},
)
jc_a.pending_qty = 0
jc_a.process_loss_qty = 2
jc_a.submit()
me_a = frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item())
me_a.submit()
self.assertEqual(flt(me_a.process_loss_qty), 2.0)
jc_b = frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card", {"work_order": work_order.name, "operation": "PL Scope Op B"}, "name"
),
)
jc_b.append(
"time_logs",
{"from_time": "2024-02-01 08:00:00", "to_time": "2024-02-01 09:00:00", "completed_qty": 3},
)
jc_b.pending_qty = 2
jc_b.submit()
me_b = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
# operation A's loss must not leak into operation B's entry
self.assertEqual(flt(me_b.process_loss_qty), 0.0)
fg_row = next(row for row in me_b.items if row.is_finished_item)
self.assertEqual(flt(fg_row.qty), 3.0)
me_b.submit()
def make_semi_fg_work_order(self, prefix, qty=5):
"""Two-operation semi FG work order: Op A makes the SFG from RM 1, final Op B
consumes it. Both operations skip material transfer; stock is pre-seeded."""
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm1 = make_item(f"{prefix} RM 1", {"is_stock_item": 1}).name
rm2 = make_item(f"{prefix} RM 2", {"is_stock_item": 1}).name
sfg = make_item(f"{prefix} SFG 1", {"is_stock_item": 1}).name
fg1 = make_item(f"{prefix} FG 1", {"is_stock_item": 1}).name
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg, quantity=1)
sfg_bom.append("items", {"item_code": rm1, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg1,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
operation1 = {
"operation": f"{prefix} Op A",
"workstation": "_Test Workstation A",
"finished_good": sfg,
"bom_no": sfg_bom.name,
"finished_good_qty": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
operation2 = {
"operation": f"{prefix} Op B",
"workstation": "_Test Workstation A",
"finished_good": fg1,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 2,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation1)
make_operation(operation1)
make_operation(operation2)
fg_bom.append("operations", operation1)
fg_bom.append("operations", operation2)
fg_bom.append("items", {"item_code": rm2, "qty": 1})
fg_bom.append("items", {"item_code": sfg, "qty": 1, "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg1,
qty=qty,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
)
for item_code in (rm1, rm2, sfg):
make_stock_entry(item_code=item_code, target=warehouse, qty=10, basic_rate=100)
return work_order
def get_semi_fg_job_card(self, work_order, operation):
return frappe.get_doc(
"Job Card",
frappe.db.get_value("Job Card", {"work_order": work_order.name, "operation": operation}, "name"),
)
def test_partial_manufacture_entry_then_finish(self):
work_order = self.make_semi_fg_work_order("PL Partial")
jc_a = self.get_semi_fg_job_card(work_order, "PL Partial Op A")
jc_a.append(
"time_logs",
{"from_time": "2024-01-01 08:00:00", "to_time": "2024-01-01 09:00:00", "completed_qty": 5},
)
jc_a.submit()
frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item()).submit()
jc_b = self.get_semi_fg_job_card(work_order, "PL Partial Op B")
jc_b.append(
"time_logs",
{"from_time": "2024-02-01 08:00:00", "to_time": "2024-02-01 09:00:00", "completed_qty": 3},
)
jc_b.pending_qty = 0
jc_b.process_loss_qty = 2
jc_b.submit()
# book 1 of the 3 finished units now; the full process loss goes with this first entry,
# so it accounts for 3 of 5 and its materials are trimmed to the same share
first = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
fg_row = next(row for row in first.items if row.is_finished_item)
fg_row.qty = 1
for row in first.items:
if row.s_warehouse and not row.is_finished_item:
row.qty = flt(row.qty) * 3 / 5
first.save()
first.submit()
# the follow-up entry must be generated net of the already-booked loss and still submit
jc_b.reload()
second = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
fg_row = next(row for row in second.items if row.is_finished_item)
self.assertEqual(flt(fg_row.qty), 2.0)
self.assertEqual(flt(second.process_loss_qty), 0.0)
second.submit()
jc_b.reload()
self.assertEqual(flt(jc_b.manufactured_qty), 3.0)
# across both entries, consumption adds up to the job card's requirement of 5, no more
consumed = frappe.get_all(
"Stock Entry Detail",
filters={"parent": ["in", [first.name, second.name]], "s_warehouse": ["is", "set"]},
fields=["item_code", {"SUM": "qty", "as": "qty"}],
group_by="item_code",
)
self.assertTrue(consumed)
for row in consumed:
self.assertEqual(flt(row.qty), 5.0, f"{row.item_code} mis-consumed across partial entries")
def test_update_after_submit_keeps_manufacture_entry_intact(self):
work_order = self.make_semi_fg_work_order("PL Update")
jc_a = self.get_semi_fg_job_card(work_order, "PL Update Op A")
jc_a.append(
"time_logs",
{"from_time": "2024-01-01 08:00:00", "to_time": "2024-01-01 09:00:00", "completed_qty": 3},
)
jc_a.pending_qty = 0
jc_a.process_loss_qty = 2
jc_a.submit()
entry = frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item())
entry.submit()
if not frappe.db.exists("Print Heading", "_Test SFG Heading"):
frappe.get_doc({"doctype": "Print Heading", "print_heading": "_Test SFG Heading"}).insert()
entry.reload()
entry.select_print_heading = "_Test SFG Heading"
entry.save()
entry.reload()
self.assertEqual(flt(entry.process_loss_qty), 2.0)
def test_stale_manufacture_draft_cannot_over_produce_without_operation_bom(self):
work_order = self.make_semi_fg_work_order("PL NoBom")
jc_a = self.get_semi_fg_job_card(work_order, "PL NoBom Op A")
jc_a.append(
"time_logs",
{"from_time": "2024-01-01 08:00:00", "to_time": "2024-01-01 09:00:00", "completed_qty": 5},
)
jc_a.submit()
frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item()).submit()
# Op B has no operation BOM, so its entries carry no For Quantity to validate against
jc_b = self.get_semi_fg_job_card(work_order, "PL NoBom Op B")
jc_b.append(
"time_logs",
{"from_time": "2024-02-01 08:00:00", "to_time": "2024-02-01 09:00:00", "completed_qty": 3},
)
jc_b.pending_qty = 0
jc_b.process_loss_qty = 2
jc_b.submit()
draft_one = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
draft_two = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
draft_one.submit()
stale = frappe.get_doc("Stock Entry", draft_two.name)
self.assertRaises(frappe.ValidationError, stale.submit)
def test_semi_fg_auto_pull_with_uom_conversion(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
@@ -1896,6 +2621,54 @@ class TestJobCard(ERPNextTestSuite):
self.assertEqual(s.additional_costs[2].amount, 480)
self.assertEqual(s.additional_costs[3].amount, 480)
def test_semi_fg_job_card_is_exempt_from_transfer_qty_check(self):
jc = frappe.new_doc("Job Card")
jc.track_semi_finished_goods = 1
jc.skip_material_transfer = 1
jc.for_quantity = 10
jc.transferred_qty = 0
jc.append("items", {"item_code": "_Test Item"})
jc.validate_transfer_qty()
# with transfer enabled, a legacy card without an FG item keeps the strict check
jc.skip_material_transfer = 0
self.assertRaises(frappe.ValidationError, jc.validate_transfer_qty)
jc.finished_good = "_Test Item"
jc.validate_transfer_qty()
jc.finished_good = None
jc.track_semi_finished_goods = 0
self.assertRaises(frappe.ValidationError, jc.validate_transfer_qty)
def test_qty_in_messages_carries_the_uom(self):
jc = frappe.new_doc("Job Card")
jc.stock_uom = "Nos"
self.assertEqual(jc.get_qty_with_uom(5), "5.0 Nos")
self.assertEqual(jc.get_qty_with_uom(0), "0.0 Nos")
def test_completion_qty_split_must_add_up(self):
jc = frappe.new_doc("Job Card")
jc.for_quantity = 5
jc.validate_completion_qty_split(
frappe._dict(for_quantity=5, qty=3, pending_qty=2, process_loss_qty=0)
)
self.assertRaises(
frappe.ValidationError,
jc.validate_completion_qty_split,
frappe._dict(for_quantity=3, qty=3, pending_qty=2, process_loss_qty=0),
)
self.assertRaises(
frappe.ValidationError,
jc.validate_completion_qty_split,
frappe._dict(for_quantity=1, qty=0.3334, pending_qty=0.3334, process_loss_qty=0.3334),
)
def create_bom_with_multiple_operations():
"Create a BOM with multiple operations and Material Transfer against Job Card"
@@ -1958,101 +2731,3 @@ def create_semi_fg_bom(semi_fg_item, raw_item, inspection_required):
bom.append("items", {"item_code": raw_item, "qty": 1})
bom.submit()
return bom.name
<<<<<<< HEAD
=======
class TestJobCardLogic(ERPNextTestSuite):
"""Field-level validations and pure quantity/capacity helpers, exercised on the
document directly so they don't need a Work Order / BOM (the integration suite does)."""
def test_processing_a_submitted_or_cancelled_card_is_blocked(self):
submitted = frappe.new_doc("Job Card")
submitted.docstatus = 1
self.assertRaises(frappe.ValidationError, submitted.validate_docstatus)
cancelled = frappe.new_doc("Job Card")
cancelled.docstatus = 2
self.assertRaises(frappe.ValidationError, cancelled.validate_docstatus)
def test_complete_job_card_qty_guards(self):
jc = frappe.new_doc("Job Card")
jc.for_quantity = 5
jc.validate_complete_job_card_qty(frappe._dict(pending_qty=3)) # within range -> passes
self.assertRaises(
frappe.ValidationError, jc.validate_complete_job_card_qty, frappe._dict(pending_qty=-1)
)
self.assertRaises(
frappe.ValidationError, jc.validate_complete_job_card_qty, frappe._dict(process_loss_qty=-1)
)
self.assertRaises(
frappe.ValidationError, jc.validate_complete_job_card_qty, frappe._dict(pending_qty=10)
)
def test_qty_in_messages_carries_the_uom(self):
jc = frappe.new_doc("Job Card")
jc.stock_uom = "Nos"
self.assertEqual(jc.get_qty_with_uom(5), "5.0 Nos")
self.assertEqual(jc.get_qty_with_uom(0), "0.0 Nos")
def test_completion_qty_split_must_add_up(self):
jc = frappe.new_doc("Job Card")
jc.for_quantity = 5
# 3 completed + 2 pending + 0 lost == 5 to manufacture -> passes
jc.validate_complete_job_card_qty(
frappe._dict(for_quantity=5, qty=3, pending_qty=2, process_loss_qty=0)
)
self.assertRaises(
frappe.ValidationError,
jc.validate_complete_job_card_qty,
frappe._dict(for_quantity=3, qty=3, pending_qty=2, process_loss_qty=0),
)
def test_completed_qty_must_reconcile_with_for_quantity(self):
jc = frappe.new_doc("Job Card")
jc.for_quantity = 10
jc.total_completed_qty = 6
jc.process_loss_qty = 0
jc.pending_qty = 0
# 6 + 0 + 0 != 10 -> throws
self.assertRaises(frappe.ValidationError, jc.validate_completed_qty_matches_for_quantity)
# completed + loss + pending == for_quantity -> passes
jc.pending_qty = 4
jc.validate_completed_qty_matches_for_quantity()
def test_set_process_loss(self):
jc = frappe.new_doc("Job Card")
jc.for_quantity = 10
jc.total_completed_qty = 6
jc.pending_qty = 1
jc.set_process_loss()
self.assertEqual(jc.process_loss_qty, 3) # 10 - 6 - 1
# no loss when nothing completed yet
nothing_done = frappe.new_doc("Job Card")
nothing_done.for_quantity = 10
nothing_done.total_completed_qty = 0
nothing_done.set_process_loss()
self.assertEqual(nothing_done.process_loss_qty, 0)
def test_capacity_overlap_detection(self):
jc = frappe.new_doc("Job Card")
sequential = [
{"from_time": "2026-01-01 10:00:00", "to_time": "2026-01-01 11:00:00"},
{"from_time": "2026-01-01 11:00:00", "to_time": "2026-01-01 12:00:00"},
]
overlapping = [
{"from_time": "2026-01-01 10:00:00", "to_time": "2026-01-01 11:00:00"},
{"from_time": "2026-01-01 10:30:00", "to_time": "2026-01-01 11:30:00"},
]
# sequential logs share one capacity slot; overlapping logs need two
self.assertEqual(len(jc.get_alloted_capacity(sequential)), 1)
self.assertEqual(len(jc.get_alloted_capacity(overlapping)), 2)
# capacity 1 overlaps with any log; capacity 2 only when both slots are taken
self.assertTrue(jc.has_overlap(1, sequential))
self.assertFalse(jc.has_overlap(2, sequential))
self.assertTrue(jc.has_overlap(2, overlapping))
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))

18
erpnext/manufacturing/doctype/work_order/test_work_order.py
View File

@@ -4839,6 +4839,24 @@ class TestWorkOrder(ERPNextTestSuite):
# generated qty (3.0 for 8 units) differs from the BOM-scaled qty (7.5 for 20 units)
self.assertEqual(flt(row.qty, 6), 3.0)
def test_wip_warehouse_required_when_tracking_semi_finished_goods(self):
wo = frappe.new_doc("Work Order")
wo.track_semi_finished_goods = 1
wo.skip_transfer = 0
wo.fg_warehouse = "_Test Warehouse 1 - _TC"
self.assertRaises(frappe.ValidationError, wo.validate_warehouse)
wo.wip_warehouse = "_Test Warehouse - _TC"
wo.validate_warehouse()
# the top-level target warehouse stays optional; operations may carry their own
wo.fg_warehouse = None
wo.validate_warehouse()
wo.track_semi_finished_goods = 0
self.assertRaises(frappe.ValidationError, wo.validate_warehouse)
def get_reserved_entries(voucher_no, warehouse=None):
doctype = frappe.qb.DocType("Stock Reservation Entry")

4
erpnext/manufacturing/doctype/work_order/work_order.json
View File

@@ -272,7 +272,7 @@
"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) && !doc.track_semi_finished_goods",
"mandatory_depends_on": "eval:!doc.skip_transfer || doc.from_wip_warehouse",
"options": "Warehouse"
},
{
@@ -739,7 +739,7 @@
"image_field": "image",
"is_submittable": 1,
"links": [],
"modified": "2026-06-03 21:35:34.175667",
"modified": "2026-08-08 12:00:00.000000",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "Work Order",

13
erpnext/manufacturing/doctype/work_order/work_order.py
View File

@@ -883,6 +883,12 @@ class WorkOrder(Document):
return flt(query.run()[0][0])
def set_process_loss_qty(self):
self.db_set("process_loss_qty", self._process_loss_qty())
def _process_loss_qty(self):
if self.track_semi_finished_goods:
return flt(sum(flt(row.process_loss_qty) for row in self.operations))
table = frappe.qb.DocType("Stock Entry")
process_loss_qty = (
frappe.qb.from_(table)
@@ -892,7 +898,7 @@ class WorkOrder(Document):
)
).run()[0][0]
self.db_set("process_loss_qty", flt(process_loss_qty))
return flt(process_loss_qty)
def update_production_plan_status(self):
production_plan = frappe.get_doc("Production Plan", self.production_plan)
@@ -915,12 +921,9 @@ class WorkOrder(Document):
production_plan.run_method("update_produced_pending_qty", produced_qty, self.production_plan_item)
def validate_warehouse(self):
if self.track_semi_finished_goods:
return
if not self.wip_warehouse and not self.skip_transfer:
frappe.throw(_("Work-in-Progress Warehouse is required before Submit"))
if not self.fg_warehouse:
if not self.fg_warehouse and not self.track_semi_finished_goods:
frappe.throw(_("Target Warehouse is required before Submit"))
def before_submit(self):

834
erpnext/manufacturing/page/shop_floor/shop_floor.py
View File

@@ -1,834 +0,0 @@
import frappe
from frappe import _
from frappe.query_builder import Order
from frappe.query_builder.functions import Count, Date
from frappe.utils import cint, flt, get_datetime, getdate, now_datetime, time_diff_in_seconds
from pypika.terms import ExistsCriterion
from erpnext.manufacturing.doctype.workstation.workstation import (
get_status_color,
get_time_logs,
)
from erpnext.stock.doctype.quality_inspection_template.quality_inspection_template import (
get_template_details,
)
JOB_CARD_FIELDS = [
"name",
"docstatus",
"production_item",
"work_order",
"operation",
"total_completed_qty",
"for_quantity",
"process_loss_qty",
"stock_uom",
"finished_good",
"transferred_qty",
"status",
"expected_start_date",
"expected_end_date",
"time_required",
"wip_warehouse",
"skip_material_transfer",
"backflush_from_wip_warehouse",
"is_paused",
"manufactured_qty",
"is_subcontracted",
"workstation",
"sequence_id",
"bom_no",
"operation_id",
"quality_inspection",
"quality_inspection_template",
]
TODAY_SESSION_FIELDS = [
"name",
"docstatus",
"production_item",
"finished_good",
"operation",
"total_completed_qty",
"for_quantity",
"process_loss_qty",
"total_time_in_mins",
"status",
"modified",
]
# Roles that unlock the Shop Floor manager board (work-order overview). Anyone else gets the
# operator view. System Manager is included so admins always see the full picture.
MANAGER_ROLES = {"Shop Floor Manager", "Manufacturing Manager", "System Manager"}
# Maps the manager buckets to the underlying Work Order statuses. "open" spans pending AND
# in-progress: starting a job card flips the Work Order to In Process, and separate tabs made
# it jump tabs on the next refresh — the operator would lose the card they were working on.
WORK_ORDER_STATUS_GROUPS = {
"open": ["In Process", "Not Started", "Submitted", "Stock Reserved", "Stock Partially Reserved"],
"completed": ["Completed"],
}
WORK_ORDER_FIELDS = [
"name",
"production_item",
"item_name",
"qty",
"produced_qty",
"status",
"planned_start_date",
"sales_order",
"bom_no",
]
@frappe.whitelist()
def submit_job_card(job_card: str):
"""Submit a draft job card whose quantity has already been recorded via End Session."""
frappe.has_permission("Job Card", "submit", throw=True)
jc = frappe.get_doc("Job Card", job_card)
if jc.docstatus == 0:
jc.submit()
return {"name": jc.name, "docstatus": jc.docstatus}
def _record_session(job_card, qty, for_quantity, pending_qty, process_loss_qty, end_time):
"""Record the session qty + close the active time log via Job Card's complete_job_card.
auto_submit=0 so the backend doesn't auto-create+submit a Manufacture Stock Entry — that's
a separate manual step driven by the post-submit prompt. Returns the reloaded doc.
"""
frappe.has_permission("Job Card", "write", throw=True)
doc = frappe.get_doc("Job Card", job_card)
doc.run_method(
"complete_job_card",
qty=flt(qty),
for_quantity=flt(for_quantity),
pending_qty=flt(pending_qty),
process_loss_qty=flt(process_loss_qty),
end_time=end_time,
auto_submit=0,
)
doc.reload()
return doc
@frappe.whitelist()
def save_and_continue(
job_card: str,
qty: float,
for_quantity: float,
pending_qty: float,
process_loss_qty: float,
end_time: str,
):
"""Record the session qty + close the time log, then mark the JC as paused
so the MES keeps it in the active slot for the next session."""
frappe.has_permission("Job Card", "write", throw=True)
doc = _record_session(job_card, qty, for_quantity, pending_qty, process_loss_qty, end_time)
if doc.docstatus == 0:
doc.db_set("is_paused", 1)
return {"name": doc.name}
@frappe.whitelist()
def complete_and_submit(
job_card: str,
qty: float,
for_quantity: float,
pending_qty: float,
process_loss_qty: float,
end_time: str,
):
"""Record the session qty + close the time log + submit the JC."""
frappe.has_permission("Job Card", "submit", throw=True)
doc = _record_session(job_card, qty, for_quantity, pending_qty, process_loss_qty, end_time)
if doc.docstatus == 0:
doc.submit()
return {"name": doc.name, "finished_good": doc.finished_good}
@frappe.whitelist()
def make_manufacture_stock_entry(job_card: str):
"""Build a "Manufacture" Stock Entry for the finished goods and save as draft.
Mirrors the Job Card form's "Make Stock Entry" button — uses the doc's own
make_stock_entry_for_semi_fg_item (purpose="Manufacture", job card linked) rather than
the generic make_stock_entry, which would produce a "Material Transfer for Manufacture".
Returns the draft SE name so the client can open it in a new tab.
"""
frappe.has_permission("Job Card", "read", throw=True)
frappe.has_permission("Stock Entry", "submit", throw=True)
doc = frappe.get_doc("Job Card", job_card)
se = doc.make_stock_entry_for_semi_fg_item(auto_submit=False)
return {"name": se.get("name")}
@frappe.whitelist()
def get_quality_inspection_checklist(job_card: str):
"""Template parameters for the inline quality check an operator fills before submitting a
job card. Returns the resolved template, its parameter rows, any already-linked inspection,
and the item being inspected.
"""
frappe.has_permission("Job Card", "read", throw=True)
jc = frappe.db.get_value(
"Job Card",
job_card,
[
"quality_inspection",
"quality_inspection_template",
"operation",
"production_item",
"finished_good",
],
as_dict=True,
)
if not jc:
frappe.throw(_("Job Card {0} not found").format(job_card))
template = jc.quality_inspection_template
if not template and jc.operation:
template = frappe.get_cached_value("Operation", jc.operation, "quality_inspection_template")
parameters = []
for p in get_template_details(template):
parameters.append(
{
"specification": p.specification,
"value": p.value,
"numeric": cint(p.numeric),
"min_value": p.min_value,
"max_value": p.max_value,
"formula_based_criteria": cint(p.formula_based_criteria),
"acceptance_formula": p.acceptance_formula,
}
)
return {
"template": template,
"parameters": parameters,
"existing": jc.quality_inspection,
"item_code": jc.finished_good or jc.production_item,
}
@frappe.whitelist()
def submit_quality_inspection(job_card: str, readings: str | None = None):
"""Create + submit an In-Process Quality Inspection for the job card and link it back, so the
standard Job Card.validate_inspection() gate passes when the card is submitted.
`readings` is a JSON list of {specification, status, reading_value} captured inline. For numeric
/ formula parameters the measured value is stored and the Quality Inspection auto-evaluates
pass/fail against min/max (or the formula); for qualitative parameters the operator's explicit
Accepted/Rejected is taken as authoritative (manual_inspection). The QI's own validation then
sets the overall Accepted/Rejected status.
"""
frappe.has_permission("Job Card", "write", throw=True)
frappe.has_permission("Quality Inspection", "submit", throw=True)
jc = frappe.get_doc("Job Card", job_card)
# Idempotent: if a submitted inspection is already linked, don't create another.
if jc.quality_inspection:
existing = frappe.db.get_value(
"Quality Inspection", jc.quality_inspection, ["status", "docstatus"], as_dict=True
)
if existing and existing.docstatus == 1:
return {"name": jc.quality_inspection, "status": existing.status}
template = jc.quality_inspection_template
if not template and jc.operation:
template = frappe.get_cached_value("Operation", jc.operation, "quality_inspection_template")
if not template:
frappe.throw(_("No Quality Inspection Template is configured for this operation."))
reading_map = {r.get("specification"): r for r in (frappe.parse_json(readings) or [])}
qi = frappe.new_doc("Quality Inspection")
qi.inspection_type = "In Process"
qi.reference_type = "Job Card"
qi.reference_name = job_card
qi.item_code = jc.finished_good or jc.production_item
qi.bom_no = jc.bom_no
qi.quality_inspection_template = template
qi.inspected_by = frappe.session.user
qi.get_item_specification_details() # load readings from the template
for reading in qi.readings:
entry = reading_map.get(reading.specification)
if not entry:
continue
value = entry.get("reading_value")
if reading.numeric or reading.formula_based_criteria:
# Measured value → let the Quality Inspection judge it against min/max or the formula.
if value not in (None, ""):
reading.reading_value = value
reading.reading_1 = value
else:
# Qualitative check → the operator's explicit pass/fail wins.
reading.manual_inspection = 1
reading.status = entry.get("status") or "Accepted"
if value not in (None, ""):
reading.reading_value = value
qi.insert()
qi.submit() # validate() → inspect_and_set_status() sets the overall Accepted/Rejected status
# Link explicitly: the QI's own back-reference matches on production_item, which can differ
# from the operation's finished_good, so we set it directly to be safe.
jc.db_set("quality_inspection", qi.name)
return {"name": qi.name, "status": qi.status}
@frappe.whitelist()
def get_shop_floor_context():
"""Which experience to render — the manager board or the operator view — plus the
signed-in operator's Employee (so Start Job can pre-fill them)."""
roles = set(frappe.get_roles())
can_manage = bool(roles & MANAGER_ROLES)
return {
"role_view": "manager" if can_manage else "operator",
"can_manage": can_manage,
"user_employee": frappe.db.get_value("Employee", {"user_id": frappe.session.user}, "name"),
}
@frappe.whitelist()
def get_work_orders(
status_group: str,
start: int = 0,
page_length: int = 20,
search: str | None = None,
with_job_cards_only: bool | int = 0,
):
"""Paginated Work Orders for one manager bucket (open / completed),
each decorated with a job-card status breakdown for the card's progress chip.
When `with_job_cards_only` is set, only Work Orders that have at least one (non-cancelled)
Job Card are returned — the board's opt-in "With job cards only" toggle.
"""
frappe.has_permission("Work Order", "read", throw=True)
if status_group not in WORK_ORDER_STATUS_GROUPS:
frappe.throw(_("Invalid status group: {0}").format(status_group))
start = cint(start)
page_length = cint(page_length) or 20
with_job_cards_only = cint(with_job_cards_only)
# Active buckets show the oldest-planned first (work the floor next); completed shows newest first.
order = Order.desc if status_group == "completed" else Order.asc
wo = frappe.qb.DocType("Work Order")
query = _apply_work_order_filters(frappe.qb.from_(wo), wo, status_group, search, with_job_cards_only)
work_orders = (
query.select(*[wo[field] for field in WORK_ORDER_FIELDS])
.orderby(wo.planned_start_date, order=order)
.limit(page_length)
.offset(start)
).run(as_dict=True)
total = _count_work_orders(status_group, search, with_job_cards_only)
_enrich_work_orders(work_orders)
return {
"work_orders": work_orders,
"total": cint(total),
"start": start,
"page_length": page_length,
}
def _has_job_cards_criterion(wo, docstatus):
jc = frappe.qb.DocType("Job Card")
return ExistsCriterion(
frappe.qb.from_(jc).select(jc.name).where((jc.work_order == wo.name) & (jc.docstatus == docstatus))
)
def _bucket_criterion(wo, status_group):
"""The floor is done with a Work Order once every job card is submitted, even though the
Work Order stays "In Process" until the finished goods are received. Such operationally
complete orders belong on the Completed tab — not stranded under Pending / In Progress."""
open_statuses = WORK_ORDER_STATUS_GROUPS["open"]
all_job_cards_done = _has_job_cards_criterion(wo, 1) & _has_job_cards_criterion(wo, 0).negate()
if status_group == "completed":
return (wo.status == "Completed") | (wo.status.isin(open_statuses) & all_job_cards_done)
return wo.status.isin(open_statuses) & (
_has_job_cards_criterion(wo, 1).negate() | _has_job_cards_criterion(wo, 0)
)
def _apply_work_order_filters(query, wo, status_group, search, with_job_cards_only):
"""Shared WHERE clauses for the board's row + count queries (bucket, search, job-card toggle)."""
query = query.where((wo.docstatus == 1) & _bucket_criterion(wo, status_group))
if search:
like = f"%{search}%"
query = query.where(wo.name.like(like) | wo.production_item.like(like) | wo.item_name.like(like))
if with_job_cards_only:
jc = frappe.qb.DocType("Job Card")
wo_with_job_cards = frappe.qb.from_(jc).select(jc.work_order).where(jc.docstatus < 2)
query = query.where(wo.name.isin(wo_with_job_cards))
return query
def _count_work_orders(status_group: str, search: str | None, with_job_cards_only: int = 0) -> int:
"""Total Work Orders in a bucket (drives pagination), honouring the same filters as the rows."""
wo = frappe.qb.DocType("Work Order")
query = _apply_work_order_filters(frappe.qb.from_(wo), wo, status_group, search, with_job_cards_only)
return cint(query.select(Count("*")).run()[0][0])
def _enrich_work_orders(work_orders: list[dict]) -> None:
"""Attach item/workstation image, status colour, % complete and a job-card breakdown to each row."""
wo_names = [row.name for row in work_orders]
jc_counts = _get_job_card_status_counts(wo_names)
workstation_map = _get_current_workstation_map(wo_names)
for row in work_orders:
row.item_image = (
frappe.get_cached_value("Item", row.production_item, "image") if row.production_item else None
)
row.status_colour = get_status_color(row.status)
row.per_completed = round(flt(row.produced_qty) / flt(row.qty) * 100, 1) if flt(row.qty) else 0
counts = jc_counts.get(row.name, {})
row.job_card_status = counts.get("by_status", {})
row.total_operations = counts.get("total", 0)
row.completed_operations = counts.get("completed", 0)
row.in_progress_operations = counts.get("in_progress", 0)
# Two segments for the card's progress bar: green (done) + orange (in progress); the rest
# of the track stays grey (pending / not started).
row.per_operations = (
round(row.completed_operations / row.total_operations * 100, 1) if row.total_operations else 0
)
row.per_in_progress = (
round(row.in_progress_operations / row.total_operations * 100, 1) if row.total_operations else 0
)
# Current/active operation's workstation (name + Active-Status image) for the card header.
workstation = workstation_map.get(row.name, {})
row.workstation = workstation.get("workstation")
row.workstation_name = workstation.get("workstation_name")
row.workstation_image = workstation.get("image")
row.current_operation = workstation.get("operation")
def _get_current_workstation_map(wo_names: list[str]) -> dict[str, dict]:
"""Map each Work Order to its current operation's workstation (name + Active-Status image).
The "current" operation is the first not-yet-completed operation in the routing (by idx);
if every operation is complete, the last one is used so finished cards still show a workstation.
"""
if not wo_names:
return {}
operations = frappe.get_all(
"Work Order Operation",
filters={"parent": ["in", wo_names]},
fields=["parent", "idx", "operation", "status", "workstation"],
order_by="parent, idx",
)
ops_by_wo: dict[str, list] = {}
for op in operations:
ops_by_wo.setdefault(op.parent, []).append(op)
chosen_by_wo = {}
workstations = set()
for wo_name, ops in ops_by_wo.items():
current = next((op for op in ops if (op.status or "") != "Completed"), ops[-1])
if current.workstation:
chosen_by_wo[wo_name] = current
workstations.add(current.workstation)
ws_details = {}
if workstations:
for ws in frappe.get_all(
"Workstation",
filters={"name": ["in", list(workstations)]},
fields=["name", "workstation_name", "on_status_image"],
):
ws_details[ws.name] = ws
result = {}
for wo_name, op in chosen_by_wo.items():
detail = ws_details.get(op.workstation, {})
result[wo_name] = {
"workstation": op.workstation,
"workstation_name": detail.get("workstation_name") or op.workstation,
"image": detail.get("on_status_image"),
"operation": op.operation,
}
return result
def _get_job_card_status_counts(wo_names: list[str]) -> dict[str, dict]:
"""One batched query → {work_order: {by_status: {status: n}, total, completed}} for the WO cards."""
if not wo_names:
return {}
rows = frappe.get_all(
"Job Card",
filters={"work_order": ["in", wo_names], "docstatus": ["<", 2]},
fields=["work_order", "status"],
)
result: dict[str, dict] = {}
for row in rows:
entry = result.setdefault(
row.work_order, {"by_status": {}, "total": 0, "completed": 0, "in_progress": 0}
)
status = "Not Started" if (row.status or "Open") == "Open" else row.status
entry["by_status"][status] = entry["by_status"].get(status, 0) + 1
entry["total"] += 1
# "To Manufacture" = operation done, only the Manufacture Stock Entry is pending — count it
# as completed so the work order's progress bar reflects the finished operation.
if status in ("Completed", "Submitted", "To Manufacture"):
entry["completed"] += 1
elif status == "Work In Progress":
entry["in_progress"] += 1
return result
@frappe.whitelist()
def get_data(workstation: str | None = None, work_order: str | None = None):
"""
Returns job-card data for the Shop Floor page.
When `work_order` is set it wins over `workstation` and the result spans every
operation of that Work Order (including subcontracted job cards). When only
`workstation` is set, the result is the open + in-progress job cards for that
workstation, excluding subcontracted.
"""
if not (workstation or work_order):
return {"job_cards": [], "capacity": 1, "mode": None}
if not frappe.has_permission("Job Card", "read"):
return {"job_cards": [], "capacity": 1, "mode": None}
filters, mode = _build_job_card_filters(workstation, work_order)
jc_data = _fetch_job_cards(filters, mode)
_enrich_job_cards(jc_data)
capacity, oee = 1, None
if mode == "workstation":
capacity = frappe.db.get_value("Workstation", workstation, "production_capacity") or 1
oee = get_workstation_oee(workstation)
return {
"job_cards": jc_data,
"capacity": capacity,
"mode": mode,
"oee": oee,
"user_employee": frappe.db.get_value("Employee", {"user_id": frappe.session.user}, "name"),
"today_sessions": get_today_sessions(workstation, work_order),
}
def _build_job_card_filters(workstation, work_order):
"""Filters + mode for the job-card query. work_order spans all ops; workstation is operator view."""
filters = {"docstatus": ("<", 2)}
if work_order:
filters["work_order"] = work_order
return filters, "work_order"
filters["workstation"] = workstation
filters["is_subcontracted"] = 0
filters["status"] = ["!=", "Stopped"]
return filters, "workstation"
def _fetch_job_cards(filters, mode):
"""Job cards matching filters. In workstation mode only drafts matter — submitted JCs are
done from MES's perspective and missed ones are picked up via the standard Job Card list.
In work_order mode the whole routing is shown (incl. completed/submitted job cards), ordered
by the operation sequence so the operator reads them in manufacturing order.
"""
order_by = (
"sequence_id asc, expected_start_date, expected_end_date"
if mode == "work_order"
else "expected_start_date, expected_end_date"
)
# Drafts are the operator's working set; submitted "To Manufacture" cards are also kept so
# the station shows what still needs a Manufacture Stock Entry (its own section, client-side).
# This must be part of the query, not a post-filter: a busy workstation's history would
# otherwise fill the row limit with old submitted cards and hide the active drafts.
or_filters = [["docstatus", "=", 0], ["status", "=", "To Manufacture"]] if mode == "workstation" else None
return frappe.get_all(
"Job Card",
fields=JOB_CARD_FIELDS,
filters=filters,
or_filters=or_filters,
order_by=order_by,
limit=50,
)
def _enrich_job_cards(jc_data):
"""Decorate every row with display + material-availability data for the page."""
job_card_names = [row.name for row in jc_data]
time_logs = get_time_logs(job_card_names) if job_card_names else {}
allow_excess_transfer = frappe.db.get_single_value("Manufacturing Settings", "job_card_excess_transfer")
for row in jc_data:
_enrich_job_card_row(row, time_logs, allow_excess_transfer)
def _enrich_job_card_row(row, time_logs, allow_excess_transfer):
"""Attach status label, item image/uom, time logs and material availability to one row."""
if row.status == "Open":
row.status = "Not Started"
item_code = row.finished_good or row.production_item
row.fg_uom = frappe.get_cached_value("Item", item_code, "stock_uom") if item_code else None
row.item_image = frappe.get_cached_value("Item", item_code, "image") if item_code else None
row.status_colour = get_status_color(row.status)
row.time_logs = time_logs.get(row.name, [])
row.make_material_request = bool(row.for_quantity > row.transferred_qty or allow_excess_transfer)
# Required vs transferred + on-hand in source — operator sees shortages before starting work.
row.materials = get_job_card_materials(row.name)
# Guided execution: per-operation work instructions + quality-check state for the card.
row.instructions = _get_operation_instructions(row.operation)
row.qc = _get_job_card_qc(row)
def _get_operation_instructions(operation: str | None) -> dict | None:
"""Description + rich Work Instructions from the Operation master, for the card's
Instructions panel. Returns None when the operation has neither, so the panel stays hidden.
`work_instruction` is a Text Editor field (HTML) — Frappe bleach-sanitizes it on save, so it
is safe to render as-is on the client. `description` is plain text and must be escaped there.
"""
if not operation:
return None
op = frappe.get_cached_value("Operation", operation, ["description", "work_instruction"], as_dict=True)
if not op:
return None
description = (op.description or "").strip()
work_instruction = (op.work_instruction or "").strip()
if not description and not work_instruction:
return None
return {"description": description, "work_instruction": work_instruction}
def _get_job_card_qc(row) -> dict:
"""Quality-check state for a job card row: whether an inspection is required before submit,
which template to use, and any inspection already linked (name + status + docstatus).
"Required" mirrors Job Card.validate_inspection() — BOM inspection_required AND the Work Order
Operation's quality_inspection_required. When only a template is configured the check is
offered but not enforced.
"""
required = bool(
row.get("bom_no")
and frappe.get_cached_value("BOM", row.bom_no, "inspection_required")
and row.get("operation_id")
and frappe.db.get_value("Work Order Operation", row.operation_id, "quality_inspection_required")
)
template = row.get("quality_inspection_template")
if not template and row.get("operation"):
template = frappe.get_cached_value("Operation", row.operation, "quality_inspection_template")
info = {
"required": required,
"template": template,
"has_checklist": bool(template),
"name": None,
"status": None,
"docstatus": None,
}
if row.get("quality_inspection"):
qi = frappe.db.get_value(
"Quality Inspection", row.quality_inspection, ["name", "status", "docstatus"], as_dict=True
)
if qi:
info.update({"name": qi.name, "status": qi.status, "docstatus": qi.docstatus})
return info
def get_job_card_materials(job_card: str) -> list[dict]:
"""Required vs transferred + on-hand stock for each raw material in the source warehouse.
Powers the active-job Materials side panel — operator sees shortages before starting work.
"""
items = frappe.get_all(
"Job Card Item",
filters={"parent": job_card},
fields=["item_code", "item_name", "source_warehouse", "required_qty", "transferred_qty", "uom"],
order_by="idx",
)
if not items:
return []
on_hand_map = _get_on_hand_map(items)
return [_build_material_row(it, on_hand_map) for it in items]
def _get_on_hand_map(items) -> dict[tuple[str, str], float]:
"""Map (item_code, warehouse) → on-hand qty via batched Bin lookups."""
pairs = {(it.item_code, it.source_warehouse) for it in items if it.source_warehouse}
if not pairs:
return {}
bin_rows = frappe.get_all(
"Bin",
filters={
"item_code": ["in", list({p[0] for p in pairs})],
"warehouse": ["in", list({p[1] for p in pairs})],
},
fields=["item_code", "warehouse", "actual_qty"],
)
return {(b.item_code, b.warehouse): flt(b.actual_qty) for b in bin_rows}
def _build_material_row(it, on_hand_map) -> dict:
"""One material entry with shortage + status pill for the side panel."""
required = flt(it.required_qty)
transferred = flt(it.transferred_qty)
on_hand = on_hand_map.get((it.item_code, it.source_warehouse), 0.0)
shortage = max(required - transferred, 0.0)
if transferred >= required:
status = "ready"
elif on_hand >= shortage:
status = "available"
else:
status = "short"
return {
"item_code": it.item_code,
"item_name": it.item_name or it.item_code,
"source_warehouse": it.source_warehouse,
"required_qty": required,
"transferred_qty": transferred,
"on_hand_qty": on_hand,
"shortage": shortage,
"uom": it.uom or "",
"status": status,
}
def get_today_sessions(workstation: str | None, work_order: str | None) -> list[dict]:
"""Submitted job cards finalized today — used for the bottom 'Today's Sessions' strip.
Filtered on docstatus=1 only (draft/cancelled excluded). The status pill follows the
job card's own status (e.g. Work In Progress → orange, Completed → green).
"""
filters = _today_sessions_filters(workstation, work_order)
if filters is None:
return []
rows = frappe.get_all(
"Job Card",
filters=filters,
fields=TODAY_SESSION_FIELDS,
order_by="modified desc",
limit=10,
)
for r in rows:
item_code = r.finished_good or r.production_item
r.item_image = frappe.get_cached_value("Item", item_code, "image") if item_code else None
r.status_colour = get_status_color(r.status)
return rows
def _today_sessions_filters(workstation, work_order) -> dict | None:
"""Submitted-today filter scoped to a work order or workstation; None if neither given.
"To Manufacture" cards are excluded — they aren't finalized yet (Manufacture Stock Entry
pending) and get their own section, so they shouldn't appear among finished sessions.
"""
filters = {
"docstatus": 1,
"modified": [">=", get_datetime(f"{getdate()} 00:00:00")],
"status": ["!=", "To Manufacture"],
}
if work_order:
filters["work_order"] = work_order
elif workstation:
filters["workstation"] = workstation
else:
return None
return filters
def get_workstation_oee(workstation: str) -> dict | None:
"""
OEE = Availability X Performance X Quality, computed for today only.
Caveat: without a downtime-reason capture step, Availability is just
(actual_run_time / scheduled_time) — it cannot distinguish planned breaks
from unplanned breakdowns. The number is directional, not audit-grade.
"""
today = getdate()
scheduled_min = flt(frappe.db.get_value("Workstation", workstation, "total_working_hours")) * 60
actual_run_min, ideal_min = _get_run_and_ideal_minutes(workstation, today)
completed_jcs = _get_completed_jcs_today(workstation, today)
# No activity at all today — nothing to display.
if actual_run_min == 0 and not completed_jcs:
return None
return _build_oee(scheduled_min, actual_run_min, ideal_min, completed_jcs)
def _get_run_and_ideal_minutes(workstation, today) -> tuple[float, float]:
"""Sum actual run minutes (clipped to today) and the ideal minutes for produced qty."""
today_start = get_datetime(f"{today} 00:00:00")
today_end = get_datetime(f"{today} 23:59:59")
now = now_datetime()
actual_run_min = 0.0
ideal_min = 0.0
for log in _get_oee_time_logs(workstation, today_start, today_end):
# Clip the log's interval to today's window for fair attribution.
start = max(get_datetime(log.from_time), today_start)
end = min(get_datetime(log.to_time) if log.to_time else now, today_end)
if end > start:
actual_run_min += time_diff_in_seconds(end, start) / 60
if log.completed_qty and log.for_quantity and log.time_required:
ideal_min += (flt(log.time_required) / flt(log.for_quantity)) * flt(log.completed_qty)
return actual_run_min, ideal_min
def _get_oee_time_logs(workstation, today_start, today_end) -> list[dict]:
"""Job Card time logs whose interval overlaps today's window."""
tl = frappe.qb.DocType("Job Card Time Log")
jc = frappe.qb.DocType("Job Card")
return (
frappe.qb.from_(tl)
.inner_join(jc)
.on(jc.name == tl.parent)
.select(tl.from_time, tl.to_time, tl.completed_qty, jc.for_quantity, jc.time_required)
.where(jc.workstation == workstation)
.where(jc.docstatus < 2)
.where(tl.from_time <= today_end)
.where(tl.to_time.isnull() | (tl.to_time >= today_start))
).run(as_dict=True)
def _get_completed_jcs_today(workstation, today) -> list[dict]:
"""Job cards completed/submitted today — process loss is finalized at submission."""
jc = frappe.qb.DocType("Job Card")
return (
frappe.qb.from_(jc)
.select(jc.total_completed_qty, jc.process_loss_qty)
.where(jc.workstation == workstation)
.where(jc.status.isin(["Completed", "Submitted"]))
.where(Date(jc.modified) == today)
).run(as_dict=True)
def _build_oee(scheduled_min, actual_run_min, ideal_min, completed_jcs) -> dict:
"""Combine the three OEE factors into the response payload."""
total_completed = sum(flt(j.total_completed_qty) for j in completed_jcs)
total_loss = sum(flt(j.process_loss_qty) for j in completed_jcs)
availability = min(actual_run_min / scheduled_min, 1.0) if scheduled_min > 0 else None
performance = min(ideal_min / actual_run_min, 1.0) if actual_run_min > 0 else 0.0
quality = max(total_completed - total_loss, 0.0) / total_completed if total_completed > 0 else 1.0
# OEE requires all three factors; without a schedule, Availability is unknown.
oee_val = round(availability * performance * quality * 100, 1) if availability is not None else None
return {
"oee": oee_val,
"availability": round(availability * 100, 1) if availability is not None else None,
"performance": round(performance * 100, 1),
"quality": round(quality * 100, 1),
}

3
erpnext/patches.txt
View File

@@ -496,8 +496,5 @@ erpnext.patches.v16_0.rename_ar_ap_ageing_filter
erpnext.patches.v16_0.fix_subcontracting_titles
erpnext.patches.v16_0.backfill_repost_accounting_ledger_status
erpnext.patches.v16_0.merge_seeded_item_group_root
<<<<<<< HEAD
erpnext.patches.v16_0.rename_italy_customer_name_fields
=======
erpnext.patches.v16_0.set_stock_uom_in_job_card
>>>>>>> 07ac4d83ef (feat(job_card): print quantities with their stock uom (#57689))

1758
erpnext/public/js/shop_floor/shop_floor.js
View File

File diff suppressed because it is too large Load Diff

75
erpnext/stock/doctype/stock_entry/stock_entry.py
View File

@@ -296,6 +296,7 @@ class StockEntry(StockController, SubcontractingInwardController):
self.validate_batch()
self.validate_inspection()
self.validate_fg_completed_qty()
self.validate_job_card_pending_production()
self.validate_difference_account()
self.set_job_card_data()
self.validate_job_card_item()
@@ -3192,21 +3193,11 @@ class StockEntry(StockController, SubcontractingInwardController):
return
precision = self.precision("process_loss_qty")
if self.work_order:
data = frappe.get_all(
"Work Order Operation",
filters={"parent": self.work_order},
fields=[{"MAX": "process_loss_qty", "as": "process_loss_qty"}],
)
process_loss_qty = self.get_pending_process_loss_qty()
if process_loss_qty and flt(self.process_loss_qty, precision) != flt(process_loss_qty, precision):
self.process_loss_qty = flt(process_loss_qty, precision)
if data and data[0].process_loss_qty:
process_loss_qty = data[0].process_loss_qty
if flt(self.process_loss_qty, precision) != flt(process_loss_qty, precision):
self.process_loss_qty = flt(process_loss_qty, precision)
frappe.msgprint(
_("The Process Loss Qty has reset as per job cards Process Loss Qty"), alert=True
)
frappe.msgprint(_("The Process Loss Qty has reset as per job cards Process Loss Qty"), alert=True)
if not self.process_loss_percentage and not self.process_loss_qty:
self.process_loss_percentage = frappe.get_cached_value(
@@ -3222,6 +3213,62 @@ class StockEntry(StockController, SubcontractingInwardController):
(flt(self.process_loss_qty) / flt(self.fg_completed_qty)) * 100
)
def validate_job_card_pending_production(self):
"""A draft created before other entries were submitted must not book more than the job
card still has left; without this, a stale draft over-produces the finished good."""
if self.purpose != "Manufacture" or not self.job_card:
return
if self._action == "update_after_submit":
return
job_card = frappe.get_doc("Job Card", self.job_card)
if job_card.is_corrective_job_card or job_card.is_subcontracted:
return
precision = frappe.get_precision("Stock Entry Detail", "qty")
pending_qty = flt(
flt(job_card.for_quantity)
- flt(job_card.pending_qty)
- flt(job_card.manufactured_qty)
- flt(job_card.get_consumed_process_loss()),
precision,
)
finished_qty = flt(sum(flt(d.transfer_qty) for d in self.items if d.is_finished_item), precision)
entry_qty = flt(finished_qty + flt(self.process_loss_qty), precision)
if entry_qty > pending_qty:
item_code = job_card.finished_good or job_card.production_item
uom = frappe.get_cached_value("Item", item_code, "stock_uom")
frappe.throw(
_(
"The Job Card {0} has only {1} left to produce, but this entry books {2} ({3} finished goods and {4} process loss). Cancel or update its other manufacture entries first."
).format(
frappe.bold(self.job_card),
frappe.bold(f"{pending_qty} {uom}"),
frappe.bold(f"{entry_qty} {uom}"),
f"{finished_qty} {uom}",
f"{flt(self.process_loss_qty, precision)} {uom}",
)
)
def get_pending_process_loss_qty(self):
"""Loss this entry should still book: the job card's unbooked loss when the entry
belongs to one, else the largest operation loss on the work order (legacy flow)."""
if self.job_card:
job_card = frappe.get_doc("Job Card", self.job_card)
return max(flt(job_card.process_loss_qty) - flt(job_card.get_consumed_process_loss()), 0)
if self.work_order:
data = frappe.get_all(
"Work Order Operation",
filters={"parent": self.work_order},
fields=[{"MAX": "process_loss_qty", "as": "process_loss_qty"}],
)
return flt(data[0].process_loss_qty) if data else 0
return 0
def set_work_order_details(self):
if not getattr(self, "pro_doc", None):
self.pro_doc = frappe._dict()

28
erpnext/stock/doctype/stock_entry_type/stock_entry_type.py
View File

@@ -125,6 +125,8 @@ class ManufactureEntry:
if backflush_based_on != "BOM":
available_serial_batches = self.get_transferred_serial_batches()
production_share = self.get_production_share()
items_to_remove = []
for item_code, _dict in item_dict.items():
_dict.from_warehouse = self.source_wh.get(item_code) or self.wip_warehouse
_dict.to_warehouse = ""
@@ -138,11 +140,33 @@ class ManufactureEntry:
_dict.qty = calculated_qty
self.update_available_serial_batches(_dict, available_serial_batches)
elif self.skip_material_transfer:
set_previous_operation_serial_batch(self.stock_entry, _dict)
else:
remaining_qty = max(flt(_dict.qty) - flt(_dict.consumed_qty), 0)
_dict.qty = min(flt(_dict.qty) * production_share, remaining_qty)
if not _dict.qty:
items_to_remove.append(item_code)
continue
if self.skip_material_transfer:
set_previous_operation_serial_batch(self.stock_entry, _dict)
for item_code in items_to_remove:
item_dict.pop(item_code)
self.stock_entry.add_to_stock_entry_detail(item_dict)
def get_production_share(self):
"""Fraction of the job card's production this entry accounts for; raw materials are
generated proportionally so several partial entries never consume more than required."""
for_quantity, pending_qty = frappe.db.get_value(
"Job Card", self.job_card, ["for_quantity", "pending_qty"]
)
qty_to_produce = flt(for_quantity) - flt(pending_qty)
if not qty_to_produce:
return 1
return min(flt(self.for_quantity) / qty_to_produce, 1)
def parse_available_serial_batches(self, item_dict, available_serial_batches):
key = (item_dict.item_code, item_dict.from_warehouse)
if key not in available_serial_batches: