From a329d127b7f02dbbd23f07e4c61c7d47bbe4f82a Mon Sep 17 00:00:00 2001 From: Denis Roussel Date: Tue, 2 Jun 2026 10:23:27 +0200 Subject: [PATCH] [IMP] shopfloor: Be able to manage several source locations When there are several move lines for the same product but with several source locations, location_content_transfer was confused about which move line to return to the operator as there was a fallback on the picking. Use directly the move line concerned by the scanned source location. --- .../location_content_transfer_sorter.py | 12 ++- .../services/location_content_transfer.py | 84 ++++++++++++++----- ...location_content_transfer_scan_location.py | 79 +++++++++++++++++ 3 files changed, 152 insertions(+), 23 deletions(-) diff --git a/shopfloor/actions/location_content_transfer_sorter.py b/shopfloor/actions/location_content_transfer_sorter.py index 7d9a5d85816..48cbedcd28e 100644 --- a/shopfloor/actions/location_content_transfer_sorter.py +++ b/shopfloor/actions/location_content_transfer_sorter.py @@ -11,11 +11,19 @@ class LocationContentTransferSorter(Component): def __init__(self, work_context): super().__init__(work_context) self._pickings = self.env["stock.picking"].browse() + self._lines = self.env["stock.move.line"].browse() self._content = None def feed_pickings(self, pickings): self._pickings |= pickings + def feed_lines(self, move_lines): + """ + TODO: Remove pickings to use move lines directly instead + """ + self._lines |= move_lines + self._pickings |= move_lines.picking_id + def move_lines(self): """Returns valid move lines. @@ -26,8 +34,10 @@ def move_lines(self): An invalid package level has one of its line not targetting the expected package. """ + # TODO: Remove this when using only move lines + lines = self._lines if self._lines else self._pickings.move_line_ids # lines without package level only (raw products) - move_lines = self._pickings.move_line_ids.filtered( + move_lines = lines.filtered( lambda line: not line.package_level_id and line.state not in ("cancel", "done") ) diff --git a/shopfloor/services/location_content_transfer.py b/shopfloor/services/location_content_transfer.py index e68bdcc8b65..0ca20a24905 100644 --- a/shopfloor/services/location_content_transfer.py +++ b/shopfloor/services/location_content_transfer.py @@ -99,6 +99,25 @@ def _response_for_scan_destination_all( next_state="scan_destination_all", data=data, message=message ) + def _response_for_scan_destination_lines_all( + self, move_lines, message=None, confirmation_required=None + ): + """Transition to the 'scan_destination_all' state + + The client screen shows a summary of all the lines and packages + to move to a single destination. + + If `confirmation_required` is set, + the client will ask to scan again the destination + """ + data = self._data_content_line_all_for_location(move_lines=move_lines) + data["confirmation_required"] = confirmation_required + if confirmation_required and not message: + message = self.msg_store.need_confirmation() + return self._response( + next_state="scan_destination_all", data=data, message=message + ) + def _response_for_start_single(self, pickings, message=None, popup=None): """Transition to the 'start_single' state @@ -130,6 +149,9 @@ def _response_for_scan_destination( return self._response(next_state="scan_destination", data=data, message=message) def _data_content_all_for_location(self, pickings): + """ + TODO: Remove this preferring using move lines instead + """ sorter = self._actions_for("location_content_transfer.sorter") sorter.feed_pickings(pickings) lines = sorter.move_lines() @@ -142,6 +164,19 @@ def _data_content_all_for_location(self, pickings): "package_levels": self.data.package_levels(package_levels), } + def _data_content_line_all_for_location(self, move_lines): + sorter = self._actions_for("location_content_transfer.sorter") + sorter.feed_lines(move_lines) + lines = sorter.move_lines() + package_levels = sorter.package_levels() + location = move_lines.location_id + assert len(location) == 1, "There should be only one src location at this stage" + return { + "location": self.data.location(location), + "move_lines": self.data.move_lines(lines), + "package_levels": self.data.package_levels(package_levels), + } + def _data_content_line_for_location(self, location, next_content): assert next_content._name in ("stock.move.line", "stock.package_level") line_data = ( @@ -166,37 +201,42 @@ def _next_content(self, pickings): return None return next_content - def _router_single_or_all_destination(self, pickings, message=None): - location_dest = pickings.mapped("move_line_ids.location_dest_id") - location_src = pickings.mapped("move_line_ids.location_id") + def _router_single_or_all_destination(self, move_lines, message=None): + location_dest = move_lines.location_dest_id + location_src = move_lines.location_id if len(location_dest) == len(location_src) == 1: - return self._response_for_scan_destination_all(pickings, message=message) + return self._response_for_scan_destination_lines_all( + move_lines, message=message + ) else: - return self._response_for_start_single(pickings, message=message) + return self._response_for_start_single( + move_lines.picking_id, message=message + ) - def _domain_recover_pickings(self): + def _domain_recover_move_lines(self): return [ - ("user_id", "=", self.env.uid), - ("state", "=", "assigned"), + ("picking_id.user_id", "=", self.env.uid), + ("picking_id.state", "=", "assigned"), ("picking_type_id", "in", self.picking_types.ids), + ("qty_done", ">", 0), ] - def _search_recover_pickings(self): - candidate_pickings = self.env["stock.picking"].search( - self._domain_recover_pickings() - ) - started_pickings = candidate_pickings.filtered( - lambda picking: any(line.qty_done for line in picking.move_line_ids) + def _search_recover_move_lines(self): + """ + Get the move lines that have already been marked as done + """ + started_move_lines = self.env["stock.move.line"].search( + self._domain_recover_move_lines() ) - return started_pickings + return started_move_lines - def _recover_started_picking(self): + def _recover_started_move_lines(self): """Get the next response if the user has work in progress.""" - started_pickings = self._search_recover_pickings() - if not started_pickings: + started_move_lines = self._search_recover_move_lines() + if not started_move_lines: return False return self._router_single_or_all_destination( - started_pickings, message=self.msg_store.recovered_previous_session() + started_move_lines, message=self.msg_store.recovered_previous_session() ) def start_or_recover(self): @@ -206,7 +246,7 @@ def start_or_recover(self): and reopen the menu, we want to directly reopen the screens to choose destinations. Otherwise, we go to the "start" state. """ - response = self._recover_started_picking() + response = self._recover_started_move_lines() return response or self._response_for_start() def _create_moves_from_location(self, location): @@ -253,7 +293,7 @@ def find_work(self): * start: no work found * scan_location: with the location to work form for confirmation """ - response = self._recover_started_picking() + response = self._recover_started_move_lines() if response: return response @@ -410,7 +450,7 @@ def scan_location(self, barcode): # noqa: C901 savepoint.release() - return self._router_single_or_all_destination(move_lines.picking_id) + return self._router_single_or_all_destination(move_lines) def _find_transfer_move_lines_domain(self, location): return [ diff --git a/shopfloor/tests/test_location_content_transfer_scan_location.py b/shopfloor/tests/test_location_content_transfer_scan_location.py index c10171c7588..a59301260d5 100644 --- a/shopfloor/tests/test_location_content_transfer_scan_location.py +++ b/shopfloor/tests/test_location_content_transfer_scan_location.py @@ -32,3 +32,82 @@ def test_lines_returned_by_scan_location(self): lines = response["data"]["scan_destination_all"]["move_lines"] line_ids = [line["id"] for line in lines] self.assertTrue(self.move1.move_line_ids.id not in line_ids) + + +class TestLocationContentTransferScanLocationSameProduct( + LocationContentTransferCommonCase +): + @classmethod + def setUpClassBaseData(cls): + super().setUpClassBaseData() + # If the product is available in several sub locations of the picking + # location (a view) and the scanned location is one of those children, + cls.parent = ( + cls.env["stock.location"] + .sudo() + .create( + { + "name": "Transfer", + "location_id": cls.wh.view_location_id.id, + } + ) + ) + cls.child_1 = ( + cls.env["stock.location"] + .sudo() + .create( + { + "name": "Child 1", + "location_id": cls.parent.id, + "barcode": "L#CHILD01", + } + ) + ) + cls.child_2 = ( + cls.env["stock.location"] + .sudo() + .create( + { + "name": "Child 2", + "location_id": cls.parent.id, + "barcode": "L#CHILD02", + } + ) + ) + cls.p_type = ( + cls.env["stock.picking.type"] + .sudo() + .create( + { + "name": "Transfer Test", + "sequence_code": "TRANS-TEST", + "default_location_dest_id": cls.wh.lot_stock_id.id, + "default_location_src_id": cls.parent.id, + } + ) + ) + cls.menu.sudo().picking_type_ids = cls.p_type + + cls._update_qty_in_location(cls.child_1, cls.product_a, 10.0) + cls._update_qty_in_location(cls.child_2, cls.product_a, 10.0) + cls.picking1 = cls._create_picking( + lines=[(cls.product_a, 12)], picking_type=cls.p_type + ) + cls.picking1.move_type = "one" + cls.picking1.action_assign() + + # During the mean time, the location has been filled in + cls._update_qty_in_location(cls.child_2, cls.product_a, 300.0) + + cls.picking1.move_line_ids[1].reserved_uom_qty = 300.0 + + def test_lines_returned_by_scan_location(self): + """Check that lines from not ready pickings are not offered to work on.""" + self.picking1.move_line_ids[1].location_dest_id = self.shelf1 + + response = self.service.dispatch( + "scan_location", params={"barcode": self.child_2.barcode} + ) + lines = response["data"]["scan_destination_all"]["move_lines"] + line_ids = [line["id"] for line in lines] + self.assertTrue(self.picking1.move_line_ids[0].id not in line_ids)