[16.0][IMP] stock_storage_type: Allow to recompute putaways + several improvements

stock_storage_type/__manifest__.py

@@ -14,6 +14,7 @@
     "application": False,
     "installable": True,
     "depends": [
+        "stock_location_fill_state",
         "stock_move_line_reserved_quant",
         "stock_putaway_hook",
         "stock_quant_package_dimension",

stock_storage_type/models/stock_location.py

@@ -7,7 +7,7 @@
 
 from odoo import api, fields, models
 from odoo.fields import Command
-from odoo.tools import float_compare, index_exists
+from odoo.tools import float_compare, groupby, index_exists
 
 _logger = logging.getLogger(__name__)
 OUT_MOVE_LINE_DOMAIN = [
@@ -58,15 +58,7 @@
         "location_id",
         string="Storage locations sequences",
     )
-    location_is_empty = fields.Boolean(
-        compute="_compute_location_is_empty",
-        store=True,
-        help="technical field: True if the location is empty "
-        "and there is no pending incoming products in the location. "
-        " Computed only if the location needs to check for emptiness "
-        '(has an "only empty" policy).',
-        recursive=True,
-    )
+
     # TODO: Maybe renaming these fields as there are already such fields
     # in core but without domains. Something like 'pending_in_move_ids'
     in_move_ids = fields.One2many(
@@ -142,6 +134,23 @@
         compute="_compute_only_empty", store=True, recursive=True
     )
 
+    has_potential_product_mix_exception = fields.Boolean(
+        compute="_compute_has_potential_product_mix_exception",
+        store=True,
+        index=True,
+        help="This will represent a situation where several moves are pointing"
+        "to the location for different products and the location does"
+        "not allow mixed products.",
+    )
+    has_potential_lot_mix_exception = fields.Boolean(
+        compute="_compute_has_potential_lot_mix_exception",
+        store=True,
+        index=True,
+        help="This will represent a situation where several moves are pointing"
+        "to the location for different product lots and the location does"
+        "not allow mixed lots.",
+    )
+
     def init(self):  # pylint: disable=missing-return
         super().init()
         if not index_exists(self._cr, "stock_move_line_location_state_index"):
@@ -155,6 +164,38 @@
                 """
             )
 
+    @api.depends("do_not_mix_lots", "location_will_contain_lot_ids", "fill_state")
+    def _compute_has_potential_lot_mix_exception(self):
+        locations_with_exception = self.browse()
+        locations_without_exception = self.browse()
+        for location in self:
+            if (
+                location.fill_state not in ("empty", "being_emptied")
+                and location._should_compute_will_contain_lot_ids()
+                and len(location.location_will_contain_lot_ids) > 1
+            ):
+                locations_with_exception |= location
+            else:
+                locations_without_exception |= location
+        locations_with_exception.has_potential_lot_mix_exception = True
+        locations_without_exception.has_potential_lot_mix_exception = False
+
+    @api.depends("do_not_mix_lots", "location_will_contain_product_ids", "fill_state")
+    def _compute_has_potential_product_mix_exception(self):
+        locations_with_exception = self.browse()
+        locations_without_exception = self.browse()
+        for location in self:
+            if (
+                location.fill_state not in ("empty", "being_emptied")
+                and location._should_compute_will_contain_product_ids()
+                and len(location.location_will_contain_product_ids) > 1
+            ):
+                locations_with_exception |= location
+            else:
+                locations_without_exception |= location
+        locations_with_exception.has_potential_product_mix_exception = True
+        locations_without_exception.has_potential_product_mix_exception = False
+
     @api.depends(
         "usage",
         "computed_storage_category_id.allow_new_product",
@@ -263,9 +304,6 @@
     def _should_compute_will_contain_lot_ids(self):
         return self.do_not_mix_lots
 
-    def _should_compute_location_is_empty(self):
-        return self.only_empty
-
     @api.depends(
         "quant_ids.quantity",
         "in_move_ids",
@@ -303,54 +341,6 @@
                 ).lot_id | rec.mapped("in_move_line_ids.lot_id")
                 rec.location_will_contain_lot_ids = lots
 
-    @api.depends(
-        "quant_ids.quantity",
-        "out_move_line_ids.qty_done",
-        "in_move_ids",
-        "in_move_line_ids",
-        "only_empty",
-    )
-    def _compute_location_is_empty(self):
-        # No restriction should apply on customer/supplier/...
-        # locations and we don't need to compute is empty
-        # if there is no limit on the location
-        only_empty_locations = self.filtered(
-            lambda l: not l._should_compute_location_is_empty()
-        )
-        only_empty_locations.update({"location_is_empty": True})
-        records = self - only_empty_locations
-        if not records:
-            return
-        location_domain = [("location_id", "in", records.ids)]
-        out_qty_by_location = {}
-        qty_by_location = {}
-        for group in self.env["stock.move.line"].read_group(
-            OUT_MOVE_LINE_DOMAIN + location_domain,
-            fields=["qty_done:sum"],
-            groupby=["location_id"],
-        ):
-            location_id = group["location_id"][0]
-            out_qty_by_location[location_id] = group["qty_done"]
-        for group in self.env["stock.quant"].read_group(
-            location_domain, fields=["quantity:sum"], groupby=["location_id"]
-        ):
-            location_id = group["location_id"][0]
-            qty_by_location[location_id] = group["quantity"]
-        for rec in records:
-            # we do want to keep a write here even if the value is the same
-            # to enforce concurrent transaction safety: 2 moves taking
-            # quantities in a location have to be executed sequentially
-            # or the location could remain "not empty"
-            if (
-                qty_by_location.get(rec.id, 0.0) - out_qty_by_location.get(rec.id, 0.0)
-                > 0
-                or rec.in_move_ids
-                or rec.in_move_line_ids
-            ):
-                rec.location_is_empty = False
-            else:
-                rec.location_is_empty = True
-
     # method provided by "stock_putaway_hook"
     def _putaway_strategy_finalizer(
         self,
@@ -635,18 +625,17 @@
         valid_no_mix = valid_locations.filtered("do_not_mix_products")
         loc_ordered_by_qty = []
         if valid_no_mix:
-            StockQuant = self.env["stock.quant"]
-            domain_quant = [("location_id", "in", valid_no_mix.ids)]
-            loc_ordered_by_qty = [
-                item["location_id"][0]
-                for item in StockQuant.read_group(
-                    domain_quant,
-                    ["location_id", "quantity"],
-                    ["location_id"],
-                    orderby="quantity",
+            for location, items in groupby(
+                valid_no_mix.quant_ids.sorted("quantity"),
+                lambda quant: quant.location_id,
+            ):
+                loc_ordered_by_qty.extend(
+                    [
+                        location.id
+                        for item in items
+                        if (float_compare(item["quantity"], 0, precision_digits=2) > 0)
+                    ]
                 )
-                if (float_compare(item["quantity"], 0, precision_digits=2) > 0)
-            ]
         valid_location_ids = set(valid_locations.ids) - set(loc_ordered_by_qty)
         ordered_valid_location_ids = loc_ordered_by_qty + [
             id_ for id_ in self.ids if id_ in valid_location_ids

stock_storage_type/models/stock_storage_category.py

@@ -1,6 +1,7 @@
 # Copyright 2022 ACSONE SA
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl)
 from odoo import api, fields, models
+from odoo.osv.expression import AND, OR
 
 
 class StockStorageCategory(models.Model):
@@ -128,21 +129,53 @@ def _compute_has_restrictions(self):
                 ]
             )
 
+    def _get_product_lot_location_domain(self, lots):
+        """
+        Helper to get product lots domain
+        """
+        lot_domain = OR(
+            [
+                [
+                    ("location_will_contain_lot_ids", "in", lots.ids),
+                ],
+                [
+                    ("location_will_contain_lot_ids", "=", False),
+                ],
+            ]
+        )
+
+        location_domain = OR(
+            [lot_domain, [("fill_state", "in", ("empty", "being_emptied"))]]
+        )
+
+        return location_domain
+
     def _get_product_location_domain(self, products):
         """
         Helper to get products location domain
         """
-        return [
-            "|",
-            # Ideally, we would like a domain which is a strict comparison:
-            # if we do not mix products, we should be able to filter on ==
-            # product.id. Here, if we can create a move for product B and
-            # set it's destination in a location already used by product A,
-            # then all the new moves for product B will be allowed in the
-            # location.
-            ("location_will_contain_product_ids", "in", products.ids),
-            ("location_will_contain_product_ids", "=", False),
-        ]
+        # Ideally, we would like a domain which is a strict comparison:
+        # if we do not mix products, we should be able to filter on ==
+        # product.id. Here, if we can create a move for product B and
+        # set it's destination in a location already used by product A,
+        # then all the new moves for product B will be allowed in the
+        # location.
+
+        # Take only locations that has no potential different products
+        # in it.
+        product_domain = OR(
+            [
+                [
+                    ("has_potential_product_mix_exception", "=", False),
+                    ("location_will_contain_product_ids", "in", products.ids),
+                ],
+                [("location_will_contain_product_ids", "=", False)],
+            ]
+        )
+        location_domain = OR(
+            [product_domain, [("fill_state", "in", ("empty", "being_emptied"))]]
+        )
+        return location_domain
 
     def _domain_location_storage_category(
         self, candidate_locations, quants, products, package_type
@@ -164,19 +197,56 @@ def _domain_location_storage_category(
             quants=quants,
         )
         if allow_new_product == "empty":
-            location_domain.append(("location_is_empty", "=", True))
+            # We should include the destination location of the current
+            # stock move line to avoid excluding it if already selected
+            # Indeed, if the current move line point to the last void location,
+            # calling the putaway apply will recompute the destination location
+            # to the related stock.move destination as the rules consider
+            # there is no more room available (which is not true).
+            exclude_sml_ids = self.env.context.get("exclude_sml_ids")
+            if exclude_sml_ids:
+                lines_locations = (
+                    self.env["stock.move.line"].browse(exclude_sml_ids).location_dest_id
+                )
+                if lines_locations:
+                    location_domain = AND(
+                        [
+                            location_domain,
+                            OR(
+                                [
+                                    [
+                                        (
+                                            "fill_state",
+                                            "in",
+                                            ("filled", "being_filled"),
+                                        ),
+                                        ("id", "in", lines_locations.ids),
+                                    ],
+                                    [("fill_state", "in", ("empty", "being_emptied"))],
+                                ]
+                            ),
+                        ]
+                    )
+            else:
+                location_domain = AND(
+                    [
+                        location_domain,
+                        [("fill_state", "in", ("empty", "being_emptied"))],
+                    ]
+                )
         elif allow_new_product == "same":
-            location_domain += self._get_product_location_domain(products)
+            location_domain = AND(
+                [location_domain, self._get_product_location_domain(products)]
+            )
         elif allow_new_product == "same_lot":
             lots = quants.mapped("lot_id")
             # As same lot should filter also on same product
-            location_domain += self._get_product_location_domain(products)
-            location_domain += [
-                "|",
-                # same comment as for the products
-                ("location_will_contain_lot_ids", "in", lots.ids),
-                ("location_will_contain_lot_ids", "=", False),
-            ]
+            location_domain = AND(
+                [location_domain, self._get_product_location_domain(products)]
+            )
+            location_domain = AND(
+                [location_domain, self._get_product_lot_location_domain(lots)]
+            )
         return location_domain
 
     def get_allow_new_product(

stock_storage_type/tests/__init__.py

@@ -1,7 +1,6 @@
 from . import (
     test_auto_assign_storage_type,
     test_package_height_required,
-    test_package_type_message,
     test_stock_location,
     test_storage_type,
     test_storage_type_move,