875 lines
50 KiB
Python
875 lines
50 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
|
||
|
|
import copy
|
||
|
|
import json
|
||
|
|
from collections import defaultdict
|
||
|
|
from odoo import _, api, fields, models
|
||
|
|
from odoo.tools import float_compare, float_repr, float_round, float_is_zero, format_date, get_lang
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from math import log10
|
||
|
|
|
||
|
|
class ReportMoOverview(models.AbstractModel):
|
||
|
|
_name = 'report.mrp.report_mo_overview'
|
||
|
|
_description = 'MO Overview Report'
|
||
|
|
|
||
|
|
@api.model
|
||
|
|
def get_report_values(self, production_id):
|
||
|
|
""" Endpoint for HTML display. """
|
||
|
|
return {
|
||
|
|
'data': self._get_report_data(production_id),
|
||
|
|
'context': self._get_display_context(),
|
||
|
|
}
|
||
|
|
|
||
|
|
@api.model
|
||
|
|
def _get_report_values(self, docids, data=None):
|
||
|
|
""" Endpoint for PDF display. """
|
||
|
|
docs = []
|
||
|
|
for prod_id in docids:
|
||
|
|
doc = self._get_report_data(prod_id)
|
||
|
|
docs.append(self._include_pdf_specifics(doc, data))
|
||
|
|
return {
|
||
|
|
'doc_ids': docids,
|
||
|
|
'doc_model': 'mrp.production',
|
||
|
|
'docs': docs,
|
||
|
|
}
|
||
|
|
|
||
|
|
def _include_pdf_specifics(self, doc, data=None):
|
||
|
|
def get_color(decorator):
|
||
|
|
return f"text-{decorator}" if decorator else ''
|
||
|
|
|
||
|
|
if not data:
|
||
|
|
data = {}
|
||
|
|
footer_colspan = 2 # Name & Quantity
|
||
|
|
doc['show_replenishments'] = data.get('replenishments') == '1'
|
||
|
|
if doc['show_replenishments']:
|
||
|
|
footer_colspan += 1
|
||
|
|
doc['show_availabilities'] = data.get('availabilities') == '1'
|
||
|
|
if doc['show_availabilities']:
|
||
|
|
footer_colspan += 2 # Free to use / On Hand & Reserved
|
||
|
|
doc['show_receipts'] = data.get('receipts') == '1'
|
||
|
|
if doc['show_receipts']:
|
||
|
|
footer_colspan += 1
|
||
|
|
doc['show_unit_costs'] = data.get('unitCosts') == '1'
|
||
|
|
if doc['show_unit_costs']:
|
||
|
|
footer_colspan += 1
|
||
|
|
doc['show_mo_costs'] = data.get('moCosts') == '1'
|
||
|
|
doc['show_real_costs'] = data.get('realCosts') == '1'
|
||
|
|
doc['show_uom'] = self.env.user.user_has_groups('uom.group_uom')
|
||
|
|
if doc['show_uom']:
|
||
|
|
footer_colspan += 1
|
||
|
|
doc['data_mo_unit_cost'] = doc['summary'].get('mo_cost', 0) / (doc['summary'].get('quantity') or 1)
|
||
|
|
doc['data_real_unit_cost'] = doc['summary'].get('real_cost', 0) / (doc['summary'].get('quantity') or 1)
|
||
|
|
doc['unfolded_ids'] = set(json.loads(data.get('unfoldedIds', '[]')))
|
||
|
|
doc['footer_colspan'] = footer_colspan
|
||
|
|
doc['get_color'] = get_color
|
||
|
|
return doc
|
||
|
|
|
||
|
|
def _get_display_context(self):
|
||
|
|
return {
|
||
|
|
'show_uom': self.env.user.user_has_groups('uom.group_uom'),
|
||
|
|
}
|
||
|
|
|
||
|
|
def _get_report_data(self, production_id):
|
||
|
|
production = self.env['mrp.production'].browse(production_id)
|
||
|
|
# Necessary to fetch the right quantities for multi-warehouse
|
||
|
|
production = production.with_context(warehouse=production.warehouse_id.id)
|
||
|
|
|
||
|
|
components = self._get_components_data(production, level=1, current_index='')
|
||
|
|
operations = self._get_operations_data(production, level=1, current_index='')
|
||
|
|
initial_mo_cost, initial_real_cost = self._compute_cost_sums(components, operations)
|
||
|
|
remaining_cost_share, byproducts = self._get_byproducts_data(production, initial_mo_cost, initial_real_cost, level=1, current_index='')
|
||
|
|
summary = self._get_mo_summary(production, components, initial_mo_cost, initial_real_cost, remaining_cost_share)
|
||
|
|
extra_lines = self._get_report_extra_lines(summary, components, operations, production.state == 'done')
|
||
|
|
return {
|
||
|
|
'id': production.id,
|
||
|
|
'name': production.display_name,
|
||
|
|
'summary': summary,
|
||
|
|
'components': components,
|
||
|
|
'operations': operations,
|
||
|
|
'byproducts': byproducts,
|
||
|
|
'extras': extra_lines,
|
||
|
|
'cost_breakdown': self._get_cost_breakdown_data(production, extra_lines, remaining_cost_share),
|
||
|
|
}
|
||
|
|
|
||
|
|
def _get_report_extra_lines(self, summary, components, operations, production_done=False):
|
||
|
|
currency = summary.get('currency', self.env.company.currency_id)
|
||
|
|
unit_mo_cost = currency.round(summary.get('mo_cost', 0) / (summary.get('quantity') or 1))
|
||
|
|
unit_real_cost = currency.round(summary.get('real_cost', 0) / (summary.get('quantity') or 1))
|
||
|
|
extras = {
|
||
|
|
'unit_mo_cost': unit_mo_cost,
|
||
|
|
'unit_mo_cost_decorator': self._get_comparison_decorator(unit_real_cost, unit_mo_cost, currency.rounding),
|
||
|
|
'unit_real_cost': unit_real_cost,
|
||
|
|
}
|
||
|
|
if production_done:
|
||
|
|
production_qty = summary.get('quantity', 1.0)
|
||
|
|
extras['total_mo_cost_components'] = sum(compo.get('summary', {}).get('mo_cost', 0.0) for compo in components)
|
||
|
|
extras['total_real_cost_components'] = sum(compo.get('summary', {}).get('real_cost', 0.0) for compo in components)
|
||
|
|
extras['total_mo_cost_components_decorator'] = self._get_comparison_decorator(extras['total_real_cost_components'], extras['total_mo_cost_components'], currency.rounding)
|
||
|
|
extras['unit_mo_cost_components'] = extras['total_mo_cost_components'] / production_qty
|
||
|
|
extras['unit_real_cost_components'] = extras['total_real_cost_components'] / production_qty
|
||
|
|
extras['total_mo_cost_operations'] = operations.get('summary', {}).get('mo_cost', 0.0)
|
||
|
|
extras['total_real_cost_operations'] = operations.get('summary', {}).get('real_cost', 0.0)
|
||
|
|
extras['total_mo_cost_operations_decorator'] = self._get_comparison_decorator(extras['total_real_cost_operations'], extras['total_mo_cost_operations'], currency.rounding)
|
||
|
|
extras['unit_mo_cost_operations'] = extras['total_mo_cost_operations'] / production_qty
|
||
|
|
extras['unit_real_cost_operations'] = extras['total_real_cost_operations'] / production_qty
|
||
|
|
extras['total_mo_cost'] = extras['total_mo_cost_components'] + extras['total_mo_cost_operations']
|
||
|
|
extras['total_real_cost'] = extras['total_real_cost_components'] + extras['total_real_cost_operations']
|
||
|
|
extras['total_mo_cost_decorator'] = self._get_comparison_decorator(extras['total_real_cost'], extras['total_mo_cost'], currency.rounding)
|
||
|
|
return extras
|
||
|
|
|
||
|
|
def _get_cost_breakdown_data(self, production, extras, remaining_cost_share):
|
||
|
|
if production.state != 'done' or not production.move_byproduct_ids:
|
||
|
|
return []
|
||
|
|
|
||
|
|
# Prepare byproducts data
|
||
|
|
quantities_by_product = defaultdict(float)
|
||
|
|
total_cost_by_product = defaultdict(float)
|
||
|
|
component_cost_by_product = defaultdict(float)
|
||
|
|
operation_cost_by_product = defaultdict(float)
|
||
|
|
for bp_move in production.move_byproduct_ids:
|
||
|
|
# Byproducts without cost share are irrelevant in a cost breakdrown.
|
||
|
|
if bp_move.state == 'cancel' or float_is_zero(bp_move.cost_share, precision_digits=2):
|
||
|
|
continue
|
||
|
|
# As UoMs can vary, we use the default UoM of each product
|
||
|
|
quantities_by_product[bp_move.product_id] += bp_move.product_qty
|
||
|
|
cost_share = bp_move.cost_share / 100
|
||
|
|
total_cost_by_product[bp_move.product_id] += extras['total_real_cost'] * cost_share
|
||
|
|
component_cost_by_product[bp_move.product_id] += extras['total_real_cost_components'] * cost_share
|
||
|
|
operation_cost_by_product[bp_move.product_id] += extras['total_real_cost_operations'] * cost_share
|
||
|
|
|
||
|
|
# Add finished product to its own default UoM (not the production UoM)
|
||
|
|
breakdown_lines = [self._format_cost_breakdown_lines(0, production.product_id.display_name, production.product_id.uom_id.display_name,
|
||
|
|
(extras['total_real_cost_components'] * remaining_cost_share) / production.product_uom_qty,
|
||
|
|
(extras['total_real_cost_operations'] * remaining_cost_share) / production.product_uom_qty,
|
||
|
|
(extras['total_real_cost'] * remaining_cost_share) / production.product_uom_qty)]
|
||
|
|
for index, product in enumerate(quantities_by_product.keys()):
|
||
|
|
breakdown_lines.append(self._format_cost_breakdown_lines(index + 1, product.display_name, product.uom_id.display_name,
|
||
|
|
component_cost_by_product[product] / quantities_by_product[product],
|
||
|
|
operation_cost_by_product[product] / quantities_by_product[product],
|
||
|
|
total_cost_by_product[product] / quantities_by_product[product]))
|
||
|
|
return breakdown_lines
|
||
|
|
|
||
|
|
def _format_cost_breakdown_lines(self, index, product_name, uom_name, component_cost, operation_cost, total_cost):
|
||
|
|
return {
|
||
|
|
'index': f"BR{index}",
|
||
|
|
'name': product_name,
|
||
|
|
'unit_avg_cost_component': component_cost,
|
||
|
|
'unit_avg_cost_operation': operation_cost,
|
||
|
|
'unit_avg_total_cost': total_cost,
|
||
|
|
'uom_name': uom_name,
|
||
|
|
}
|
||
|
|
|
||
|
|
def _get_mo_summary(self, production, components, current_mo_cost, current_real_cost, remaining_cost_share):
|
||
|
|
currency = (production.company_id or self.env.company).currency_id
|
||
|
|
product = production.product_id
|
||
|
|
mo_cost = current_mo_cost * remaining_cost_share
|
||
|
|
real_cost = current_real_cost * remaining_cost_share
|
||
|
|
return {
|
||
|
|
'level': 0,
|
||
|
|
'model': production._name,
|
||
|
|
'id': production.id,
|
||
|
|
'name': production.product_id.display_name,
|
||
|
|
'product_model': production.product_id._name,
|
||
|
|
'product_id': production.product_id.id,
|
||
|
|
'state': production.state,
|
||
|
|
'formatted_state': self._format_state(production, components),
|
||
|
|
'quantity': production.product_qty if production.state != 'done' else production.qty_produced,
|
||
|
|
'uom_name': production.product_uom_id.display_name,
|
||
|
|
'uom_precision': self._get_uom_precision(production.product_uom_id.rounding or 0.01),
|
||
|
|
'quantity_free': product.uom_id._compute_quantity(max(product.free_qty, 0), production.product_uom_id) if product.type == 'product' else False,
|
||
|
|
'quantity_on_hand': product.uom_id._compute_quantity(product.qty_available, production.product_uom_id) if product.type == 'product' else False,
|
||
|
|
'quantity_reserved': 0.0,
|
||
|
|
'receipt': self._check_planned_start(production.date_deadline, self._get_replenishment_receipt(production, components)),
|
||
|
|
'unit_cost': self._get_unit_cost(production.move_finished_ids.filtered(lambda m: m.product_id == production.product_id)),
|
||
|
|
'mo_cost': currency.round(mo_cost),
|
||
|
|
'mo_cost_decorator': self._get_comparison_decorator(real_cost, mo_cost, currency.rounding),
|
||
|
|
'real_cost': currency.round(real_cost),
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
}
|
||
|
|
|
||
|
|
def _get_unit_cost(self, move):
|
||
|
|
if not move:
|
||
|
|
return 0.0
|
||
|
|
return move.product_id.uom_id._compute_price(move.product_id.standard_price, move.product_uom)
|
||
|
|
|
||
|
|
def _format_state(self, record, components=False):
|
||
|
|
""" For MOs, provide a custom state based on the demand vs quantities available for components.
|
||
|
|
All other records types will provide their standard state value.
|
||
|
|
:param dict components: components in the structure provided by `_get_components_data`
|
||
|
|
:return: string to be used as custom state
|
||
|
|
"""
|
||
|
|
if record._name != 'mrp.production' or record.state not in ('draft', 'confirmed') or not components:
|
||
|
|
return dict(record._fields['state']._description_selection(self.env)).get(record.state)
|
||
|
|
components_qty_to_produce = defaultdict(float)
|
||
|
|
components_qty_reserved = defaultdict(float)
|
||
|
|
components_qty_free = defaultdict(float)
|
||
|
|
for component in components:
|
||
|
|
component = component["summary"]
|
||
|
|
product = component["product"]
|
||
|
|
if product.type != 'product':
|
||
|
|
continue
|
||
|
|
uom = component["uom"]
|
||
|
|
components_qty_to_produce[product] += uom._compute_quantity(component["quantity"], product.uom_id)
|
||
|
|
components_qty_reserved[product] += uom._compute_quantity(component["quantity_reserved"], product.uom_id)
|
||
|
|
components_qty_free[product] = uom._compute_quantity(component["quantity_free"], product.uom_id)
|
||
|
|
producible_qty = record.product_qty
|
||
|
|
for product, comp_qty_to_produce in components_qty_to_produce.items():
|
||
|
|
if float_is_zero(comp_qty_to_produce, precision_rounding=product.uom_id.rounding):
|
||
|
|
continue
|
||
|
|
comp_producible_qty = float_round(
|
||
|
|
record.product_qty * (components_qty_reserved[product] + components_qty_free[product]) / comp_qty_to_produce,
|
||
|
|
precision_rounding=record.product_uom_id.rounding, rounding_method='DOWN'
|
||
|
|
)
|
||
|
|
if float_compare(comp_producible_qty, 0, precision_rounding=record.product_uom_id.rounding) <= 0:
|
||
|
|
return _("Not Ready")
|
||
|
|
producible_qty = min(comp_producible_qty, producible_qty)
|
||
|
|
if float_compare(producible_qty, 0, precision_rounding=record.product_uom_id.rounding) <= 0:
|
||
|
|
return _("Not Ready")
|
||
|
|
elif float_compare(producible_qty, record.product_qty, precision_rounding=record.product_uom_id.rounding) == -1:
|
||
|
|
producible_qty = float_repr(producible_qty, self.env['decimal.precision'].precision_get('Product Unit of Measure'))
|
||
|
|
return _("%(producible_qty)s Ready", producible_qty=producible_qty)
|
||
|
|
return _("Ready")
|
||
|
|
|
||
|
|
def _get_uom_precision(self, uom_rounding):
|
||
|
|
return max(0, int(-(log10(uom_rounding))))
|
||
|
|
|
||
|
|
def _get_comparison_decorator(self, expected, current, rounding):
|
||
|
|
compare = float_compare(current, expected, precision_rounding=rounding)
|
||
|
|
if float_is_zero(current, precision_rounding=rounding) or compare == 0:
|
||
|
|
return False
|
||
|
|
elif compare > 0:
|
||
|
|
return 'danger'
|
||
|
|
else:
|
||
|
|
return 'success'
|
||
|
|
|
||
|
|
def _get_operations_data(self, production, level=0, current_index=False):
|
||
|
|
if production.state == "done":
|
||
|
|
return self._get_finished_operation_data(production, level, current_index)
|
||
|
|
currency = (production.company_id or self.env.company).currency_id
|
||
|
|
operation_uom = _("Minutes")
|
||
|
|
operations = []
|
||
|
|
total_expected_time = 0.0
|
||
|
|
total_current_time = 0.0
|
||
|
|
total_expected_cost = 0.0
|
||
|
|
total_real_cost = 0.0
|
||
|
|
for index, workorder in enumerate(production.workorder_ids):
|
||
|
|
wo_duration = workorder.get_duration()
|
||
|
|
expected_cost = workorder._compute_expected_operation_cost()
|
||
|
|
current_cost = workorder._compute_current_operation_cost()
|
||
|
|
mo_cost = expected_cost if production.state != 'done' else current_cost
|
||
|
|
is_workorder_started = not float_is_zero(wo_duration, precision_digits=2)
|
||
|
|
if is_workorder_started:
|
||
|
|
real_cost = current_cost
|
||
|
|
elif workorder.operation_id:
|
||
|
|
operation = workorder.operation_id
|
||
|
|
capacity = operation.workcenter_id._get_capacity(production.product_id)
|
||
|
|
operation_cycle = float_round(production.product_uom_qty / capacity, precision_rounding=1, rounding_method='UP')
|
||
|
|
bom_duration_expected = (operation_cycle * operation.time_cycle * 100.0 / operation.workcenter_id.time_efficiency) + \
|
||
|
|
operation.workcenter_id._get_expected_duration(production.product_id)
|
||
|
|
real_cost = expected_cost / (workorder.duration_expected or 1) * bom_duration_expected
|
||
|
|
else:
|
||
|
|
real_cost = expected_cost
|
||
|
|
|
||
|
|
operations.append({
|
||
|
|
'level': level,
|
||
|
|
'index': f"{current_index}W{index}",
|
||
|
|
'model': workorder._name,
|
||
|
|
'id': workorder.id,
|
||
|
|
'name': workorder.name,
|
||
|
|
'state': workorder.state,
|
||
|
|
'formatted_state': self._format_state(workorder),
|
||
|
|
'quantity': workorder.duration_expected if float_is_zero(wo_duration, precision_digits=2) else wo_duration,
|
||
|
|
'quantity_decorator': self._get_comparison_decorator(workorder.duration_expected, wo_duration, 0.01),
|
||
|
|
'uom_name': operation_uom,
|
||
|
|
'production_id': production.id,
|
||
|
|
'unit_cost': expected_cost / (workorder.duration_expected or 1),
|
||
|
|
'mo_cost': mo_cost,
|
||
|
|
'real_cost': real_cost,
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
})
|
||
|
|
total_expected_time += workorder.duration_expected
|
||
|
|
total_current_time += wo_duration if is_workorder_started else workorder.duration_expected
|
||
|
|
total_expected_cost += mo_cost
|
||
|
|
total_real_cost += real_cost
|
||
|
|
|
||
|
|
return {
|
||
|
|
'summary': {
|
||
|
|
'index': f"{current_index}W",
|
||
|
|
'quantity': total_current_time,
|
||
|
|
'quantity_decorator': self._get_comparison_decorator(total_expected_time, total_current_time, 0.01),
|
||
|
|
'mo_cost': total_expected_cost,
|
||
|
|
'real_cost': total_real_cost,
|
||
|
|
'uom_name': operation_uom,
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
},
|
||
|
|
'details': operations,
|
||
|
|
}
|
||
|
|
|
||
|
|
def _get_finished_operation_data(self, production, level=0, current_index=False):
|
||
|
|
currency = (production.company_id or self.env.company).currency_id
|
||
|
|
done_operation_uom = _("Hours")
|
||
|
|
operations = []
|
||
|
|
total_duration = total_duration_expected = total_cost = 0
|
||
|
|
for index, workorder in enumerate(production.workorder_ids):
|
||
|
|
hourly_cost = workorder.costs_hour or workorder.workcenter_id.costs_hour
|
||
|
|
duration = workorder.get_duration() / 60
|
||
|
|
total_duration += duration
|
||
|
|
total_duration_expected += workorder.duration_expected
|
||
|
|
operation_cost = duration * hourly_cost
|
||
|
|
total_cost += operation_cost
|
||
|
|
operations.append({
|
||
|
|
'level': level,
|
||
|
|
'index': f"{current_index}W{index}",
|
||
|
|
'name': f"{workorder.workcenter_id.display_name}: {workorder.display_name}",
|
||
|
|
'quantity': duration,
|
||
|
|
'uom_name': done_operation_uom,
|
||
|
|
'uom_precision': 4,
|
||
|
|
'unit_cost': hourly_cost,
|
||
|
|
'mo_cost': currency.round(operation_cost),
|
||
|
|
'real_cost': currency.round(operation_cost),
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
})
|
||
|
|
return {
|
||
|
|
'summary': {
|
||
|
|
'index': f"{current_index}W",
|
||
|
|
'done': True,
|
||
|
|
'quantity': total_duration,
|
||
|
|
'quantity_decorator': self._get_comparison_decorator(total_duration_expected, total_duration, 0.01),
|
||
|
|
'mo_cost': total_cost,
|
||
|
|
'real_cost': total_cost,
|
||
|
|
'uom_name': done_operation_uom,
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
},
|
||
|
|
'details': operations,
|
||
|
|
}
|
||
|
|
|
||
|
|
def _get_byproducts_data(self, production, current_mo_cost, current_real_cost, level=0, current_index=False):
|
||
|
|
currency = (production.company_id or self.env.company).currency_id
|
||
|
|
byproducts = []
|
||
|
|
byproducts_cost_portion = 0
|
||
|
|
total_mo_cost = 0
|
||
|
|
total_real_cost = 0
|
||
|
|
for index, move_bp in enumerate(production.move_byproduct_ids):
|
||
|
|
product = move_bp.product_id
|
||
|
|
cost_share = move_bp.cost_share / 100
|
||
|
|
byproducts_cost_portion += cost_share
|
||
|
|
mo_cost = current_mo_cost * cost_share
|
||
|
|
real_cost = current_real_cost * cost_share
|
||
|
|
total_mo_cost += mo_cost
|
||
|
|
total_real_cost += real_cost
|
||
|
|
byproducts.append({
|
||
|
|
'level': level,
|
||
|
|
'index': f"{current_index}B{index}",
|
||
|
|
'model': product._name,
|
||
|
|
'id': product.id,
|
||
|
|
'name': product.display_name,
|
||
|
|
'quantity': move_bp.product_uom_qty,
|
||
|
|
'uom_name': move_bp.product_uom.display_name,
|
||
|
|
'uom_precision': self._get_uom_precision(move_bp.product_uom.rounding),
|
||
|
|
'unit_cost': self._get_unit_cost(move_bp),
|
||
|
|
'mo_cost': currency.round(mo_cost),
|
||
|
|
'mo_cost_decorator': self._get_comparison_decorator(real_cost, mo_cost, currency.rounding),
|
||
|
|
'real_cost': currency.round(real_cost),
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
})
|
||
|
|
|
||
|
|
return float_round(1 - byproducts_cost_portion, precision_rounding=0.0001), {
|
||
|
|
'summary': {
|
||
|
|
'index': f"{current_index}B",
|
||
|
|
'mo_cost': currency.round(total_mo_cost),
|
||
|
|
'mo_cost_decorator': self._get_comparison_decorator(total_real_cost, total_mo_cost, currency.rounding),
|
||
|
|
'real_cost': currency.round(total_real_cost),
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
},
|
||
|
|
'details': byproducts,
|
||
|
|
}
|
||
|
|
|
||
|
|
def _compute_cost_sums(self, components, operations=False):
|
||
|
|
total_mo_cost = total_real_cost = 0
|
||
|
|
if operations:
|
||
|
|
total_mo_cost = operations.get('summary', {}).get('mo_cost', 0.0)
|
||
|
|
total_real_cost = operations.get('summary', {}).get('real_cost', 0.0)
|
||
|
|
for component in components:
|
||
|
|
total_mo_cost += component.get('summary', {}).get('mo_cost', 0.0)
|
||
|
|
total_real_cost += component.get('summary', {}).get('real_cost', 0.0)
|
||
|
|
return total_mo_cost, total_real_cost
|
||
|
|
|
||
|
|
def _get_components_data(self, production, replenish_data=False, level=0, current_index=False):
|
||
|
|
if not replenish_data:
|
||
|
|
replenish_data = {
|
||
|
|
'products': {},
|
||
|
|
'warehouses': {},
|
||
|
|
'qty_already_reserved': defaultdict(float),
|
||
|
|
'qty_reserved': {},
|
||
|
|
}
|
||
|
|
components = []
|
||
|
|
if production.state == 'done':
|
||
|
|
replenish_data = self._get_replenishment_from_moves(production, replenish_data)
|
||
|
|
else:
|
||
|
|
replenish_data = self._get_replenishments_from_forecast(production, replenish_data)
|
||
|
|
for count, move_raw in enumerate(production.move_raw_ids):
|
||
|
|
if production.state == 'done' and float_is_zero(move_raw.quantity, precision_rounding=move_raw.product_uom.rounding):
|
||
|
|
# If a product wasn't consumed in the MO by the time it is done, no need to display it on the final Overview.
|
||
|
|
continue
|
||
|
|
component_index = f"{current_index}{count}"
|
||
|
|
replenishments = self._get_replenishment_lines(production, move_raw, replenish_data, level, component_index)
|
||
|
|
# If not enough replenishment -> To Order / Might get "non-available" in summary since all component won't be there in time
|
||
|
|
components.append({
|
||
|
|
'summary': self._format_component_move(production, move_raw, replenishments, replenish_data, level, component_index),
|
||
|
|
'replenishments': replenishments
|
||
|
|
})
|
||
|
|
|
||
|
|
return components
|
||
|
|
|
||
|
|
def _format_component_move(self, production, move_raw, replenishments, replenish_data, level, index):
|
||
|
|
currency = (production.company_id or self.env.company).currency_id
|
||
|
|
product = move_raw.product_id
|
||
|
|
quantity = move_raw.product_uom_qty if move_raw.state != 'done' else move_raw.quantity
|
||
|
|
replenish_mo_cost, dummy_real_cost = self._compute_cost_sums(replenishments)
|
||
|
|
replenish_quantity = sum(rep.get('summary', {}).get('quantity', 0.0) for rep in replenishments)
|
||
|
|
missing_quantity = quantity - replenish_quantity
|
||
|
|
missing_quantity_cost = self._get_component_real_cost(move_raw, missing_quantity)
|
||
|
|
component = {
|
||
|
|
'level': level,
|
||
|
|
'index': index,
|
||
|
|
'id': product.id,
|
||
|
|
'model': product._name,
|
||
|
|
'name': product.display_name,
|
||
|
|
'product_model': product._name,
|
||
|
|
'product': product,
|
||
|
|
'product_id': product.id,
|
||
|
|
'quantity': quantity,
|
||
|
|
'uom': move_raw.product_uom,
|
||
|
|
'uom_name': move_raw.product_uom.display_name,
|
||
|
|
'uom_precision': self._get_uom_precision(move_raw.product_uom.rounding),
|
||
|
|
'quantity_free': product.uom_id._compute_quantity(max(product.free_qty, 0), move_raw.product_uom) if product.type == 'product' else False,
|
||
|
|
'quantity_on_hand': product.uom_id._compute_quantity(product.qty_available, move_raw.product_uom) if product.type == 'product' else False,
|
||
|
|
'quantity_reserved': self._get_reserved_qty(move_raw, production.warehouse_id, replenish_data),
|
||
|
|
'receipt': self._check_planned_start(production.date_start, self._get_component_receipt(product, move_raw, production.warehouse_id, replenishments, replenish_data)),
|
||
|
|
'unit_cost': self._get_unit_cost(move_raw),
|
||
|
|
'mo_cost': currency.round(replenish_mo_cost + missing_quantity_cost),
|
||
|
|
'real_cost': currency.round(self._get_component_real_cost(move_raw, quantity)),
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
}
|
||
|
|
component['mo_cost_decorator'] = self._get_comparison_decorator(component['real_cost'], component['mo_cost'], currency.rounding)
|
||
|
|
if product.type != 'product':
|
||
|
|
return component
|
||
|
|
if any(rep.get('summary', {}).get('model') == 'to_order' for rep in replenishments):
|
||
|
|
# Means that there's an extra "To Order" line summing up what's left to order.
|
||
|
|
component['formatted_state'] = _("To Order")
|
||
|
|
component['state'] = 'to_order'
|
||
|
|
|
||
|
|
return component
|
||
|
|
|
||
|
|
def _get_component_real_cost(self, move_raw, quantity):
|
||
|
|
if float_is_zero(quantity, precision_rounding=move_raw.product_uom.rounding):
|
||
|
|
return 0
|
||
|
|
return self._get_unit_cost(move_raw) * quantity
|
||
|
|
|
||
|
|
def _check_planned_start(self, mo_planned_start, receipt):
|
||
|
|
if mo_planned_start and receipt.get('date', False) and receipt['date'] > mo_planned_start:
|
||
|
|
receipt['decorator'] = 'danger'
|
||
|
|
return receipt
|
||
|
|
|
||
|
|
def _get_component_receipt(self, product, move, warehouse, replenishments, replenish_data):
|
||
|
|
def get(replenishment, key, check_in_receipt=False):
|
||
|
|
fetch = replenishment.get('summary', {})
|
||
|
|
if check_in_receipt:
|
||
|
|
fetch = fetch.get('receipt', {})
|
||
|
|
return fetch.get(key, False)
|
||
|
|
|
||
|
|
if any(get(rep, 'type', True) == 'unavailable' for rep in replenishments):
|
||
|
|
return self._format_receipt_date('unavailable')
|
||
|
|
if product.type != 'product' or move.state == 'done':
|
||
|
|
return self._format_receipt_date('available')
|
||
|
|
|
||
|
|
has_to_order_line = any(rep.get('summary', {}).get('model') == 'to_order' for rep in replenishments)
|
||
|
|
reserved_quantity = self._get_reserved_qty(move, warehouse, replenish_data)
|
||
|
|
missing_quantity = move.product_uom_qty - reserved_quantity
|
||
|
|
free_qty = product.uom_id._compute_quantity(product.free_qty, move.product_uom)
|
||
|
|
if float_compare(missing_quantity, 0.0, precision_rounding=move.product_uom.rounding) <= 0 \
|
||
|
|
or (not has_to_order_line
|
||
|
|
and float_compare(missing_quantity, free_qty, precision_rounding=move.product_uom.rounding) <= 0):
|
||
|
|
return self._format_receipt_date('available')
|
||
|
|
|
||
|
|
replenishments_with_date = list(filter(lambda r: r.get('summary', {}).get('receipt', {}).get('date'), replenishments))
|
||
|
|
max_date = max([get(rep, 'date', True) for rep in replenishments_with_date], default=fields.datetime.today())
|
||
|
|
if has_to_order_line or any(get(rep, 'type', True) == 'estimated' for rep in replenishments):
|
||
|
|
return self._format_receipt_date('estimated', max_date)
|
||
|
|
else:
|
||
|
|
return self._format_receipt_date('expected', max_date)
|
||
|
|
|
||
|
|
def _get_replenishment_lines(self, production, move_raw, replenish_data, level, current_index):
|
||
|
|
product = move_raw.product_id
|
||
|
|
quantity = move_raw.product_uom_qty if move_raw.state != 'done' else move_raw.quantity
|
||
|
|
currency = (production.company_id or self.env.company).currency_id
|
||
|
|
forecast = replenish_data['products'][product.id].get('forecast', [])
|
||
|
|
current_lines = filter(lambda line: line.get('document_in', False) and line.get('document_out', False)
|
||
|
|
and line['document_out'].get('id', False) == production.id and not line.get('already_used'), forecast)
|
||
|
|
total_ordered = 0
|
||
|
|
replenishments = []
|
||
|
|
for count, forecast_line in enumerate(current_lines):
|
||
|
|
if float_compare(total_ordered, quantity, precision_rounding=move_raw.product_uom.rounding) == 0:
|
||
|
|
# If a same product is used twice in the same MO, don't duplicate the replenishment lines
|
||
|
|
break
|
||
|
|
doc_in = self.env[forecast_line['document_in']['_name']].browse(forecast_line['document_in']['id'])
|
||
|
|
replenishment_index = f"{current_index}{count}"
|
||
|
|
replenishment = {}
|
||
|
|
forecast_uom_id = forecast_line['uom_id']
|
||
|
|
line_quantity = min(quantity, forecast_uom_id._compute_quantity(forecast_line['quantity'], move_raw.product_uom)) # Avoid over-rounding
|
||
|
|
replenishment['summary'] = {
|
||
|
|
'level': level + 1,
|
||
|
|
'index': replenishment_index,
|
||
|
|
'id': doc_in.id,
|
||
|
|
'model': doc_in._name,
|
||
|
|
'name': doc_in.display_name,
|
||
|
|
'product_model': product._name,
|
||
|
|
'product_id': product.id,
|
||
|
|
'state': doc_in.state,
|
||
|
|
'quantity': line_quantity,
|
||
|
|
'uom_name': move_raw.product_uom.display_name,
|
||
|
|
'uom_precision': self._get_uom_precision(forecast_line['uom_id']['rounding']),
|
||
|
|
'unit_cost': self._get_unit_cost(move_raw),
|
||
|
|
'mo_cost': forecast_line.get('cost', self._get_replenishment_mo_cost(product, line_quantity, move_raw.product_uom, currency, forecast_line.get('move_in'))),
|
||
|
|
'real_cost': currency.round(self._get_component_real_cost(move_raw, line_quantity)),
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
}
|
||
|
|
if doc_in._name == 'mrp.production':
|
||
|
|
replenishment['components'] = self._get_components_data(doc_in, replenish_data, level + 2, replenishment_index)
|
||
|
|
replenishment['operations'] = self._get_operations_data(doc_in, level + 2, replenishment_index)
|
||
|
|
initial_mo_cost, initial_real_cost = self._compute_cost_sums(replenishment['components'], replenishment['operations'])
|
||
|
|
remaining_cost_share, byproducts = self._get_byproducts_data(doc_in, initial_mo_cost, initial_real_cost, level + 2, replenishment_index)
|
||
|
|
replenishment['byproducts'] = byproducts
|
||
|
|
replenishment['summary']['real_cost'] = initial_real_cost * remaining_cost_share
|
||
|
|
replenishment['summary']['mo_cost'] = initial_mo_cost * remaining_cost_share
|
||
|
|
if self._is_doc_in_done(doc_in):
|
||
|
|
replenishment['summary']['receipt'] = self._format_receipt_date('available')
|
||
|
|
else:
|
||
|
|
replenishment['summary']['receipt'] = self._check_planned_start(production.date_start, self._get_replenishment_receipt(doc_in, replenishment.get('components', [])))
|
||
|
|
replenishment['summary']['mo_cost_decorator'] = self._get_comparison_decorator(replenishment['summary']['real_cost'], replenishment['summary']['mo_cost'], replenishment['summary']['currency'].rounding)
|
||
|
|
replenishment['summary']['formatted_state'] = self._format_state(doc_in, replenishment['components']) if doc_in._name == 'mrp.production' else self._format_state(doc_in)
|
||
|
|
replenishments.append(replenishment)
|
||
|
|
forecast_line['already_used'] = True
|
||
|
|
total_ordered += replenishment['summary']['quantity']
|
||
|
|
|
||
|
|
# Add "In transit" line if necessary
|
||
|
|
in_transit_line = self._add_transit_line(move_raw, forecast, production, level, current_index)
|
||
|
|
if in_transit_line:
|
||
|
|
total_ordered += in_transit_line['summary']['quantity']
|
||
|
|
replenishments.append(in_transit_line)
|
||
|
|
|
||
|
|
reserved_quantity = self._get_reserved_qty(move_raw, production.warehouse_id, replenish_data)
|
||
|
|
# Avoid creating a "to_order" line to compensate for missing stock (i.e. negative free_qty).
|
||
|
|
free_qty = max(0, product.uom_id._compute_quantity(product.free_qty, move_raw.product_uom))
|
||
|
|
missing_quantity = quantity - (reserved_quantity + free_qty + total_ordered)
|
||
|
|
if product.type == 'product' and production.state not in ('done', 'cancel')\
|
||
|
|
and float_compare(missing_quantity, 0, precision_rounding=move_raw.product_uom.rounding) > 0:
|
||
|
|
# Need to order more products to fulfill the need
|
||
|
|
resupply_rules = self._get_resupply_rules(production, product, replenish_data)
|
||
|
|
rules_delay = sum(rule.delay for rule in resupply_rules)
|
||
|
|
resupply_data = self._get_resupply_data(resupply_rules, rules_delay, missing_quantity, move_raw.product_uom, product, production)
|
||
|
|
|
||
|
|
to_order_line = {'summary': {
|
||
|
|
'level': level + 1,
|
||
|
|
'index': f"{current_index}TO",
|
||
|
|
'name': _("To Order"),
|
||
|
|
'model': "to_order",
|
||
|
|
'product_model': product._name,
|
||
|
|
'product_id': product.id,
|
||
|
|
'quantity': missing_quantity,
|
||
|
|
'replenish_quantity': move_raw.product_uom._compute_quantity(missing_quantity, product.uom_id),
|
||
|
|
'uom_name': move_raw.product_uom.display_name,
|
||
|
|
'uom_precision': self._get_uom_precision(move_raw.product_uom.rounding),
|
||
|
|
'real_cost': currency.round(product.standard_price * move_raw.product_uom._compute_quantity(missing_quantity, product.uom_id)),
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
}}
|
||
|
|
if resupply_data:
|
||
|
|
mo_cost = resupply_data['currency']._convert(resupply_data['cost'], currency, (production.company_id or self.env.company), fields.Date.today())
|
||
|
|
to_order_line['summary']['mo_cost'] = mo_cost
|
||
|
|
to_order_line['summary']['receipt'] = self._check_planned_start(production.date_start, self._format_receipt_date('estimated', fields.datetime.today() + timedelta(days=resupply_data['delay'])))
|
||
|
|
else:
|
||
|
|
to_order_line['summary']['mo_cost'] = currency.round(product.standard_price * move_raw.product_uom._compute_quantity(missing_quantity, product.uom_id))
|
||
|
|
to_order_line['summary']['receipt'] = self._format_receipt_date('unavailable')
|
||
|
|
to_order_line['summary']['unit_cost'] = currency.round(to_order_line['summary']['mo_cost'] / missing_quantity)
|
||
|
|
to_order_line['summary']['mo_cost_decorator'] = self._get_comparison_decorator(to_order_line['summary']['real_cost'], to_order_line['summary']['mo_cost'], currency.rounding)
|
||
|
|
replenishments.append(to_order_line)
|
||
|
|
|
||
|
|
return replenishments
|
||
|
|
|
||
|
|
def _add_transit_line(self, move_raw, forecast, production, level, current_index):
|
||
|
|
def is_related_to_production(document, production):
|
||
|
|
if not document:
|
||
|
|
return False
|
||
|
|
return document.get('_name') == production._name and document.get('id') == production.id
|
||
|
|
|
||
|
|
in_transit = next(filter(lambda line: line.get('in_transit') and is_related_to_production(line.get('document_out'), production), forecast), None)
|
||
|
|
if not in_transit or is_related_to_production(in_transit.get('reservation'), production):
|
||
|
|
return None
|
||
|
|
|
||
|
|
product = move_raw.product_id
|
||
|
|
currency = (production.company_id or self.env.company).currency_id
|
||
|
|
lg = self.env['res.lang']._lang_get(self.env.user.lang) or get_lang(self.env)
|
||
|
|
receipt_date = datetime.strptime(in_transit['delivery_date'], lg.date_format)
|
||
|
|
mo_cost = self._get_replenishment_mo_cost(product, in_transit['quantity'], in_transit['uom_id'], currency)
|
||
|
|
real_cost = product.standard_price * in_transit['uom_id']._compute_quantity(in_transit['quantity'], product.uom_id)
|
||
|
|
return {'summary': {
|
||
|
|
'level': level + 1,
|
||
|
|
'index': f"{current_index}IT",
|
||
|
|
'name': _("In Transit"),
|
||
|
|
'model': "in_transit",
|
||
|
|
'product_model': product._name,
|
||
|
|
'product_id': product.id,
|
||
|
|
'quantity': min(move_raw.product_uom_qty, in_transit['uom_id']._compute_quantity(in_transit['quantity'], move_raw.product_uom)), # Avoid over-rounding
|
||
|
|
'uom_name': move_raw.product_uom.display_name,
|
||
|
|
'uom_precision': self._get_uom_precision(move_raw.product_uom.rounding),
|
||
|
|
'mo_cost': mo_cost,
|
||
|
|
'mo_cost_decorator': self._get_comparison_decorator(real_cost, mo_cost, currency.rounding),
|
||
|
|
'real_cost': currency.round(real_cost),
|
||
|
|
'receipt': self._check_planned_start(production.date_start, self._format_receipt_date('expected', receipt_date)),
|
||
|
|
'currency_id': currency.id,
|
||
|
|
'currency': currency,
|
||
|
|
}}
|
||
|
|
|
||
|
|
def _get_replenishment_mo_cost(self, product, quantity, uom_id, currency, move_in=False):
|
||
|
|
return currency.round(product.standard_price * uom_id._compute_quantity(quantity, product.uom_id))
|
||
|
|
|
||
|
|
def _is_doc_in_done(self, doc_in):
|
||
|
|
if doc_in._name == 'mrp.production':
|
||
|
|
return doc_in.state == 'done'
|
||
|
|
return False
|
||
|
|
|
||
|
|
def _get_replenishment_receipt(self, doc_in, components):
|
||
|
|
if doc_in._name == 'stock.picking':
|
||
|
|
return self._format_receipt_date('expected', doc_in.scheduled_date)
|
||
|
|
|
||
|
|
if doc_in._name == 'mrp.production':
|
||
|
|
max_date_start = doc_in.date_start
|
||
|
|
all_available = True
|
||
|
|
some_unavailable = False
|
||
|
|
some_estimated = False
|
||
|
|
for component in components:
|
||
|
|
if component['summary']['receipt']['date']:
|
||
|
|
max_date_start = max(max_date_start, component['summary']['receipt']['date'])
|
||
|
|
all_available = all_available and component['summary']['receipt']['type'] == 'available'
|
||
|
|
some_unavailable = some_unavailable or component['summary']['receipt']['type'] == 'unavailable'
|
||
|
|
some_estimated = some_estimated or component['summary']['receipt']['type'] == 'estimated'
|
||
|
|
|
||
|
|
if some_unavailable:
|
||
|
|
return self._format_receipt_date('unavailable')
|
||
|
|
if all_available:
|
||
|
|
return self._format_receipt_date('expected', doc_in.date_finished)
|
||
|
|
|
||
|
|
new_date = max_date_start + timedelta(days=doc_in.bom_id.produce_delay)
|
||
|
|
receipt_state = 'estimated' if some_estimated else 'expected'
|
||
|
|
return self._format_receipt_date(receipt_state, new_date)
|
||
|
|
return self._format_receipt_date('unavailable')
|
||
|
|
|
||
|
|
def _format_receipt_date(self, state, date=False):
|
||
|
|
if state == 'available':
|
||
|
|
return {'display': _("Available"), 'type': 'available', 'decorator': 'success', 'date': False}
|
||
|
|
elif state == 'estimated':
|
||
|
|
return {'display': _("Estimated %s", format_date(self.env, date)), 'type': 'estimated', 'decorator': False, 'date': date}
|
||
|
|
elif state == 'expected':
|
||
|
|
return {'display': _("Expected %s", format_date(self.env, date)), 'type': 'expected', 'decorator': 'warning', 'date': date}
|
||
|
|
else:
|
||
|
|
return {'display': _("Not Available"), 'type': 'unavailable', 'decorator': 'danger', 'date': False}
|
||
|
|
|
||
|
|
def _get_replenishments_from_forecast(self, production, replenish_data):
|
||
|
|
products = production.move_raw_ids.product_id
|
||
|
|
unknown_products = products.filtered(lambda product: product.id not in replenish_data.get('products', {}))
|
||
|
|
if unknown_products:
|
||
|
|
warehouse = production.warehouse_id
|
||
|
|
wh_location_ids = self._get_warehouse_locations(warehouse, replenish_data)
|
||
|
|
forecast_lines = self.env['stock.forecasted_product_product']._get_report_lines(False, unknown_products.ids, wh_location_ids, warehouse.lot_stock_id, read=False)
|
||
|
|
forecast_lines = self._add_origins_to_forecast(forecast_lines)
|
||
|
|
for product in unknown_products:
|
||
|
|
extra_docs = self._get_extra_replenishments(product)
|
||
|
|
# Sorting the extra documents so that the ones flagged with an explicit production_id are on top of the list.
|
||
|
|
extra_docs.sort(key=lambda ex: ex.get('production_id', False), reverse=True)
|
||
|
|
product_forecast_lines = list(filter(lambda line: line.get('product', {}).get('id') == product.id, forecast_lines))
|
||
|
|
updated_forecast_lines = self._add_extra_in_forecast(product_forecast_lines, extra_docs, product.uom_id.rounding)
|
||
|
|
replenish_data = self._set_replenish_data(updated_forecast_lines, product, replenish_data)
|
||
|
|
|
||
|
|
return replenish_data
|
||
|
|
|
||
|
|
def _get_replenishment_from_moves(self, production, replenish_data):
|
||
|
|
# Go through the component's move see if we can find an incoming origin
|
||
|
|
for component_move in production.move_raw_ids:
|
||
|
|
product_lines = []
|
||
|
|
product = component_move.product_id
|
||
|
|
required_qty = component_move.product_uom_qty
|
||
|
|
for move_origin in self.env['stock.move'].browse(component_move._rollup_move_origs()):
|
||
|
|
doc_origin = self._get_origin(move_origin)
|
||
|
|
if doc_origin:
|
||
|
|
to_uom_qty = move_origin.product_uom._compute_quantity(move_origin.product_uom_qty, component_move.product_uom)
|
||
|
|
used_qty = min(required_qty, to_uom_qty)
|
||
|
|
required_qty -= used_qty
|
||
|
|
# Create a fake "forecast line" so it will be processed as normal afterwards with only the required info
|
||
|
|
product_lines.append({
|
||
|
|
'document_in': {'_name': doc_origin._name, 'id': doc_origin.id},
|
||
|
|
'document_out': {'_name': 'mrp.production', 'id': production.id},
|
||
|
|
'quantity': used_qty,
|
||
|
|
'uom_id': component_move.product_uom,
|
||
|
|
'move_in': move_origin,
|
||
|
|
'product': product,
|
||
|
|
})
|
||
|
|
if float_compare(required_qty, 0, precision_rounding=component_move.product_uom.rounding) <= 0:
|
||
|
|
break
|
||
|
|
replenish_data = self._set_replenish_data(product_lines, product, replenish_data)
|
||
|
|
|
||
|
|
return replenish_data
|
||
|
|
|
||
|
|
def _set_replenish_data(self, new_lines, product, replenish_data):
|
||
|
|
if product.id not in replenish_data['products']:
|
||
|
|
replenish_data['products'][product.id] = {'forecast': []}
|
||
|
|
replenish_data['products'][product.id]['forecast'] += new_lines
|
||
|
|
return replenish_data
|
||
|
|
|
||
|
|
def _get_resupply_rules(self, production, product, replenish_data):
|
||
|
|
if not replenish_data['products'][product.id].get('resupply_rules'):
|
||
|
|
replenish_data['products'][product.id]['resupply_rules'] = product._get_rules_from_location(production.warehouse_id.lot_stock_id)
|
||
|
|
return replenish_data['products'][product.id]['resupply_rules']
|
||
|
|
|
||
|
|
def _add_origins_to_forecast(self, forecast_lines):
|
||
|
|
# Keeps the link to its origin even when the product is now in stock.
|
||
|
|
new_lines = []
|
||
|
|
for line in filter(lambda line: not line.get('document_in', False) and line.get('move_out', False), forecast_lines):
|
||
|
|
move_out_qty = line['move_out'].product_uom._compute_quantity(line['move_out'].product_uom_qty, line['uom_id'])
|
||
|
|
for move_origin in self.env['stock.move'].browse(line['move_out']._rollup_move_origs()):
|
||
|
|
doc_origin = self._get_origin(move_origin)
|
||
|
|
if doc_origin:
|
||
|
|
# Remove 'in_transit' for MTO replenishments
|
||
|
|
line['in_transit'] = False
|
||
|
|
move_origin_qty = move_origin.product_uom._compute_quantity(move_origin.product_uom_qty, line['uom_id'])
|
||
|
|
# Move quantity matches forecast, can add origin to the line
|
||
|
|
if float_compare(line['quantity'], move_origin_qty, precision_rounding=line['uom_id'].rounding) == 0:
|
||
|
|
line['document_in'] = {'_name': doc_origin._name, 'id': doc_origin.id}
|
||
|
|
line['move_in'] = move_origin
|
||
|
|
break
|
||
|
|
|
||
|
|
# Quantity doesn't match, either multiple origins for a single line or multiple lines for a single origin
|
||
|
|
used_quantity = min(move_out_qty, move_origin_qty)
|
||
|
|
new_line = copy.copy(line)
|
||
|
|
new_line['quantity'] = used_quantity
|
||
|
|
new_line['document_in'] = {'_name': doc_origin._name, 'id': doc_origin.id}
|
||
|
|
new_line['move_in'] = move_origin
|
||
|
|
new_lines.append(new_line)
|
||
|
|
# Remove used quantity from original forecast line
|
||
|
|
line['quantity'] -= used_quantity
|
||
|
|
|
||
|
|
move_out_qty -= used_quantity
|
||
|
|
if float_compare(move_out_qty, 0, line['move_out'].product_uom.rounding) <= 0:
|
||
|
|
break
|
||
|
|
return new_lines + forecast_lines
|
||
|
|
|
||
|
|
def _get_origin(self, move):
|
||
|
|
if move.production_id:
|
||
|
|
return move.production_id
|
||
|
|
return False
|
||
|
|
|
||
|
|
def _add_extra_in_forecast(self, forecast_lines, extras, product_rounding):
|
||
|
|
if not extras:
|
||
|
|
return forecast_lines
|
||
|
|
|
||
|
|
lines_with_extras = []
|
||
|
|
for forecast_line in forecast_lines:
|
||
|
|
if forecast_line.get('document_in', False) or forecast_line.get('replenishment_filled'):
|
||
|
|
lines_with_extras.append(forecast_line)
|
||
|
|
continue
|
||
|
|
line_qty = forecast_line['quantity']
|
||
|
|
if forecast_line.get('document_out', False) and forecast_line['document_out']['_name'] == 'mrp.production':
|
||
|
|
production_id = forecast_line['document_out']['id']
|
||
|
|
else:
|
||
|
|
production_id = False
|
||
|
|
index_to_remove = []
|
||
|
|
for index, extra in enumerate(extras):
|
||
|
|
if float_is_zero(extra['quantity'], precision_rounding=product_rounding):
|
||
|
|
index_to_remove.append(index)
|
||
|
|
continue
|
||
|
|
if production_id and extra.get('production_id', False) and extra['production_id'] != production_id:
|
||
|
|
continue
|
||
|
|
if 'init_quantity' not in extra:
|
||
|
|
extra['init_quantity'] = extra['quantity']
|
||
|
|
converted_qty = extra['uom']._compute_quantity(extra['quantity'], forecast_line['uom_id'])
|
||
|
|
taken_from_extra = min(line_qty, converted_qty)
|
||
|
|
ratio = taken_from_extra / extra['uom']._compute_quantity(extra['init_quantity'], forecast_line['uom_id'])
|
||
|
|
line_qty -= taken_from_extra
|
||
|
|
# Create copy of the current forecast line to add a possible replenishment.
|
||
|
|
# Needs to be a copy since it might take multiple replenishment to fulfill a single "out" line.
|
||
|
|
new_extra_line = copy.copy(forecast_line)
|
||
|
|
new_extra_line['quantity'] = taken_from_extra
|
||
|
|
new_extra_line['document_in'] = {
|
||
|
|
'_name': extra['_name'],
|
||
|
|
'id': extra['id'],
|
||
|
|
}
|
||
|
|
new_extra_line['cost'] = extra['cost'] * ratio
|
||
|
|
lines_with_extras.append(new_extra_line)
|
||
|
|
extra['quantity'] -= forecast_line['uom_id']._compute_quantity(taken_from_extra, extra['uom'])
|
||
|
|
if float_compare(extra['quantity'], 0, precision_rounding=product_rounding) <= 0:
|
||
|
|
index_to_remove.append(index)
|
||
|
|
if float_is_zero(line_qty, precision_rounding=product_rounding):
|
||
|
|
break
|
||
|
|
|
||
|
|
for index in reversed(index_to_remove):
|
||
|
|
del extras[index]
|
||
|
|
|
||
|
|
return lines_with_extras
|
||
|
|
|
||
|
|
def _get_extra_replenishments(self, product):
|
||
|
|
return []
|
||
|
|
|
||
|
|
def _get_resupply_data(self, rules, rules_delay, quantity, uom_id, product, production):
|
||
|
|
manufacture_rules = [rule for rule in rules if rule.action == 'manufacture']
|
||
|
|
if manufacture_rules:
|
||
|
|
# Need to get rules from Production location to get delays before production
|
||
|
|
wh_manufacture_rules = product._get_rules_from_location(product.property_stock_production, route_ids=production.warehouse_id.route_ids)
|
||
|
|
wh_manufacture_rules -= rules
|
||
|
|
rules_delay += sum(rule.delay for rule in wh_manufacture_rules)
|
||
|
|
related_bom = self.env['mrp.bom']._bom_find(product)[product]
|
||
|
|
if not related_bom:
|
||
|
|
return False
|
||
|
|
return {
|
||
|
|
'delay': related_bom.produce_delay + rules_delay,
|
||
|
|
'cost': product.standard_price * uom_id._compute_quantity(quantity, product.uom_id),
|
||
|
|
'currency': (production.company_id or self.env.company).currency_id,
|
||
|
|
}
|
||
|
|
return False
|
||
|
|
|
||
|
|
def _get_warehouse_locations(self, warehouse, replenish_data):
|
||
|
|
if not replenish_data['warehouses'].get(warehouse.id):
|
||
|
|
replenish_data['warehouses'][warehouse.id] = [loc['id'] for loc in self.env['stock.location'].search_read(
|
||
|
|
[('id', 'child_of', warehouse.view_location_id.id)],
|
||
|
|
['id']
|
||
|
|
)]
|
||
|
|
return replenish_data['warehouses'][warehouse.id]
|
||
|
|
|
||
|
|
def _get_reserved_qty(self, move_raw, warehouse, replenish_data):
|
||
|
|
if not replenish_data['qty_reserved'].get(move_raw):
|
||
|
|
total_reserved = 0
|
||
|
|
wh_location_ids = self._get_warehouse_locations(warehouse, replenish_data)
|
||
|
|
linked_moves = self.env['stock.move'].browse(move_raw._rollup_move_origs()).filtered(lambda m: m.location_id.id in wh_location_ids)
|
||
|
|
for move in linked_moves:
|
||
|
|
if move.state not in ('partially_available', 'assigned'):
|
||
|
|
continue
|
||
|
|
# count reserved stock in move_raw's uom
|
||
|
|
reserved = move.product_uom._compute_quantity(move.quantity, move_raw.product_uom)
|
||
|
|
# check if the move reserved qty was counted before (happens if multiple outs share pick/pack)
|
||
|
|
reserved = min(reserved - move.product_uom._compute_quantity(replenish_data['qty_already_reserved'][move], move_raw.product_uom), move_raw.product_uom_qty)
|
||
|
|
total_reserved += reserved
|
||
|
|
replenish_data['qty_already_reserved'][move] += move_raw.product_uom._compute_quantity(reserved, move.product_uom)
|
||
|
|
if float_compare(total_reserved, move_raw.product_qty, precision_rounding=move.product_id.uom_id.rounding) >= 0:
|
||
|
|
break
|
||
|
|
replenish_data['qty_reserved'][move_raw] = total_reserved
|
||
|
|
|
||
|
|
return replenish_data['qty_reserved'][move_raw]
|