[IMP] shopfloor: Be able to manage several source locations

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")
         )

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 [

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)