chore: resolve conflict

This commit is contained in:
Mihir Kandoi
2026-08-09 21:07:17 +05:30
parent 27625a6f1f
commit a5e4bbd436
8 changed files with 66 additions and 1081 deletions

View File

@@ -305,10 +305,9 @@ class BOM(WebsiteGenerator):
self.set_fg_cost_allocation()
self.validate_total_cost_allocation()
<<<<<<< HEAD
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
@@ -321,7 +320,6 @@ class BOM(WebsiteGenerator):
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")
>>>>>>> 1e2e87daac (fix: derive operation FG items before material expansion, keep the final one the BOM's item)
def validate_semi_finished_goods(self):
if not self.track_semi_finished_goods or not self.operations:
@@ -866,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
@@ -887,27 +878,6 @@ class BOM(WebsiteGenerator):
self.save()
<<<<<<< HEAD
=======
def _add_raw_material_row(self, operation_row_id, row):
row = parse_json(row)
row.update(get_item_details(row.get("item_code")))
row.operation_row_id = operation_row_id
item_row = self.get_item_data(row.item_code, operation_row_id)
if item_row:
item_row.qty = row.get("qty")
else:
row.idx = None
row.name = None
row.do_not_explode = 1
row.is_sub_assembly_item = self.is_sub_assembly_item(row.item_code)
self.append("items", row)
>>>>>>> 24f1f3dea8 (fix: add raw material to its operation even when another operation uses the item)
def is_sub_assembly_item(self, item_code):
if not self.operations:
return False

View File

@@ -1041,32 +1041,7 @@ class JobCard(Document):
)
def update_work_order_data(self, for_quantity, process_loss_qty, pending_qty, time_in_mins, wo):
<<<<<<< HEAD
workstation_hour_rate = frappe.get_value("Workstation", self.workstation, "hour_rate")
=======
time_data = self.get_operation_time_data()
for data in wo.operations:
if data.get("name") == self.operation_id:
self.update_wo_operation_row(
data, for_quantity, process_loss_qty, pending_qty, time_in_mins, time_data
)
wo.flags.ignore_validate_update_after_submit = True
wo.update_operation_status()
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"
wo.save()
def get_operation_time_data(self):
>>>>>>> 0eb61c9fac (fix: roll up process loss to the work order for semi finished goods)
jc = frappe.qb.DocType("Job Card")
jctl = frappe.qb.DocType("Job Card Time Log")
@@ -1101,6 +1076,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"
@@ -1367,57 +1345,6 @@ class JobCard(Document):
if not (self.work_order and self.sequence_id):
return
<<<<<<< HEAD
=======
current_operation_qty = self.get_current_operation_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)
def get_previous_operations(self):
previous_operations = frappe.get_all(
"Work Order Operation",
fields=["name", "operation", "status", "completed_qty", "sequence_id", "finished_good"],
filters={"docstatus": 1, "parent": self.work_order, "sequence_id": ("<", self.sequence_id)},
order_by="sequence_id, idx",
)
if self.track_semi_finished_goods and previous_operations:
totals = self.get_manufactured_qty_per_operation([row.name for row in previous_operations])
for row in previous_operations:
operation_totals = totals.get(row.name)
row.manufactured_qty = flt(operation_totals and operation_totals.manufactured_qty)
row.process_loss_qty = flt(operation_totals and operation_totals.process_loss_qty)
return previous_operations
def get_manufactured_qty_per_operation(self, operation_ids):
job_card = frappe.qb.DocType("Job Card")
data = (
frappe.qb.from_(job_card)
.select(
job_card.operation_id,
Sum(job_card.manufactured_qty).as_("manufactured_qty"),
Sum(job_card.process_loss_qty).as_("process_loss_qty"),
)
.where(
(job_card.work_order == self.work_order)
& (job_card.docstatus == 1)
& (IfNull(job_card.is_corrective_job_card, 0) == 0)
& (job_card.operation_id.isin(operation_ids))
)
.groupby(job_card.operation_id)
).run(as_dict=True)
return {row.operation_id: row for row in data}
def get_current_operation_completed_qty(self):
>>>>>>> 1e22695eae (fix: stop asking for a manufacturing entry when process loss explains the shortfall)
current_operation_qty = 0.0
data = self.get_current_operation_data()
if data and len(data) > 0:
@@ -1453,7 +1380,6 @@ class JobCard(Document):
OperationSequenceError,
)
<<<<<<< HEAD
if row.completed_qty < current_operation_qty:
frappe.throw(
_(
@@ -1465,49 +1391,6 @@ class JobCard(Document):
bold(row.operation),
)
)
=======
if not manufactured_qty:
frappe.throw(
_(
"Job Card {0}: As per the sequence of the operations in the work order {1}, submit the manufacturing entry for the operation {2} before the operation {3}."
).format(
bold(self.name),
bold(get_link_to_form("Work Order", self.work_order)),
bold(row.operation),
bold(self.operation),
),
OperationSequenceError,
)
if manufactured_qty >= current_operation_qty:
return
if manufactured_qty + flt(row.process_loss_qty) >= current_operation_qty:
frappe.throw(
_(
"The completed quantity {0} of an operation {1} cannot be greater than the manufactured quantity {2} of a previous operation {3}, as {4} was booked as process loss there."
).format(
bold(self.get_qty_with_uom(current_operation_qty)),
bold(self.operation),
bold(self.get_qty_with_uom(manufactured_qty, row.finished_good)),
bold(row.operation),
bold(self.get_qty_with_uom(flt(row.process_loss_qty), row.finished_good)),
),
OperationSequenceError,
)
>>>>>>> 1e22695eae (fix: stop asking for a manufacturing entry when process loss explains the shortfall)
frappe.throw(
_(
"The completed quantity {0} of an operation {1} cannot be greater than the manufactured quantity {2} of a previous operation {3}. Submit the manufacturing entry for the operation {3} first."
).format(
bold(self.get_qty_with_uom(current_operation_qty)),
bold(self.operation),
bold(self.get_qty_with_uom(manufactured_qty, row.finished_good)),
bold(row.operation),
),
OperationSequenceError,
)
def validate_work_order(self):
if self.is_work_order_closed():
@@ -1684,33 +1567,27 @@ class JobCard(Document):
_("Job Card {0} has been completed").format(get_link_to_form("Job Card", self.name))
)
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
<<<<<<< HEAD
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),
=======
consumed_process_loss = self.get_consumed_process_loss()
return ManufactureEntry(
{
"for_quantity": self.get_qty_to_produce() - self.manufactured_qty - consumed_process_loss,
"for_quantity": self.for_quantity
- self.pending_qty
- self.manufactured_qty
- consumed_process_loss,
"process_loss_qty": max(self.process_loss_qty - consumed_process_loss, 0),
>>>>>>> b8dd886cd4 (fix: generate the next manufacture entry net of booked process loss)
"job_card": self.name,
"skip_material_transfer": self.skip_material_transfer,
"backflush_from_wip_warehouse": self.backflush_from_wip_warehouse,

View File

@@ -1265,89 +1265,6 @@ class TestJobCard(ERPNextTestSuite):
8,
)
<<<<<<< HEAD
=======
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 = self.get_first_job_card(work_order.name)
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, "To Manufacture")
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_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
@@ -1400,7 +1317,16 @@ class TestJobCard(ERPNextTestSuite):
make_stock_entry(item_code=rm, target=warehouse, qty=100, basic_rate=100)
job_card = self.get_first_job_card(work_order.name)
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()
@@ -1528,7 +1454,6 @@ class TestJobCard(ERPNextTestSuite):
work_order.reload()
self.assertEqual(flt(work_order.process_loss_qty), 2)
# Operation A handed over only 8 units, so the final operation works on 8.
jc_b = get_job_card("Intermediate Loss Op B")
jc_b.for_quantity = 8
for row in jc_b.items:
@@ -1546,145 +1471,6 @@ class TestJobCard(ERPNextTestSuite):
self.assertEqual(flt(work_order.process_loss_qty), 2)
self.assertEqual(work_order.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)
>>>>>>> 24de81f9fa (test: work order process loss for semi finished goods)
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
@@ -2530,6 +2316,27 @@ 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 create_bom_with_multiple_operations():
"Create a BOM with multiple operations and Material Transfer against Job Card"
@@ -2592,141 +2399,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))
def test_previous_operation_shortfall_from_process_loss_gets_the_right_message(self):
jc = frappe.new_doc("Job Card")
jc.operation = "_Test Painting"
jc.stock_uom = "Nos"
row = frappe._dict(
operation="_Test Assembly", manufactured_qty=8, process_loss_qty=2, finished_good=None
)
with self.assertRaises(OperationSequenceError) as loss_error:
jc.validate_previous_operation_manufactured_qty(row, 10)
self.assertIn("process loss", str(loss_error.exception))
row.process_loss_qty = 0
with self.assertRaises(OperationSequenceError) as pending_error:
jc.validate_previous_operation_manufactured_qty(row, 10)
self.assertIn("Submit the manufacturing entry", str(pending_error.exception))
jc.validate_previous_operation_manufactured_qty(row, 8)
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)
>>>>>>> 4b3904c6d7 (test: semi FG job card is exempt from the legacy transfer qty check)

View File

@@ -1,471 +0,0 @@
# Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
"""Status and quantity-rollup logic for Work Order.
Extracted from work_order.py. ``StatusService`` wraps a Work Order document
(composition); work_order.py keeps thin delegating stubs so the many external
callers (job cards, sales orders, production plans, patches) keep working.
"""
import frappe
from frappe import _
from frappe.query_builder.functions import Sum
from frappe.utils import cint, flt, get_link_to_form
from erpnext.stock.stock_balance import get_planned_qty, update_bin_qty
_QTY_PURPOSES = (
("Manufacture", "produced_qty"),
("Material Transfer for Manufacture", "material_transferred_for_manufacturing"),
("Material Transfer for Manufacture", "additional_transferred_qty"),
)
class StatusService:
def __init__(self, doc):
self.doc = doc
def validate_work_order_against_so(self):
from erpnext.manufacturing.doctype.work_order.work_order import OverProductionError
total_qty = flt(self._ordered_qty_against_so()) + flt(self.doc.qty)
so_qty = flt(self._so_item_qty()) + flt(self._packed_item_qty())
allowance_percentage = flt(
frappe.db.get_single_value("Manufacturing Settings", "overproduction_percentage_for_sales_order")
)
if total_qty <= so_qty + (allowance_percentage / 100 * so_qty):
return
frappe.throw(
_("Cannot produce more Item {0} than Sales Order quantity {1} {2}").format(
get_link_to_form("Item", self.doc.production_item),
frappe.bold(so_qty),
frappe.bold(frappe.get_value("Item", self.doc.production_item, "stock_uom")),
),
OverProductionError,
)
def _ordered_qty_against_so(self):
wo = frappe.qb.DocType("Work Order")
return (
frappe.qb.from_(wo)
.select(Sum(wo.qty - wo.process_loss_qty))
.where(
(wo.production_item == self.doc.production_item)
& (wo.sales_order == self.doc.sales_order)
& (wo.docstatus == 1)
& (wo.status != "Closed")
& (wo.name != self.doc.name)
)
).run()[0][0]
def _so_item_qty(self):
so_item = frappe.qb.DocType("Sales Order Item")
return (
frappe.qb.from_(so_item)
.select(Sum(so_item.stock_qty))
.where(
(so_item.parent == self.doc.sales_order)
& (so_item.item_code == self.doc.production_item)
& (so_item.docstatus == 1)
)
).run()[0][0]
def _packed_item_qty(self):
packed_item = frappe.qb.DocType("Packed Item")
return (
frappe.qb.from_(packed_item)
.select(Sum(packed_item.qty))
.where(
(packed_item.parent == self.doc.sales_order)
& (packed_item.parenttype == "Sales Order")
& (packed_item.item_code == self.doc.production_item)
& (packed_item.docstatus == 1)
)
).run()[0][0]
def update_status(self, status=None):
"""Update status of work order if unknown"""
if self.doc.docstatus == 1:
# Refresh material_transferred_for_manufacturing before deciding status so pick-list-
# driven transfers (where this qty is derived from item transfers, not fg_completed_qty)
# are reflected immediately, instead of only after the next status update call.
self.doc.refresh_material_transferred_for_manufacturing()
if self.doc.status != "Closed":
if status not in ["Stopped", "Closed"]:
status = self.get_status(status)
if status != self.doc.status:
self.doc.db_set("status", status)
self.doc.update_required_items()
return status or self.doc.status
def get_status(self, status=None):
"""Return the status based on stock entries against this work order"""
status = status or self.doc.status
if self.doc.docstatus == 0:
status = "Draft"
elif self.doc.docstatus == 1:
status = self._submitted_status(status)
else:
status = "Cancelled"
if self._is_partial_skip_transfer():
status = "In Process"
if status != "Completed" and not all(d.status == "Pending" for d in self.doc.operations):
status = "In Process"
if status == "Not Started" and self.doc.reserve_stock:
status = self._reservation_status(status)
return status
def _submitted_status(self, status):
if status in ["Closed", "Stopped"]:
return status
status = (
"In Process"
if flt(self.doc.material_transferred_for_manufacturing) > 0
or self.doc.skip_transfer
or self._has_transferred_material()
else "Not Started"
)
precision = frappe.get_precision("Work Order", "produced_qty")
total_qty = flt(self.doc.produced_qty, precision) + flt(self.doc.process_loss_qty, precision)
if flt(total_qty, precision) >= flt(self.doc.qty, precision):
status = "Completed"
return status
def _has_transferred_material(self):
"""True if any raw material was transferred against this work order via a pick list
or a material request (these leave material_transferred_for_manufacturing at 0 via
the min-fraction rule)."""
ste = frappe.qb.DocType("Stock Entry")
ste_child = frappe.qb.DocType("Stock Entry Detail")
mr_child = frappe.qb.DocType("Stock Entry Detail")
# Stock Entry only carries `material_request` at the child-row level, so a Stock
# Entry is "MR-sourced" if *any* of its rows link back to a Material Request; once
# that's established, sum every row's transfer_qty, not just the linked ones (a
# manually appended extra row on the same entry has no material_request of its own).
mr_sourced_stock_entries = (
frappe.qb.from_(mr_child).select(mr_child.parent).where(mr_child.material_request.isnotnull())
)
qty = (
frappe.qb.from_(ste)
.inner_join(ste_child)
.on(ste_child.parent == ste.name)
.select(Sum(ste_child.transfer_qty))
.where(
(ste.work_order == self.doc.name)
& (ste.docstatus == 1)
& (ste.purpose == "Material Transfer for Manufacture")
& (ste.is_return == 0)
& (ste.pick_list.isnotnull() | ste.name.isin(mr_sourced_stock_entries))
)
).run()[0][0]
return flt(qty) > 0
def _is_partial_skip_transfer(self):
return bool(
self.doc.skip_transfer
and self.doc.produced_qty
and self.doc.qty > (flt(self.doc.produced_qty) + flt(self.doc.process_loss_qty))
)
def _reservation_status(self, status):
for row in self.doc.required_items:
if not row.stock_reserved_qty:
continue
if row.stock_reserved_qty >= row.required_qty:
status = "Stock Reserved"
else:
return "Stock Partially Reserved"
return status
def update_work_order_qty(self):
"""Update Manufactured Qty and Material Transferred for Qty based on Stock Entry"""
if self.doc.track_semi_finished_goods:
return
for purpose, fieldname in _QTY_PURPOSES:
self._update_qty_for_purpose(purpose, fieldname)
if self.doc.production_plan:
self.set_produced_qty_for_sub_assembly_item()
self.update_production_plan_status()
if self.doc.additional_transferred_qty:
self.doc.validate_additional_transferred_qty()
def _update_qty_for_purpose(self, purpose, fieldname):
from erpnext.manufacturing.doctype.work_order.work_order import StockOverProductionError
if self._skip_transfer_purpose(purpose):
return
qty = self.get_transferred_or_manufactured_qty(purpose, fieldname)
completed_qty = self.doc.qty + (self._qty_allowance(purpose) / 100 * self.doc.qty)
if qty > completed_qty:
frappe.throw(
_("{0} ({1}) cannot be greater than planned quantity ({2}) in Work Order {3}").format(
_(self.doc.meta.get_label(fieldname)), qty, completed_qty, self.doc.name
),
StockOverProductionError,
)
self.doc.db_set(fieldname, qty)
self.set_process_loss_qty()
self._update_produced_qty_in_so()
def _skip_transfer_purpose(self, purpose):
return bool(
purpose == "Material Transfer for Manufacture"
and self.doc.operations
and self.doc.transfer_material_against == "Job Card"
)
def _qty_allowance(self, purpose):
allowance = flt(
frappe.db.get_single_value("Manufacturing Settings", "overproduction_percentage_for_work_order")
)
if not allowance and purpose == "Material Transfer for Manufacture":
allowance = flt(
frappe.db.get_single_value("Manufacturing Settings", "transfer_extra_materials_percentage")
)
return allowance
def _update_produced_qty_in_so(self):
from erpnext.selling.doctype.sales_order.sales_order import update_produced_qty_in_so_item
if (
self.doc.sales_order
and self.doc.sales_order_item
and not self.doc.production_plan_sub_assembly_item
):
update_produced_qty_in_so_item(self.doc.sales_order, self.doc.sales_order_item)
def update_disassembled_qty(self, qty, is_cancel=False):
if is_cancel:
self.doc.disassembled_qty = max(0, self.doc.disassembled_qty - qty)
else:
if self.doc.docstatus == 1:
self.doc.disassembled_qty += qty
if not is_cancel and self.doc.disassembled_qty > self.doc.produced_qty:
frappe.throw(_("Cannot disassemble more than produced quantity."))
self.doc.db_set("disassembled_qty", self.doc.disassembled_qty)
def get_transferred_or_manufactured_qty(self, purpose, fieldname):
parent = frappe.qb.DocType("Stock Entry")
is_additional = cint(fieldname == "additional_transferred_qty")
query = frappe.qb.from_(parent).where(self._stock_entry_filter(parent, purpose, is_additional))
if purpose == "Manufacture":
child = frappe.qb.DocType("Stock Entry Detail")
query = (
query.join(child)
.on(parent.name == child.parent)
.select(Sum(child.transfer_qty))
.where(child.is_finished_item == 1)
)
else:
query = query.select(Sum(parent.fg_completed_qty))
return flt(query.run()[0][0])
def _stock_entry_filter(self, parent, purpose, is_additional):
return (
(parent.work_order == self.doc.name)
& (parent.docstatus == 1)
& (parent.purpose == purpose)
& (parent.is_additional_transfer_entry == is_additional)
)
def set_process_loss_qty(self):
self.doc.db_set("process_loss_qty", self._process_loss_qty())
def _process_loss_qty(self):
if self.doc.track_semi_finished_goods:
return flt(sum(flt(row.process_loss_qty) for row in self.doc.operations))
table = frappe.qb.DocType("Stock Entry")
process_loss_qty = (
frappe.qb.from_(table)
.select(Sum(table.process_loss_qty))
.where(
(table.work_order == self.doc.name)
& (table.purpose == "Manufacture")
& (table.docstatus == 1)
)
).run()[0][0]
return flt(process_loss_qty)
def update_production_plan_status(self):
production_plan = frappe.get_doc("Production Plan", self.doc.production_plan)
produced_qty = 0
if self.doc.production_plan_item:
total_qty = frappe.get_all(
"Work Order",
fields=[{"SUM": "produced_qty", "as": "produced_qty"}],
filters={
"docstatus": 1,
"production_plan": self.doc.production_plan,
"production_plan_item": self.doc.production_plan_item,
},
as_list=1,
)
produced_qty = total_qty[0][0] if total_qty else 0
self.update_status()
production_plan.run_method("update_produced_pending_qty", produced_qty, self.doc.production_plan_item)
def update_planned_qty(self):
if self.doc.track_semi_finished_goods:
return
update_bin_qty(self.doc.production_item, self.doc.fg_warehouse, self._planned_qty_dict())
if self.doc.material_request:
mr_obj = frappe.get_doc("Material Request", self.doc.material_request)
mr_obj.update_requested_qty([self.doc.material_request_item])
def _planned_qty_dict(self):
from erpnext.manufacturing.doctype.production_plan.production_plan import (
get_reserved_qty_for_sub_assembly,
)
qty_dict = {"planned_qty": get_planned_qty(self.doc.production_item, self.doc.fg_warehouse)}
if self.doc.production_plan_sub_assembly_item and self.doc.production_plan:
qty_dict["reserved_qty_for_production_plan"] = get_reserved_qty_for_sub_assembly(
self.doc.production_item, self.doc.fg_warehouse
)
return qty_dict
def set_produced_qty_for_sub_assembly_item(self):
produced_qty = self._sub_assembly_produced_qty()
frappe.db.set_value(
"Production Plan Sub Assembly Item",
self.doc.production_plan_sub_assembly_item,
"wo_produced_qty",
produced_qty,
)
def _sub_assembly_produced_qty(self):
table = frappe.qb.DocType("Work Order")
query = (
frappe.qb.from_(table)
.select(Sum(table.produced_qty))
.where(
(table.production_plan == self.doc.production_plan)
& (table.production_plan_sub_assembly_item == self.doc.production_plan_sub_assembly_item)
& (table.docstatus == 1)
)
).run()
return flt(query[0][0]) if query else 0
def update_ordered_qty(self):
if not (
self.doc.production_plan
and (self.doc.production_plan_item or self.doc.production_plan_sub_assembly_item)
):
return
qty = self._production_plan_ordered_qty()
if self.doc.production_plan_item:
frappe.db.set_value("Production Plan Item", self.doc.production_plan_item, "ordered_qty", qty)
elif self.doc.production_plan_sub_assembly_item:
field = self.doc.production_plan_sub_assembly_item
frappe.db.set_value("Production Plan Sub Assembly Item", field, "ordered_qty", qty)
doc = frappe.get_doc("Production Plan", self.doc.production_plan)
doc.set_status()
doc.db_set("status", doc.status)
def _production_plan_ordered_qty(self):
table = frappe.qb.DocType("Work Order")
query = (
frappe.qb.from_(table)
.select(Sum(table.qty))
.where((table.production_plan == self.doc.production_plan) & (table.docstatus == 1))
)
if self.doc.production_plan_item:
query = query.where(table.production_plan_item == self.doc.production_plan_item)
elif self.doc.production_plan_sub_assembly_item:
query = query.where(
table.production_plan_sub_assembly_item == self.doc.production_plan_sub_assembly_item
)
result = query.run()
return flt(result[0][0]) if result else 0
def update_work_order_qty_in_so(self):
if (
not self.doc.sales_order and not self.doc.sales_order_item
) or self.doc.production_plan_sub_assembly_item:
return
total_bundle_qty = self._total_bundle_qty()
work_order_qty = self._sales_order_work_order_qty()
frappe.db.set_value(
"Sales Order Item",
self.doc.sales_order_item,
"work_order_qty",
flt(work_order_qty / total_bundle_qty, 2),
)
def _sales_order_work_order_qty(self):
wo = frappe.qb.DocType("Work Order")
query = (
frappe.qb.from_(wo)
.select(Sum(wo.qty))
.where((wo.sales_order == self.doc.sales_order) & (wo.docstatus == 1) & (wo.status != "Closed"))
)
if self.doc.product_bundle_item:
query = query.where(wo.product_bundle_item == self.doc.product_bundle_item)
else:
query = query.where(wo.production_item == self.doc.production_item)
qty = query.run(as_list=1)
return qty[0][0] if qty and qty[0][0] else 0
def update_work_order_qty_in_combined_so(self):
total_bundle_qty = self._total_bundle_qty()
prod_plan = frappe.get_doc("Production Plan", self.doc.production_plan)
item_reference = frappe.get_value(
"Production Plan Item", self.doc.production_plan_item, "sales_order_item"
)
for plan_reference in prod_plan.prod_plan_references:
if plan_reference.item_reference != item_reference:
continue
qty = flt(plan_reference.qty) / total_bundle_qty if self.doc.docstatus == 1 else 0.0
frappe.db.set_value("Sales Order Item", plan_reference.sales_order_item, "work_order_qty", qty)
def _total_bundle_qty(self):
if not self.doc.product_bundle_item:
return 1
pbi = frappe.qb.DocType("Product Bundle Item")
total_bundle_qty = (
frappe.qb.from_(pbi).select(Sum(pbi.qty)).where(pbi.parent == self.doc.product_bundle_item)
).run()[0][0]
# product bundle is 0 (product bundle allows 0 qty for items)
return total_bundle_qty or 1
def update_completed_qty_in_material_request(self):
if self.doc.material_request and self.doc.material_request_item:
frappe.get_doc("Material Request", self.doc.material_request).update_completed_qty(
[self.doc.material_request_item]
)

View File

@@ -4839,59 +4839,6 @@ 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)
<<<<<<< HEAD
=======
def test_transferred_qty_not_misattributed_between_item_and_its_substitute(self):
"""When one item is transferred both for itself and as a substitute for another required item,
each transfer must be credited to the right required item.
_material_transfer_qty_by_item grouped Stock Entry Detail by item_code only and picked
Max(original_item); for item B transferred once for itself (original_item NULL) and once as a
substitute for A (original_item=A), Max picked A and credited B's whole transfer to A, leaving
B at 0. Grouping by (item_code, original_item) and accumulating into the keyed dict attributes
each transfer correctly, deterministically on MariaDB and Postgres.
"""
from erpnext.manufacturing.doctype.work_order.services.required_items import RequiredItemsService
source_warehouse = "Stores - _TC"
fg_item = make_item("Test WO SelfSub FG", {"is_stock_item": 1}).name
item_a = make_item("Test WO SelfSub RM A", {"is_stock_item": 1, "allow_alternative_item": 1}).name
item_b = make_item("Test WO SelfSub RM B", {"is_stock_item": 1, "allow_alternative_item": 1}).name
# B is a registered alternative for A
if not frappe.db.exists("Item Alternative", {"item_code": item_a, "alternative_item_code": item_b}):
frappe.get_doc(
{
"doctype": "Item Alternative",
"item_code": item_a,
"alternative_item_code": item_b,
"two_way": 1,
}
).insert()
# stock B generously (covers B-for-A plus B-for-itself)
for item, qty in ((item_a, 50), (item_b, 100)):
test_stock_entry.make_stock_entry(
item_code=item, target=source_warehouse, qty=qty, basic_rate=100
)
make_bom(item=fg_item, source_warehouse=source_warehouse, raw_materials=[item_a, item_b])
wo = make_wo_order_test_record(item=fg_item, qty=10, source_warehouse=source_warehouse)
transfer = frappe.get_doc(make_stock_entry(wo.name, "Material Transfer for Manufacture", 10))
transfer.save()
# substitute B for the A line; the existing B line stays as B's own transfer
for d in transfer.items:
if d.item_code == item_a:
d.item_code = item_b
d.original_item = item_a
transfer.submit()
qty_by_item = RequiredItemsService(wo)._material_transfer_qty_by_item(is_return=0)
# B transferred as a substitute for A -> credited to A; B transferred for itself -> credited to B.
self.assertEqual(flt(qty_by_item.get(item_a)), 10.0)
self.assertEqual(flt(qty_by_item.get(item_b)), 10.0)
def test_wip_warehouse_required_when_tracking_semi_finished_goods(self):
wo = frappe.new_doc("Work Order")
wo.track_semi_finished_goods = 1
@@ -4903,9 +4850,6 @@ class TestWorkOrder(ERPNextTestSuite):
wo.wip_warehouse = "_Test Warehouse - _TC"
wo.validate_warehouse()
<<<<<<< HEAD
>>>>>>> f61f6523b9 (test: WIP warehouse required for work orders tracking semi finished goods)
=======
# the top-level target warehouse stays optional; operations may carry their own
wo.fg_warehouse = None
wo.validate_warehouse()
@@ -4913,7 +4857,6 @@ class TestWorkOrder(ERPNextTestSuite):
wo.track_semi_finished_goods = 0
self.assertRaises(frappe.ValidationError, wo.validate_warehouse)
>>>>>>> db99657c47 (test: target warehouse stays optional for semi FG work orders)
def get_reserved_entries(voucher_no, warehouse=None):
doctype = frappe.qb.DocType("Stock Reservation Entry")

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)

View File

@@ -3197,23 +3197,8 @@ class StockEntry(StockController, SubcontractingInwardController):
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)
frappe.msgprint(
_("The Process Loss Qty has been reset as per the job card's Process Loss Qty"),
alert=True,
)
frappe.msgprint(_("The Process Loss Qty has reset as per job cards Process Loss Qty"), alert=True)
<<<<<<< HEAD
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
)
=======
>>>>>>> 1b335973b7 (fix: scope manufacture entry process loss to its own job card)
if not self.process_loss_percentage and not self.process_loss_qty:
self.process_loss_percentage = frappe.get_cached_value(
"BOM", self.bom_no, "process_loss_percentage"
@@ -3243,7 +3228,8 @@ class StockEntry(StockController, SubcontractingInwardController):
precision = frappe.get_precision("Stock Entry Detail", "qty")
pending_qty = flt(
flt(job_card.get_qty_to_produce())
flt(job_card.for_quantity)
- flt(job_card.pending_qty)
- flt(job_card.manufactured_qty)
- flt(job_card.get_consumed_process_loss()),
precision,

View File

@@ -126,6 +126,7 @@ class ManufactureEntry:
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 = ""
@@ -143,11 +144,15 @@ class ManufactureEntry:
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):