awesome improvements

stock_valuation_fifo_lot/__manifest__.py

@@ -1,4 +1,5 @@
 # Copyright 2023 Ecosoft Co., Ltd (https://ecosoft.co.th)
+# Copyright 2024 Quartile (https://www.quartile.co)
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html)
 
 {
@@ -7,11 +8,12 @@
     "category": "Warehouse Management",
     "development_status": "Alpha",
     "license": "AGPL-3",
-    "author": "Ecosoft, Odoo Community Association (OCA)",
+    "author": "Ecosoft, Quartile, Odoo Community Association (OCA)",
     "website": "https://github.com/OCA/stock-logistics-workflow",
     "depends": ["stock_account", "stock_no_negative"],
     "data": [
         "views/res_config_settings_views.xml",
+        "views/stock_move_line_views.xml",
         "views/stock_valuation_layer_views.xml",
     ],
     "installable": True,

stock_valuation_fifo_lot/models/__init__.py

@@ -4,4 +4,5 @@
 from . import res_company
 from . import res_config_settings
 from . import stock_move
+from . import stock_move_line
 from . import stock_valuation_layer

stock_valuation_fifo_lot/models/product.py

@@ -1,111 +1,82 @@
 # Copyright 2023 Ecosoft Co., Ltd (https://ecosoft.co.th)
+# Copyright 2024 Quartile (https://www.quartile.co)
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html)
 
+from collections import defaultdict
+
 from odoo import models
+from odoo.osv import expression
 from odoo.tools import float_is_zero
 
 
 class ProductProduct(models.Model):
     _inherit = "product.product"
 
+    def _get_fifo_candidates_domain(self, company):
+        res = super()._get_fifo_candidates_domain(company)
+        fifo_lot = self.env.context.get("fifo_lot")
+        if not fifo_lot:
+            return res
+        return expression.AND([res, [("lot_ids", "in", fifo_lot.ids)]])
+
     def _sort_by_all_candidates(self, all_candidates, sort_by):
         """Hook function for other sort by"""
         return all_candidates
 
     def _get_fifo_candidates(self, company):
         all_candidates = super()._get_fifo_candidates(company)
+        fifo_lot = self.env.context.get("fifo_lot")
+        if fifo_lot:
+            for svl in all_candidates:
+                if not svl._get_unconsumed_in_move_line(fifo_lot):
+                    all_candidates -= svl
         sort_by = self.env.context.get("sort_by")
         if sort_by == "lot_create_date":
 
             def sorting_key(candidate):
                 if candidate.lot_ids:
                     return min(candidate.lot_ids.mapped("create_date"))
-                else:
-                    return candidate.create_date
+                return candidate.create_date
 
             all_candidates = all_candidates.sorted(key=sorting_key)
         elif sort_by is not None:
             all_candidates = self._sort_by_all_candidates(all_candidates, sort_by)
         return all_candidates
 
-    def _run_fifo(self, quantity, company):
-        self.ensure_one()
-        move_id = self._context.get("used_in_move_id")
-        if self.tracking == "none" or not move_id:
-            return super()._run_fifo(quantity, company)
-
-        move = self.env["stock.move"].browse(move_id)
-        move_lines = move._get_out_move_lines()
-        tmp_value = 0
-        tmp_remaining_qty = 0
-        for move_line in move_lines:
-            # Find back incoming stock valuation layers
-            # (called candidates here) to value `quantity`.
-            qty_to_take_on_candidates = move_line.product_uom_id._compute_quantity(
-                move_line.qty_done, move.product_id.uom_id
+    # Depends on https://github.com/odoo/odoo/pull/180245
+    def _get_qty_taken_on_candidate(self, qty_to_take_on_candidates, candidate):
+        fifo_lot = self.env.context.get("fifo_lot")
+        if fifo_lot:
+            candidate_move_line = candidate._get_unconsumed_in_move_line(fifo_lot)
+            qty_to_take_on_candidates = min(
+                qty_to_take_on_candidates, candidate_move_line.qty_remaining
             )
-            # Find incoming stock valuation layers that have lot_ids on their moves
-            # Check with stock_move_id.lot_ids to cover the situation where the stock
-            # was received either before or after the installation of this module
-            candidates = self._get_fifo_candidates(company).filtered(
-                lambda l: move_line.lot_id in l.stock_move_id.lot_ids
+            candidate_move_line.qty_consumed += qty_to_take_on_candidates
+            candidate_move_line.cost_consumed += qty_to_take_on_candidates * (
+                candidate.remaining_value / candidate.remaining_qty
             )
-            for candidate in candidates:
-                qty_taken_on_candidate = min(
-                    qty_to_take_on_candidates, candidate.remaining_qty
-                )
-
-                candidate_unit_cost = (
-                    candidate.remaining_value / candidate.remaining_qty
-                )
-                value_taken_on_candidate = qty_taken_on_candidate * candidate_unit_cost
-                value_taken_on_candidate = candidate.currency_id.round(
-                    value_taken_on_candidate
-                )
-                new_remaining_value = (
-                    candidate.remaining_value - value_taken_on_candidate
-                )
-
-                candidate_vals = {
-                    "remaining_qty": candidate.remaining_qty - qty_taken_on_candidate,
-                    "remaining_value": new_remaining_value,
-                }
-
-                candidate.write(candidate_vals)
+        return super()._get_qty_taken_on_candidate(qty_to_take_on_candidates, candidate)
 
-                qty_to_take_on_candidates -= qty_taken_on_candidate
-                tmp_value += value_taken_on_candidate
-
-                if float_is_zero(
-                    qty_to_take_on_candidates,
-                    precision_rounding=self.uom_id.rounding,
-                ):
-                    break
-
-            if candidates and qty_to_take_on_candidates > 0:
-                tmp_value += abs(candidate.unit_cost * -qty_to_take_on_candidates)
-                tmp_remaining_qty += qty_to_take_on_candidates
-
-        # Calculate standard price (Sorted by lot created date)
-        all_candidates = self.with_context(
-            sort_by="lot_create_date"
-        )._get_fifo_candidates(company)
-        new_standard_price = 0.0
-        if all_candidates:
-            new_standard_price = all_candidates[0].unit_cost
-        elif candidates:
-            new_standard_price = candidate.unit_cost
-
-        # Update standard price
-        if new_standard_price and self.cost_method == "fifo":
-            self.sudo().with_company(company.id).with_context(
-                disable_auto_svl=True
-            ).standard_price = new_standard_price
-
-        # Value
-        vals = {
-            "remaining_qty": -tmp_remaining_qty,
-            "value": -tmp_value,
-            "unit_cost": tmp_value / (quantity + tmp_remaining_qty),
-        }
+    def _run_fifo(self, quantity, company):
+        self.ensure_one()
+        fifo_move = self._context.get("fifo_move")
+        if self.tracking == "none" or not fifo_move:
+            return super()._run_fifo(quantity, company)
+        remaining_qty = quantity
+        vals = defaultdict(float)
+        correction_move_line = self.env.context.get("correction_move_line")
+        move_lines = correction_move_line or fifo_move._get_out_move_lines()
+        for ml in move_lines:
+            ml_qty = fifo_move.product_uom._compute_quantity(ml.qty_done, self.uom_id)
+            fifo_qty = min(remaining_qty, ml_qty)
+            self = self.with_context(fifo_lot=ml.lot_id, fifo_qty=fifo_qty)
+            ml_fifo_vals = super()._run_fifo(fifo_qty, company)
+            for key, value in ml_fifo_vals.items():
+                if key in ("remaining_qty", "value"):
+                    vals[key] += value
+                    continue
+                vals[key] = value  # unit_cost
+            remaining_qty -= fifo_qty
+            if float_is_zero(remaining_qty, precision_rounding=self.uom_id.rounding):
+                break
         return vals

stock_valuation_fifo_lot/models/stock_move.py

@@ -1,98 +1,73 @@
 # Copyright 2023 Ecosoft Co., Ltd (https://ecosoft.co.th)
+# Copyright 2024 Quartile (https://www.quartile.co)
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html)
 
-from odoo import models
+from odoo import Command, models
 
 
 class StockMove(models.Model):
     _inherit = "stock.move"
 
     def _prepare_common_svl_vals(self):
-        """
-        Prepare lots/serial numbers on stock valuation report
-        """
+        """Add lots/serials to the stock valuation layer."""
         self.ensure_one()
         res = super()._prepare_common_svl_vals()
-        res.update(
-            {
-                "lot_ids": [(6, 0, self.lot_ids.ids)],
-            }
-        )
+        res.update({"lot_ids": [Command.set(self.lot_ids.ids)]})
         return res
 
     def _create_out_svl(self, forced_quantity=None):
-        """
-        Send context current move to _create_out_svl function
-        """
+        """Set the move as a context for processing in _run_fifo()."""
         layers = self.env["stock.valuation.layer"]
         for move in self:
-            move = move.with_context(used_in_move_id=move.id)
+            move = move.with_context(fifo_move=move)
             layer = super(StockMove, move)._create_out_svl(
                 forced_quantity=forced_quantity
             )
             layers |= layer
         return layers
 
     def _create_in_svl(self, forced_quantity=None):
-        """
-        1. Check stock move - Multiple lot on the stock move is not
-           allowed for incoming transfer
-        2. Change product standard price to first available lot price
-        """
+        """Change product standard price to the first available lot price."""
         layers = self.env["stock.valuation.layer"]
         for move in self:
             layer = super(StockMove, move)._create_in_svl(
                 forced_quantity=forced_quantity
             )
-            # Calculate standard price (Sorted by lot created date)
-            if (
-                move.product_id.cost_method == "fifo"
-                and move.product_id.tracking != "none"
-            ):
-                all_candidates = move.product_id.with_context(
-                    sort_by="lot_create_date"
-                )._get_fifo_candidates(move.company_id)
-                if all_candidates:
-                    move.product_id.sudo().with_company(
-                        move.company_id.id
-                    ).with_context(
-                        disable_auto_svl=True
-                    ).standard_price = all_candidates[
-                        0
-                    ].unit_cost
+            product = move.product_id
+            # Calculate standard price (sorted by lot created date)
+            if product.cost_method != "fifo" or product.tracking == "none":
+                continue
+            product = product.with_context(sort_by="lot_create_date")
+            candidate = product._get_fifo_candidates(move.company_id)[:1]
+            if not candidate:
+                continue
+            product = product.with_company(move.company_id.id)
+            product = product.with_context(disable_auto_svl=True)
+            product.sudo().standard_price = candidate.unit_cost
             layers |= layer
         return layers
 
     def _get_price_unit(self):
-        """
-        No PO, Get price unit from lot price
+        """No PO (e.g. customer returns) and get the price unit from the last consumed
+        incoming move line for the lot.
         """
         self.ensure_one()
         if not self.company_id.use_lot_get_price_unit_fifo:
             return super()._get_price_unit()
-        if (
-            hasattr(self, "purchase_line_id")
-            and not self.purchase_line_id
-            and self.product_id.cost_method == "fifo"
-            and len(self.lot_ids) == 1
-        ):
-            candidate = (
-                self.env["stock.valuation.layer"]
-                .sudo()
-                .search(
-                    [
-                        ("product_id", "=", self.product_id.id),
-                        (
-                            "lot_ids",
-                            "in",
-                            self.lot_ids.ids,
-                        ),
-                        ("remaining_qty", ">", 0),
-                        ("company_id", "=", self.company_id.id),
-                    ],
-                    limit=1,
-                )
+        if hasattr(self, "purchase_line_id") and self.purchase_line_id:
+            return super()._get_price_unit()
+        if self.product_id.cost_method == "fifo" and len(self.lot_ids) == 1:
+            # Get the last consumed incoming move line.
+            move_line = self.env["stock.move.line"].search(
+                [
+                    ("product_id", "=", self.product_id.id),
+                    ("lot_id", "=", self.lot_ids.id),
+                    ("qty_consumed", ">", 0),
+                    ("company_id", "=", self.company_id.id),
+                ],
+                order="id desc",
+                limit=1,
             )
-            if candidate:
-                return candidate.remaining_value / candidate.remaining_qty
+            if move_line:
+                return move_line.cost_consumed / move_line.qty_consumed
         return super()._get_price_unit()

stock_valuation_fifo_lot/models/stock_move_line.py

@@ -0,0 +1,41 @@
+# Copyright 2024 Quartile (https://www.quartile.co)
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html)
+
+from odoo import api, fields, models
+
+
+class StockMoveLine(models.Model):
+    _inherit = "stock.move.line"
+
+    qty_consumed = fields.Float(
+        help="Quantity that has gone out of the inventory for valued incoming moves "
+        "for FIFO products with a lot/serial.",
+    )
+    qty_remaining = fields.Float(
+        compute="_compute_qty_remaining",
+        store=True,
+        help="Remaining quantity for valued incoming moves for FIFO products with a "
+        "lot/serial.",
+    )
+    company_currency_id = fields.Many2one(related="company_id.currency_id")
+    cost_consumed = fields.Monetary(
+        currency_field="company_currency_id",
+        help="The value of the inventory that has been consumed for FIFO products with "
+        "a lot/serial.",
+    )
+
+    @api.depends("qty_done", "qty_consumed")
+    def _compute_qty_remaining(self):
+        for rec in self:
+            if rec.location_usage not in (
+                "internal",
+                "transit",
+            ) and rec.location_dest_usage in ("internal", "transit"):
+                rec.qty_remaining = rec.qty_done - rec.qty_consumed
+
+    def _create_correction_svl(self, move, diff):
+        # Pass the move line as a context value in case qty_done is overridden in a done
+        # transfer, to correctly identify which record should be processed in
+        # _run_fifo().
+        move = move.with_context(correction_move_line=self)
+        return super()._create_correction_svl(move, diff)

stock_valuation_fifo_lot/models/stock_valuation_layer.py

@@ -11,3 +11,9 @@ class StockValuationLayer(models.Model):
         comodel_name="stock.lot",
         string="Lots/Serial Numbers",
     )
+
+    def _get_unconsumed_in_move_line(self, lot):
+        self.ensure_one()
+        return self.stock_move_id.move_line_ids.filtered(
+            lambda x: x.lot_id == lot and x.qty_remaining
+        )

stock_valuation_fifo_lot/tests/test_stock_valuation_fifo_lot.py

@@ -1,6 +1,3 @@
-# Copyright 2024 Quartile (https://www.quartile.co)
-# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html)
-
 # Copyright 2024 Quartile (https://www.quartile.co)
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).