trytond-carrier_load_ul/load.py

# The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
from trytond.model import fields, ModelView, Workflow
from trytond.pool import PoolMeta, Pool
from trytond.pyson import Eval
from trytond.transaction import Transaction
from trytond.wizard import Wizard, StateTransition, StateView, Button

__metaclass__ = PoolMeta

__all__ = ['Configuration', 'Load', 'LoadOrder', 'LoadOrderLine',
           'LoadUnitLoad', 'LoadUnitLoadOrder', 'LoadUnitLoadData']


class Configuration:
    __name__ = 'carrier.configuration'

    ul_origin_restrict = fields.Boolean('Restrict UL origin',
                                        help='Restricts origin of UL when loading in a Load order.')

    @classmethod
    def default_ul_origin_restrict(cls):
        return True


class Load:
    __name__ = 'carrier.load'

    unit_loads = fields.Function(
        fields.One2Many('stock.unit_load', None, 'Unit loads'),
        'get_unit_loads', searcher='search_unit_loads')

    def get_unit_loads(self, name=None):
        if not self.orders:
            return []
        return [ul.id for l in self.orders for ul in l.unit_loads if l.unit_loads]

    @classmethod
    def search_unit_loads(cls, name, clause):
        return [('orders.unit_loads', ) + tuple(clause[1:])]


class LoadOrder:
    __name__ = 'carrier.load.order'

    unit_loads = fields.Function(
        fields.One2Many('stock.unit_load', None, 'Unit loads',
                        states={'readonly': Eval('state').in_(['cancel', 'done'])},
                        depends=['state']),
        'get_unit_loads', setter='set_unit_loads', searcher='search_unit_loads')
    ul_origin_restrict = fields.Boolean('Restrict UL origin',
                                        states={'readonly': Eval('state') != 'draft'},
                                        depends=['state'])

    @classmethod
    def __setup__(cls):
        super(LoadOrder, cls).__setup__()
        cls._buttons.update({
            'run_try': {'icon': 'tryton-go-next',
                        'invisible': ~Eval('state').in_(['waiting', 'running'])},
        })
        cls._error_messages.update({
            'cancel_ul': 'Cannot cancel load orders with loaded ULs',
            'ul_state': 'UL "%s" must be in Done state.',
            'ul_loaded': 'UL "%s" is already loaded.',
            'ul_location': 'UL "%s" must be in a storage location.',
            'ul_warehouse': 'UL "%s" must be in warehouse "%s" to be loaded in order "%s".',
            'ul_origin': 'UL "%s" does not belong to any origin of this load order.',
            'ul_overload': 'All valid lines of load order "%s" are complete. Cannot load more ULs.',
            'ul_data': 'Product data of UL "%s" does not match with any line of load order "%s".'})

    @classmethod
    def default_ul_origin_restrict(cls):
        pool = Pool()
        Configuration = pool.get('carrier.configuration')

        conf = Configuration(1)
        return conf.ul_origin_restrict

    def get_unit_loads(self, name=None):
        if not self.lines:
            return []
        return [ul.id for l in self.lines for ul in l.unit_loads if l.unit_loads]

    @classmethod
    def set_unit_loads(cls, records, name, value):
        pass

    @classmethod
    def search_unit_loads(cls, name, clause):
        return [('lines.unit_loads', ) + tuple(clause[1:])]

    def get_carrier_amount(self, name=None):
        if not self.load.unit_price:
            return 0
        return self.load.currency.round(
            (len(self.unit_loads) / len(self.load.unit_loads)) * self.load.unit_price)

    @classmethod
    def cancel(cls, records):
        if any(r.unit_loads for r in records):
            cls.raise_user_error('cancel_ul')
        super(LoadOrder, cls).cancel(records)

    def _get_load_sale(self, Sale):
        res = super(LoadOrder, self)._get_load_sale(Sale)
        res.shipment_method = 'manual'
        return res

    def _get_load_sale_line(self, key, grouped_items):
        res = super(LoadOrder, self)._get_load_sale_line(key, grouped_items)
        res.ul_quantity = len(grouped_items)
        return res

    def _get_shipment_sale(self, sale):
        res = super(LoadOrder, self)._get_shipment_sale(sale)
        res.start_date = self.start_date
        res.on_change_start_date()
        res.end_date = self.end_date
        return res

    def _get_shipment_moves(self, sale_line, grouped_items):
        pool = Pool()
        Move = pool.get('stock.move')

        moves = []
        other_moves = []
        for item in grouped_items:
            new_moves = item._get_new_moves({'from_location': sale_line.from_location.id,
                                             'to_location': sale_line.to_location.id,
                                             'date_time_': self.end_date})
            move, = [m for m in new_moves if m.product == item.product]
            move.origin = sale_line
            moves.append(move)

            for new_move in new_moves:
                if new_move.product.id == item.product.id:
                    continue
                new_move.origin = item.load_line
                other_moves.append(new_move)

        if other_moves:
            Move.save(other_moves)
        return moves

    def _get_items(self):
        return self.unit_loads

    @classmethod
    def do(cls, records):
        pool = Pool()
        Move = pool.get('stock.move')

        super(LoadOrder, cls).do(records)
        moves = []
        for record in records:
            moves.extend(
                [record._get_inventory_move(m) for m in record.outgoing_moves])
        if moves:
            Move.save(moves)
        Move.do([m for r in records for l in r.lines for m in l.moves])

    def _get_inventory_move(self, move):
        pool = Pool()
        Move = pool.get('stock.move')

        location = move.unit_load.get_location(
            [move.unit_load], product_id=move.product.id, type='storage')[move.unit_load.id]
        return Move(
            from_location=location,
            to_location=move.from_location,
            product=move.product,
            uom=move.uom,
            quantity=move.quantity,
            date_time_=self.start_date,
            company=move.company,
            currency=move.currency,
            unit_price=move.unit_price,
            unit_load=move.unit_load,
            origin=move.origin,
            lot=getattr(move, 'lot', None)
        )

    @classmethod
    @ModelView.button_action('carrier_load_ul.wizard_load_ul')
    def run_try(cls, records):
        pass

    def add_ul(self, unit_loads):
        pool = Pool()
        UL = pool.get('stock.unit_load')

        order_lines = {}
        for unit_load in unit_loads:
            # check state
            if unit_load.state != 'done':
                self.raise_user_error('ul_state', unit_load.rec_name)

            # check it is not loaded yet
            if unit_load.load_line:
                self.raise_user_error('ul_loaded', unit_load.rec_name)
            location = unit_load.location

            # check it is in storage location
            if location.type != 'storage':
                self.raise_user_error('ul_location', unit_load.rec_name)

            # check it is in warehouse
            wh = None
            while not wh:
                location = location.parent
                if not location:
                    break
                if location.type == 'warehouse':
                    wh = location
            if not wh or wh.id != self.load.warehouse.id:
                self.raise_user_error('ul_warehouse', (
                    unit_load.rec_name, self.load.warehouse.rec_name,
                    self.rec_name))

            # check it is linked to origin lines
            lines = [l for l in self.lines if l.origin and unit_load in l.origin.unit_loads]
            if not lines:
                if self.ul_origin_restrict:
                    self.raise_user_error('ul_origin', unit_load.rec_name)
                self.raise_user_warning('loading_ul_origin_%s' % unit_load.id, 'ul_origin',
                                        unit_load.rec_name)
                lines = self.lines

            # check data matches
            lines = self.check_ul_data_match(lines, unit_load)
            if not lines:
                self.raise_user_error('ul_data', (unit_load.rec_name, self.rec_name))

            # check overload line qty
            line = None
            for _line in lines:
                order_lines.setdefault(_line.id, _line.ul_quantity - len(_line.unit_loads))
                if order_lines[_line.id] > 0:
                    line = _line
                    break
            if not line:
                self.raise_user_error('ul_overload', self.rec_name)
            order_lines[line.id] -= 1

            # load UL
            unit_load.load_line = _line

        UL.save(unit_loads)

    def check_ul_data_match(self, lines, unit_load):
        valid_lines = []
        for line in lines:
            product = getattr(line.origin, 'product', None)
            if not line.origin or not product:
                valid_lines.append(line)
                continue
            if product.id == unit_load.product.id:
                valid_lines.append(line)
        return valid_lines


class LoadOrderLine:
    __name__ = 'carrier.load.order.line'

    ul_quantity = fields.Float('ULs', digits=(16, 0))
    quantity_per_ul = fields.Function(
        fields.Float('Quantity per UL', digits=(16, Eval('unit_digits', 0)),
                     depends=['unit_digits']),
        'on_change_with_quantity_per_ul')
    unit_loads = fields.One2Many('stock.unit_load', 'load_line', 'Unit loads',
                                 readonly=True)

    @fields.depends('quantity', 'ul_quantity', 'uom')
    def on_change_with_quantity_per_ul(self, name=None):
        if self.quantity and self.ul_quantity:
            return self.uom.round(self.quantity / self.ul_quantity, self.uom.rounding)
        return None

    @classmethod
    def _get_quantity_field(cls):
        return 'ul_quantity'


class LoadUnitLoadOrder(ModelView):
    """Carrier load unit load"""
    __name__ = 'carrier.load_uls.order'

    load_order = fields.Many2One('carrier.load.order', 'Order',
                                 required=True,
                                 domain=[('state', 'in', ['waiting', 'running'])])


# TODO: domain of uls_to_load: filter by in stock (storage location).
# It is verified later in add_ul, but would be fine to filter here
# TODO: configure ul_code reading by a string pattern (ex: P${code}) in carrier.configuration
# to read it with barcode scanner
class LoadUnitLoadData(ModelView):
    """Carrier load unit load"""
    __name__ = 'carrier.load_uls.data'

    load_order = fields.Many2One('carrier.load.order', 'Order',
                                 readonly=True)
    ul_code = fields.Char('UL')
    uls_to_load = fields.One2Many('stock.unit_load', None, 'ULs to load',
                                  domain=[('state', '=', 'done')])
    uls_loaded = fields.One2Many('stock.unit_load', None, 'Loaded ULs',
                                 domain=[('load_order', '=', Eval('load_order'))],
                                 context={'ul_loading': True},
                                 depends=['load_order'])
    loaded_uls = fields.Float('Loaded ULs', digits=(16, 0),
                              readonly=True)


class LoadUnitLoad(Wizard):
    """Carrier load unit load"""
    __name__ = 'carrier.load_uls'
    
    start = StateTransition()
    order = StateView('carrier.load_uls.order',
                      'carrier_load_ul.load_uls_order_view_form',
                      [Button('Cancel', 'end', 'tryton-cancel'),
                       Button('OK', 'data', 'tryton-ok', default=True)])
    data = StateView('carrier.load_uls.data',
                     'carrier_load_ul.load_uls_data_view_form',
                     [Button('Exit', 'end', 'tryton-cancel'),
                      Button('Unload ULs', 'unload_', 'tryton-clear'),
                      Button('Do', 'do_', 'tryton-ok'),
                      Button('Load', 'load_', 'tryton-list-add', default=True)])
    load_ = StateTransition()
    unload_ = StateTransition()
    do_ = StateTransition()

    @classmethod
    def __setup__(cls):
        super(LoadUnitLoad, cls).__setup__()
        cls._error_messages.update({
            'invalid_ul': 'Cannot find Unit load "%s".',
            'ul_required': 'Must define an UL to load.',
            'unload_any': 'Must select some Unit load from loaded ULs list in order to unload them.'})

    def transition_start(self):
        if Transaction().context.get('active_model') != LoadOrder.__name__:
            return 'order'
        return 'data'

    def default_data(self, fields):
        order = self._get_load_order()
        res = {'load_order': order.id,
               'loaded_uls': 0}
        if order.unit_loads:
            res['loaded_uls'] = len(order.unit_loads)
        return res

    def transition_load_(self):
        pool = Pool()
        UL = pool.get('stock.unit_load')

        if not self.data.ul_code and not self.data.uls_to_load:
            self.raise_user_error('ul_required')

        order = self._get_load_order()
        uls = []
        if self.data.ul_code:
            ul = UL.search([('code', '=', self.data.ul_code)])
            if not ul:
                self.raise_user_error('invalid_ul', self.data.ul_code)
            uls.append(ul[0])
        if self.data.uls_to_load:
            uls.extend(self.data.uls_to_load)
        if uls:
            order.add_ul(uls)
        if order.state != 'running':
            order.run([order])
        return 'data'

    def transition_do_(self):
        pool = Pool()
        Order = pool.get('carrier.load.order')

        Order.do([self._get_load_order()])
        # TODO: print reports
        return 'end'

    def _get_load_order(self):
        pool = Pool()
        Order = pool.get('carrier.load.order')

        if Transaction().context.get('active_model') == LoadOrder.__name__:
            return Order(Transaction().context.get('active_id'))
        return Order(self.order.load_order.id)

    def transition_unload_(self):
        pool = Pool()
        UnitLoad = pool.get('stock.unit_load')

        if not self.data.uls_loaded:
            self.raise_user_error('unload_any')
        UnitLoad.unload(list(self.data.uls_loaded))
        return 'data'