[WIP] shopfloor: zone_picking, show if location are about to be empty

shopfloor/models/stock_location.py

@@ -45,25 +45,36 @@ def _compute_reserved_move_lines(self):
         for rec in self:
             rec.update({"reserved_move_line_ids": rec._get_reserved_move_lines()})
 
-    def planned_qty_in_location_is_empty(self):
+    def planned_qty_in_location_is_empty(self, move_lines=None):
         """Return if a location will be empty when move lines will be confirmed
 
         Used for the "zero check". We need to know if a location is empty, but since
         we set the move lines to "done" only at the end of the unload workflow, we
         have to look at the qty_done of the move lines from this location.
+
+        With `move_lines` we can force the use of the given move lines for the check.
+        This allows to know that the location will be empty if we process only
+        these move lines.
         """
         self.ensure_one()
         quants = self.env["stock.quant"].search(
             [("quantity", ">", 0), ("location_id", "=", self.id)]
         )
         remaining = sum(quants.mapped("quantity"))
-        lines = self.env["stock.move.line"].search(
-            [
-                ("state", "!=", "done"),
-                ("location_id", "=", self.id),
-                ("qty_done", ">", 0),
-            ]
-        )
-        planned = remaining - sum(lines.mapped("qty_done"))
+        move_line_qty_field = "qty_done"
+        if move_lines:
+            move_lines = move_lines.filtered(
+                lambda m: m.state not in ("cancel", "done")
+            )
+            move_line_qty_field = "product_uom_qty"
+        else:
+            move_lines = self.env["stock.move.line"].search(
+                [
+                    ("state", "not in", ("cancel", "done")),
+                    ("location_id", "=", self.id),
+                    ("qty_done", ">", 0),
+                ]
+            )
+        planned = remaining - sum(move_lines.mapped(move_line_qty_field))
         compare = float_compare(planned, 0, precision_rounding=0.01)
         return compare <= 0

shopfloor/services/zone_picking.py

@@ -220,11 +220,17 @@ def _data_for_move_line(self, zone_location, picking_type, move_line):
         }
 
     def _data_for_move_lines(self, zone_location, picking_type, move_lines):
-        return {
+        data = {
             "zone_location": self.data.location(zone_location),
             "picking_type": self.data.picking_type(picking_type),
             "move_lines": self.data.move_lines(move_lines, with_picking=True),
         }
+        for data_move_line in data["move_lines"]:
+            move_line = self.env["stock.move.line"].browse(data_move_line["id"])
+            data_move_line[
+                "empty_location_src"
+            ] = move_line.location_id.planned_qty_in_location_is_empty(move_line)
+        return data
 
     def _data_for_location(self, zone_location, picking_type, location):
         return {
@@ -1428,7 +1434,7 @@ def _states(self):
         return {
             "start": {},
             "select_picking_type": self._schema_for_select_picking_type,
-            "select_line": self._schema_for_move_lines,
+            "select_line": self._schema_for_move_lines_empty_location,
             "set_line_destination": self._schema_for_move_line,
             "zero_check": self._schema_for_zero_check,
             "change_pack_lot": self._schema_for_move_line,
@@ -1548,6 +1554,16 @@ def _schema_for_move_lines(self):
         }
         return schema
 
+    @property
+    def _schema_for_move_lines_empty_location(self):
+        schema = self._schema_for_move_lines
+        schema["move_lines"]["schema"]["schema"]["empty_location_src"] = {
+            "type": "boolean",
+            "nullable": False,
+            "required": True,
+        }
+        return schema
+
     @property
     def _schema_for_zero_check(self):
         schema = {

shopfloor/tests/test_zone_picking_base.py

@@ -216,16 +216,18 @@ def _assert_response_select_line(
         message=None,
         popup=None,
     ):
+        data = {
+            "zone_location": self.data.location(zone_location),
+            "picking_type": self.data.picking_type(picking_type),
+            "move_lines": self.data.move_lines(move_lines, with_picking=True),
+        }
+        for data_move_line in data["move_lines"]:
+            move_line = self.env["stock.move.line"].browse(data_move_line["id"])
+            data_move_line[
+                "empty_location_src"
+            ] = move_line.location_id.planned_qty_in_location_is_empty(move_line)
         self.assert_response(
-            response,
-            next_state=state,
-            data={
-                "zone_location": self.data.location(zone_location),
-                "picking_type": self.data.picking_type(picking_type),
-                "move_lines": self.data.move_lines(move_lines, with_picking=True),
-            },
-            message=message,
-            popup=popup,
+            response, next_state=state, data=data, message=message, popup=popup,
         )
 
     def assert_response_select_line(

shopfloor/tests/test_zone_picking_select_line.py

@@ -396,3 +396,42 @@ def test_prepare_unload_buffer_multi_line_same_destination(self):
         self.assert_response_unload_all(
             response, zone_location, picking_type, self.picking5.move_line_ids,
         )
+
+    def test_list_move_lines_empty_location(self):
+        zone_location = self.zone_location
+        picking_type = self.picking1.picking_type_id
+        response = self.service.dispatch(
+            "list_move_lines",
+            params={
+                "zone_location_id": zone_location.id,
+                "picking_type_id": picking_type.id,
+                "order": "location",
+            },
+        )
+        move_lines = self.service._find_location_move_lines(
+            zone_location, picking_type, order="location"
+        )
+        self.assert_response_select_line(
+            response, zone_location, picking_type, move_lines,
+        )
+        data_move_lines = response["data"]["select_line"]["move_lines"]
+        # Check that the move line in "Zone sub-location 1" is about to empty
+        # its location if we process it
+        data_move_line = [
+            m
+            for m in data_move_lines
+            if m["location_src"]["barcode"] == "ZONE_SUBLOCATION_1"
+        ][0]
+        self.assertTrue(data_move_line["empty_location_src"])
+        # Same check with the internal method
+        move_line = self.env["stock.move.line"].browse(data_move_line["id"])
+        location_src = move_line.location_id
+        move_line_will_empty_location = location_src.planned_qty_in_location_is_empty(
+            move_lines=move_line
+        )
+        self.assertTrue(move_line_will_empty_location)
+        # But if we check the location without giving the move line as parameter,
+        # knowing that this move line hasn't its 'qty_done' field filled,
+        # the location won't be considered empty with such pending move line
+        move_line_will_empty_location = location_src.planned_qty_in_location_is_empty()
+        self.assertFalse(move_line_will_empty_location)