diff --git a/inventory_tools/hooks.py b/inventory_tools/hooks.py index b65aa2f..612ab45 100644 --- a/inventory_tools/hooks.py +++ b/inventory_tools/hooks.py @@ -125,6 +125,7 @@ override_doctype_class = { "Delivery Note": "inventory_tools.inventory_tools.overrides.delivery_note.InventoryToolsDeliveryNote", + "Pick List": "inventory_tools.inventory_tools.overrides.pick_list.InventoryToolsPickList", "Quality Inspection": "inventory_tools.inventory_tools.overrides.quality_inspection.InventoryToolsQualityInspection", "Job Card": "inventory_tools.inventory_tools.overrides.job_card.InventoryToolsJobCard", "Production Plan": "inventory_tools.inventory_tools.overrides.production_plan.InventoryToolsProductionPlan", diff --git a/inventory_tools/inventory_tools/doctype/inventory_tools_settings/inventory_tools_settings.json b/inventory_tools/inventory_tools/doctype/inventory_tools_settings/inventory_tools_settings.json index 32fc288..f8b3ef1 100644 --- a/inventory_tools/inventory_tools/doctype/inventory_tools_settings/inventory_tools_settings.json +++ b/inventory_tools/inventory_tools/doctype/inventory_tools_settings/inventory_tools_settings.json @@ -26,6 +26,7 @@ "section_break_0", "update_warehouse_path", "prettify_warehouse_tree", + "default_route_optimization_strategy", "column_break_ddssn", "allow_alternative_workstations", "uoms_section", @@ -208,6 +209,13 @@ "fieldtype": "Check", "label": "Prettify Warehouse Tree" }, + { + "default": "Use Source Document Order", + "fieldname": "default_route_optimization_strategy", + "fieldtype": "Select", + "label": "Default Route Optimization Strategy", + "options": "Use Source Document Order\nFIFO\nLIFO\nDeplete maximum number of Bins\nDeplete minimum number of Bins" + }, { "fieldname": "quarantine_quality_control_section", "fieldtype": "Section Break", diff --git a/inventory_tools/inventory_tools/doctype/warehouse_plan/warehouse_plan.py b/inventory_tools/inventory_tools/doctype/warehouse_plan/warehouse_plan.py index 74ae600..50aad4f 100644 --- a/inventory_tools/inventory_tools/doctype/warehouse_plan/warehouse_plan.py +++ b/inventory_tools/inventory_tools/doctype/warehouse_plan/warehouse_plan.py @@ -3,9 +3,12 @@ import frappe from frappe.model.document import Document +from frappe.utils import safe_json_loads import networkx as nx import numpy as np +GRAPH_CACHE_KEY = "warehouse_plan:graph:{}" + class WarehousePlan(Document): # begin: auto-generated types @@ -28,6 +31,28 @@ class WarehousePlan(Document): vertical: DF.Float # end: auto-generated types + @property + def graph(self) -> "Grid_TSP | None": + g = frappe.cache.get_value(GRAPH_CACHE_KEY.format(self.name)) + if g is None and self.matrix: + g = self._build_graph() + frappe.cache.set_value(GRAPH_CACHE_KEY.format(self.name), g) + return g + + def _build_graph(self) -> "Grid_TSP": + grid = np.array(safe_json_loads(self.matrix)) + return Grid_TSP(grid, scale=self.horizontal / grid.shape[1]) + + def on_load(self): + if self.matrix and not frappe.cache.get_value(GRAPH_CACHE_KEY.format(self.name)): + frappe.cache.set_value(GRAPH_CACHE_KEY.format(self.name), self._build_graph()) + + def on_save(self): + if self.matrix: + frappe.cache.set_value(GRAPH_CACHE_KEY.format(self.name), self._build_graph()) + else: + frappe.cache.delete_value(GRAPH_CACHE_KEY.format(self.name)) + @frappe.whitelist() def get_plan_warehouses(self): return frappe.get_all( @@ -38,34 +63,35 @@ def get_plan_warehouses(self): @frappe.whitelist() def set_warehouse_plan_details(self, warehouses: list): - existing_warehouses = frappe.get_all( - "Warehouse", - filters={"warehouse_plan": self.name}, - pluck="name", + Wh = frappe.qb.DocType("Warehouse") + existing_warehouses = ( + frappe.qb.from_(Wh).select(Wh.name).where(Wh.warehouse_plan == self.name).run(pluck=True) ) for warehouse in warehouses: - warehouse_doc = frappe.get_doc("Warehouse", warehouse.get("warehouse_name")) - warehouse_doc.update( - { - "warehouse_plan": self.name, - "warehouse_plan_coordinates": warehouse.get("coordinates"), - "rotation": warehouse.get("rotation"), - "accessible_path": warehouse.get("accessible_path"), - } + wh_name = warehouse.get("warehouse_name") + ( + frappe.qb.update(Wh) + .set(Wh.warehouse_plan, self.name) + .set(Wh.warehouse_plan_coordinates, warehouse.get("coordinates")) + .set(Wh.rotation, warehouse.get("rotation")) + .set(Wh.accessible_path, warehouse.get("accessible_path")) + .where(Wh.name == wh_name) + .run() + ) + if wh_name in existing_warehouses: + existing_warehouses.remove(wh_name) + + if existing_warehouses: + ( + frappe.qb.update(Wh) + .set(Wh.warehouse_plan, None) + .set(Wh.warehouse_plan_coordinates, None) + .set(Wh.rotation, 0) + .set(Wh.accessible_path, None) + .where(Wh.name.isin(existing_warehouses)) + .run() ) - warehouse_doc.save() - - if warehouse_doc.name in existing_warehouses: - existing_warehouses.remove(warehouse_doc.name) - - # if warehouses are deleted, remove them from the warehouse plan - if len(existing_warehouses) > 0: - for warehouse in existing_warehouses: - frappe.db.set_value("Warehouse", warehouse, "warehouse_plan", None) - frappe.db.set_value("Warehouse", warehouse, "warehouse_plan_coordinates", None) - frappe.db.set_value("Warehouse", warehouse, "rotation", 0) - frappe.db.set_value("Warehouse", warehouse, "accessible_path", None) @frappe.whitelist() def get_warehouse_dimensions(self, warehouse: str): diff --git a/inventory_tools/inventory_tools/overrides/pick_list.py b/inventory_tools/inventory_tools/overrides/pick_list.py index c916053..5e5572e 100644 --- a/inventory_tools/inventory_tools/overrides/pick_list.py +++ b/inventory_tools/inventory_tools/overrides/pick_list.py @@ -2,39 +2,62 @@ # For license information, please see license.txt import frappe -from frappe.utils import safe_json_loads +from erpnext.stock.doctype.pick_list.pick_list import PickList as ERPNextPickList from frappe.utils.data import nowdate -import numpy as np -from inventory_tools.inventory_tools.doctype.warehouse_plan.warehouse_plan import Grid_TSP + +class InventoryToolsPickList(ERPNextPickList): + def _get_default_strategy(self) -> str | None: + if not self.company: + return None + return frappe.db.get_value( + "Inventory Tools Settings", + {"company": self.company}, + "default_route_optimization_strategy", + ) + + def after_mapping(self, source_doc): # noqa + strategy = self._get_default_strategy() + self.set_onload("default_route_optimization_strategy", strategy or "Use Source Document Order") + if not strategy or strategy == "Use Source Document Order" or not self.get("locations"): + return + try: + result = optimize_path(self.as_dict(), strategy) + self.set("locations", []) + for item in result: + self.append("locations", item) + except Exception: + pass + + def onload(self): + strategy = self._get_default_strategy() + self.set_onload("default_route_optimization_strategy", strategy or "Use Source Document Order") class PathFinder: @staticmethod - def _process_entries(item_code, qty, company, order_by, root_warehouse, to_date): - # Retrieve stock ledger entries with the provided filters and ordering. - sle = frappe.get_all( - "Stock Ledger Entry", - fields=["actual_qty", "posting_date", "creation", "warehouse"], - filters={ - "item_code": item_code, - "company": company, - "posting_date": ["<=", to_date], - "is_cancelled": 0, - "actual_qty": [">", 0], - }, - order_by=order_by, + def _process_entries(item_code, qty, company, order_by_clauses, plan_warehouses, to_date): + SLE = frappe.qb.DocType("Stock Ledger Entry") + query = ( + frappe.qb.from_(SLE) + .select(SLE.actual_qty, SLE.posting_date, SLE.creation, SLE.warehouse) + .where(SLE.item_code == item_code) + .where(SLE.company == company) + .where(SLE.posting_date <= to_date) + .where(SLE.is_cancelled == 0) + .where(SLE.actual_qty > 0) ) + if plan_warehouses: + query = query.where(SLE.warehouse.isin(list(plan_warehouses))) + for field, order in order_by_clauses: + query = query.orderby(field, order=order) + + sle = query.run(as_dict=True) newsle = [] qty_obtained = 0 - # Process each entry until the required quantity is fulfilled. for entry in sle: - # If a root warehouse is specified, ensure the entry belongs to it. - if root_warehouse and get_root_warehouse(entry["warehouse"]) != root_warehouse: - continue - remaining_qty = qty - qty_obtained if entry["actual_qty"] >= remaining_qty: @@ -47,57 +70,64 @@ def _process_entries(item_code, qty, company, order_by, root_warehouse, to_date) ) qty_obtained += entry["actual_qty"] - # If the accumulated quantity doesn't match the requested quantity, raise an error. if (qty - qty_obtained) != 0: raise frappe.ValidationError("Not enough items in root warehouse") return newsle @staticmethod - def FIFO(item_code, qty, company, root_warehouse=None, to_date=None): - # FIFO: Order by posting_date and creation in ascending order. - if to_date is None: - to_date = nowdate() + def FIFO(item_code, qty, company, plan_warehouses=None, to_date=None): + SLE = frappe.qb.DocType("Stock Ledger Entry") + order_by = [(SLE.posting_date, frappe.qb.asc), (SLE.creation, frappe.qb.asc)] return PathFinder._process_entries( - item_code, qty, company, "posting_date, creation", root_warehouse, to_date + item_code, qty, company, order_by, plan_warehouses, to_date or nowdate() ) @staticmethod - def LIFO(item_code, qty, company, root_warehouse=None, to_date=None): - # LIFO: Order by posting_date and creation in descending order. - if to_date is None: - to_date = nowdate() + def LIFO(item_code, qty, company, plan_warehouses=None, to_date=None): + SLE = frappe.qb.DocType("Stock Ledger Entry") + order_by = [(SLE.posting_date, frappe.qb.desc), (SLE.creation, frappe.qb.desc)] return PathFinder._process_entries( - item_code, qty, company, "posting_date desc, creation desc", root_warehouse, to_date + item_code, qty, company, order_by, plan_warehouses, to_date or nowdate() ) @staticmethod - def deplete_max_bins(item_code, qty, company, root_warehouse=None, to_date=None): - if to_date is None: - to_date = nowdate() + def deplete_max_bins(item_code, qty, company, plan_warehouses=None, to_date=None): + SLE = frappe.qb.DocType("Stock Ledger Entry") + order_by = [ + (SLE.actual_qty, frappe.qb.asc), + (SLE.posting_date, frappe.qb.asc), + (SLE.creation, frappe.qb.asc), + ] return PathFinder._process_entries( - item_code, qty, company, "actual_qty, posting_date, creation", root_warehouse, to_date + item_code, qty, company, order_by, plan_warehouses, to_date or nowdate() ) @staticmethod - def deplete_min_bins(item_code, qty, company, root_warehouse=None, to_date=None): - if to_date is None: - to_date = nowdate() + def deplete_min_bins(item_code, qty, company, plan_warehouses=None, to_date=None): + SLE = frappe.qb.DocType("Stock Ledger Entry") + order_by = [ + (SLE.actual_qty, frappe.qb.desc), + (SLE.posting_date, frappe.qb.asc), + (SLE.creation, frappe.qb.asc), + ] return PathFinder._process_entries( - item_code, qty, company, "actual_qty desc, posting_date, creation", root_warehouse, to_date + item_code, qty, company, order_by, plan_warehouses, to_date or nowdate() ) def get_root_warehouse(warehouse): - # Finds closest parent warehouse with a walkable floor plan; otherwise returns None - wh_plans = [wh["name"] for wh in frappe.get_all("Warehouse Plan")] - if warehouse in wh_plans: - wp_doc = frappe.get_doc("Warehouse Plan", warehouse) - if wp_doc.as_dict()["matrix"] is not None: - return warehouse - parent_warehouse = frappe.get_doc("Warehouse", warehouse).as_dict()["parent_warehouse"] - if parent_warehouse == "": - raise frappe.ValidationError("Warehouse does not have a parent warehouse") - return get_root_warehouse(parent_warehouse) + WP = frappe.qb.DocType("Warehouse Plan") + has_matrix = ( + frappe.qb.from_(WP).select(WP.name).where((WP.name == warehouse) & (WP.matrix.isnotnull())).run() + ) + if has_matrix: + return warehouse + + Wh = frappe.qb.DocType("Warehouse") + result = frappe.qb.from_(Wh).select(Wh.warehouse_plan).where(Wh.name == warehouse).run() + if result and result[0][0]: + return result[0][0] + raise frappe.ValidationError(f"Warehouse '{warehouse}' is not part of any Warehouse Plan") @frappe.whitelist() @@ -119,27 +149,22 @@ def optimize_route_picklist(item_whs: list, root_warehouse: str) -> list: Returns: list: A reordered list of dictionaries, optimized for the pick-up route. """ - - grid = np.array( - safe_json_loads(frappe.get_doc("Warehouse Plan", root_warehouse).as_dict()["matrix"]) - ) - - imaginary_x = grid.shape[1] - real_x = frappe.get_doc("Warehouse Plan", root_warehouse).as_dict()["horizontal"] - scale = real_x / imaginary_x - - g = Grid_TSP(grid, scale=scale) - - root_wh = frappe.get_doc("Warehouse Plan", root_warehouse).as_dict() - dropoff = [g.pos2node((root_wh["pickup_point_x"], root_wh["pickup_point_y"]))] + wp = frappe.get_cached_doc("Warehouse Plan", root_warehouse) + g = wp.graph + dropoff = [g.pos2node((wp.pickup_point_x, wp.pickup_point_y))] unique_whs = list({item_wh["warehouse"] for item_wh in item_whs}) - warehouse_to_node = {} - for wh in unique_whs: - accessible_path = frappe.get_doc("Warehouse", wh).as_dict()["accessible_path"].split(",") - coordinate = (int(accessible_path[0]), int(accessible_path[1])) - warehouse_to_node[wh] = g.pos2node(coordinate) + Wh = frappe.qb.DocType("Warehouse") + wh_paths = ( + frappe.qb.from_(Wh) + .select(Wh.name, Wh.accessible_path) + .where(Wh.name.isin(unique_whs)) + .run(as_dict=True) + ) + warehouse_to_node = { + row.name: g.pos2node(tuple(int(x) for x in row.accessible_path.split(","))) for row in wh_paths + } node_to_warehouse = {node: wh for wh, node in warehouse_to_node.items()} pickup_list = list(warehouse_to_node.values()) @@ -187,6 +212,7 @@ def optimize_path(doc: str | dict, strategy: str) -> list[dict]: doc_dict: dict = frappe.get_doc("Pick List", doc).as_dict() else: doc_dict = doc + itemdict: dict[str, dict[str, float]] = {} for loc in doc_dict["locations"]: code = loc["item_code"] @@ -195,27 +221,40 @@ def optimize_path(doc: str | dict, strategy: str) -> list[dict]: itemdict[code]["qty"] += qty else: itemdict[code] = {"qty": qty} + company = doc_dict["company"] - root_warehouses = [get_root_warehouse(loc["warehouse"]) for loc in doc_dict["locations"]] + + unique_locations = {loc["warehouse"] for loc in doc_dict["locations"]} + root_wh_map = {wh: get_root_warehouse(wh) for wh in unique_locations} + root_warehouses = [root_wh_map[loc["warehouse"]] for loc in doc_dict["locations"]] if not all(wh == root_warehouses[0] for wh in root_warehouses): raise frappe.ValidationError("All items in pick list do not share a common warehouse plan") root_warehouse = root_warehouses[0] + Wh = frappe.qb.DocType("Warehouse") + plan_warehouses = frozenset( + frappe.qb.from_(Wh).select(Wh.name).where(Wh.warehouse_plan == root_warehouse).run(pluck=True) + ) + item_whs = [] for item in itemdict.keys(): if strategy == "FIFO": - item_whs += PathFinder.FIFO(item, itemdict[item]["qty"], company, root_warehouse=root_warehouse) + item_whs += PathFinder.FIFO( + item, itemdict[item]["qty"], company, plan_warehouses=plan_warehouses + ) elif strategy == "LIFO": - item_whs += PathFinder.LIFO(item, itemdict[item]["qty"], company, root_warehouse=root_warehouse) + item_whs += PathFinder.LIFO( + item, itemdict[item]["qty"], company, plan_warehouses=plan_warehouses + ) elif strategy == "Deplete maximum number of Bins": item_whs += PathFinder.deplete_max_bins( - item, itemdict[item]["qty"], company, root_warehouse=root_warehouse + item, itemdict[item]["qty"], company, plan_warehouses=plan_warehouses ) elif strategy == "Deplete minimum number of Bins": item_whs += PathFinder.deplete_min_bins( - item, itemdict[item]["qty"], company, root_warehouse=root_warehouse + item, itemdict[item]["qty"], company, plan_warehouses=plan_warehouses ) try: return optimize_route_picklist(item_whs, root_warehouse) diff --git a/inventory_tools/inventory_tools/overrides/workstation.py b/inventory_tools/inventory_tools/overrides/workstation.py index e64b585..9125529 100644 --- a/inventory_tools/inventory_tools/overrides/workstation.py +++ b/inventory_tools/inventory_tools/overrides/workstation.py @@ -12,7 +12,7 @@ class InventoryToolsWorkstation(Workstation): def validate_working_hours(self, row): """ - HASH: 3b4d39766f78492bd2ba92dc6c6c5b91263d3e6d + HASH: 9771ed4c572510ec51586606f9d57ab6459717f1 REPO: https://github.com/frappe/erpnext/ PATH: erpnext/manufacturing/doctype/workstation/workstation.py METHOD: validate_working_hours @@ -29,7 +29,7 @@ def validate_working_hours(self, row): def set_total_working_hours(self): """ - HASH: 3b4d39766f78492bd2ba92dc6c6c5b91263d3e6d + HASH: 9771ed4c572510ec51586606f9d57ab6459717f1 REPO: https://github.com/frappe/erpnext/ PATH: erpnext/manufacturing/doctype/workstation/workstation.py METHOD: set_total_working_hours @@ -51,7 +51,7 @@ def set_total_working_hours(self): def validate_overlap_for_operation_timings(self): """ - HASH: 3b4d39766f78492bd2ba92dc6c6c5b91263d3e6d + HASH: 9771ed4c572510ec51586606f9d57ab6459717f1 REPO: https://github.com/frappe/erpnext/ PATH: erpnext/manufacturing/doctype/workstation/workstation.py METHOD: validate_overlap_for_operation_timings @@ -75,7 +75,7 @@ def validate_overlap_for_operation_timings(self): def set_hour_rate(self): """ - HASH: 3b4d39766f78492bd2ba92dc6c6c5b91263d3e6d + HASH: 9771ed4c572510ec51586606f9d57ab6459717f1 REPO: https://github.com/frappe/erpnext/ PATH: erpnext/manufacturing/doctype/workstation/workstation.py METHOD: set_hour_rate diff --git a/inventory_tools/public/js/custom/pick_list_custom.js b/inventory_tools/public/js/custom/pick_list_custom.js index 5f459fc..c1d807d 100644 --- a/inventory_tools/public/js/custom/pick_list_custom.js +++ b/inventory_tools/public/js/custom/pick_list_custom.js @@ -1,7 +1,13 @@ // Copyright (c) 2025, AgriTheory and contributors // For license information, please see license.txt +const STRATEGY_OPTIONS = ['FIFO', 'LIFO', 'Deplete maximum number of Bins', 'Deplete minimum number of Bins'] + frappe.ui.form.on('Pick List', { + onload: frm => { + const setting = frm.doc.__onload && frm.doc.__onload.default_route_optimization_strategy + frm._default_route_strategy = setting && setting !== 'Use Source Document Order' ? setting : STRATEGY_OPTIONS[0] + }, refresh: frm => { add_path_button(frm) }, @@ -21,9 +27,9 @@ function path_dialog(frm) { label: __('Strategy'), fieldname: 'strategy', fieldtype: 'Select', - options: ['FIFO', 'LIFO', 'Deplete maximum number of Bins', 'Deplete minimum number of Bins'], + options: STRATEGY_OPTIONS, reqd: 1, - default: 'Deplete maximum number of Bins', + default: frm._default_route_strategy || STRATEGY_OPTIONS[0], }, ], primary_action: async () => { diff --git a/inventory_tools/tests/test_warehouse_plan.py b/inventory_tools/tests/test_warehouse_plan.py index 04e9b1f..9153f06 100644 --- a/inventory_tools/tests/test_warehouse_plan.py +++ b/inventory_tools/tests/test_warehouse_plan.py @@ -14,6 +14,12 @@ optimize_path, ) +STRATEGY_WAREHOUSES = [ + {"item_code": "Cranberry", "qty": 2, "warehouse": "Fruit Storage 50 - CFC"}, + {"item_code": "Banana", "qty": 1, "warehouse": "Fruit Storage 1 - CFC"}, + {"item_code": "Coconut", "qty": 1, "warehouse": "Fruit Storage 10 - CFC"}, +] + # --- Grid_TSP Tests --- @@ -333,3 +339,89 @@ def test_optimize_path_deplete_min_bins(): assert len(result) > 0 total_qty = sum(r["qty"] for r in result) assert total_qty == 5 + + +# --- InventoryToolsPickList.after_mapping Tests --- + + +@pytest.mark.order(75) +def test_after_mapping_optimizes_with_default_strategy(): + """after_mapping reorders locations using the company's default strategy.""" + settings = frappe.get_doc("Inventory Tools Settings", "Chelsea Fruit Co") + original_strategy = settings.default_route_optimization_strategy + settings.default_route_optimization_strategy = "FIFO" + settings.save() + + try: + pl = frappe.new_doc("Pick List") + pl.company = "Chelsea Fruit Co" + for loc in STRATEGY_WAREHOUSES: + pl.append("locations", loc) + + pl.after_mapping(None) + + expected = optimize_path( + {"company": "Chelsea Fruit Co", "locations": STRATEGY_WAREHOUSES}, + "FIFO", + ) + assert [loc.warehouse for loc in pl.locations] == [r["warehouse"] for r in expected] + finally: + settings.default_route_optimization_strategy = original_strategy + settings.save() + + +@pytest.mark.order(76) +def test_after_mapping_is_noop_with_source_document_order(): + """after_mapping leaves locations unchanged when strategy is Use Source Document Order.""" + settings = frappe.get_doc("Inventory Tools Settings", "Chelsea Fruit Co") + original_strategy = settings.default_route_optimization_strategy + settings.default_route_optimization_strategy = "Use Source Document Order" + settings.save() + + try: + pl = frappe.new_doc("Pick List") + pl.company = "Chelsea Fruit Co" + for loc in STRATEGY_WAREHOUSES: + pl.append("locations", loc) + + original_order = [loc["warehouse"] for loc in STRATEGY_WAREHOUSES] + pl.after_mapping(None) + + assert [loc.warehouse for loc in pl.locations] == original_order + finally: + settings.default_route_optimization_strategy = original_strategy + settings.save() + + +@pytest.mark.order(77) +def test_after_mapping_populates_onload_data(): + """after_mapping sets __onload.default_route_optimization_strategy for the frontend dialog.""" + settings = frappe.get_doc("Inventory Tools Settings", "Chelsea Fruit Co") + original_strategy = settings.default_route_optimization_strategy + settings.default_route_optimization_strategy = "LIFO" + settings.save() + + try: + pl = frappe.new_doc("Pick List") + pl.company = "Chelsea Fruit Co" + pl.append("locations", STRATEGY_WAREHOUSES[0]) + + pl.after_mapping(None) + + assert pl.get_onload("default_route_optimization_strategy") == "LIFO" + finally: + settings.default_route_optimization_strategy = original_strategy + settings.save() + + +@pytest.mark.order(78) +def test_after_mapping_is_noop_without_company(): + """after_mapping does nothing when company is not set on the document.""" + pl = frappe.new_doc("Pick List") + for loc in STRATEGY_WAREHOUSES: + pl.append("locations", loc) + + original_order = [loc["warehouse"] for loc in STRATEGY_WAREHOUSES] + pl.after_mapping(None) + + assert [loc.warehouse for loc in pl.locations] == original_order