# This file is part of Tryton. The COPYRIGHT file at the top level of # this repository contains the full copyright notices and license terms. import datetime from decimal import Decimal from trytond.model import ModelSQL, ModelView, Workflow, fields from trytond.pool import Pool, PoolMeta from trytond.pyson import Eval from trytond.exceptions import UserError COFFEE_SAMPLE_TYPES = [ (None, ''), ('offer', 'Offer sample'), ('pre_shipment', 'Pre-shipment sample'), ('shipment', 'Shipment sample'), ('arrival', 'Arrival sample'), ('stock', 'Stock sample'), ('customer', 'Customer sample'), ('retained', 'Retained sample'), ] COFFEE_PROCESS_TYPES = [ (None, ''), ('washed', 'Washed'), ('natural', 'Natural'), ('honey', 'Honey'), ('semi_washed', 'Semi-washed'), ('other', 'Other'), ] COFFEE_TYPES = [ (None, ''), ('arabica', 'Arabica'), ('robusta', 'Robusta'), ('blend', 'Blend'), ('other', 'Other'), ] def coffee_compatibility_active(): try: Configuration = Pool().get('purchase_trade.configuration') configurations = Configuration.search([], limit=1) except Exception: return False return bool( configurations and configurations[0].active_coffee_compatibility) class CoffeeSample(Workflow, ModelSQL, ModelView): 'Coffee Sample' __name__ = 'coffee.sample' reference = fields.Char('Reference') direction = fields.Selection([ (None, ''), ('purchase', 'Purchase'), ('sale', 'Sale'), ], 'Direction', required=True) sample_type = fields.Selection( COFFEE_SAMPLE_TYPES, 'Sample Type', required=True) state = fields.Selection([ (None, ''), ('requested', 'Requested'), ('received', 'Received'), ('sent_to_lab', 'Sent to lab'), ('under_review', 'Under review'), ('approved', 'Approved'), ('rejected', 'Rejected'), ('expired', 'Expired'), ('archived', 'Archived'), ], 'State', readonly=True, required=True) purchase_line = fields.Many2One( 'purchase.line', 'Purchase Line', ondelete='CASCADE', states={'invisible': Eval('direction') != 'purchase'}, depends=['direction']) sale_line = fields.Many2One( 'sale.line', 'Sale Line', ondelete='CASCADE', states={'invisible': Eval('direction') != 'sale'}, depends=['direction']) lot = fields.Many2One('lot.lot', 'Lot') product = fields.Function( fields.Many2One('product.product', 'Product'), 'get_product') party = fields.Function( fields.Many2One('party.party', 'Party'), 'get_party') public_origin = fields.Function( fields.Char('Origin'), 'get_public_origin') public_coffee_type = fields.Function( fields.Selection(COFFEE_TYPES, 'Coffee type'), 'get_public_coffee_type') requested_date = fields.Date('Requested Date') received_date = fields.Date('Received Date') sent_date = fields.Date('Sent Date') review_date = fields.Date('Review Date') decision_date = fields.Date('Decision Date') shelf_life_days = fields.Integer('Shelf life (days)') expiry_date = fields.Date('Expiry Date') is_expired = fields.Function( fields.Boolean('Expired'), 'get_is_expired', searcher='search_is_expired') expires_soon = fields.Function( fields.Boolean('Expires soon'), 'get_expires_soon', searcher='search_expires_soon') quantity = fields.Numeric('Quantity', digits='unit') unit = fields.Many2One('product.uom', 'Unit') storage_location = fields.Char('Sample location') delegated_to_lab = fields.Boolean('Delegated to lab') lab = fields.Many2One( 'party.party', 'Lab', states={'invisible': ~Eval('delegated_to_lab')}, depends=['delegated_to_lab']) lab_due_date = fields.Date( 'Lab due date', states={'invisible': ~Eval('delegated_to_lab')}, depends=['delegated_to_lab']) moisture = fields.Numeric('Moisture %', digits=(5, 2)) screen_size = fields.Char('Screen size') defect_count = fields.Integer('Defects count') cupping_lines = fields.One2Many( 'coffee.cupping.session.sample', 'sample', 'Cupping sessions') notes = fields.Text('Notes') decision_note = fields.Text('Decision note') @classmethod def __setup__(cls): super().__setup__() cls._transitions |= set(( ('requested', 'received'), ('received', 'sent_to_lab'), ('received', 'under_review'), ('sent_to_lab', 'under_review'), ('under_review', 'approved'), ('under_review', 'rejected'), ('requested', 'expired'), ('received', 'expired'), ('sent_to_lab', 'expired'), ('under_review', 'expired'), ('approved', 'archived'), ('rejected', 'archived'), ('expired', 'archived'), ('archived', 'requested'), )) cls._buttons.update({ 'receive': { 'invisible': Eval('state') != 'requested', 'depends': ['state'], }, 'send_to_lab': { 'invisible': ~Eval('state').in_(['received']), 'depends': ['state'], }, 'review': { 'invisible': ~Eval('state').in_(['received', 'sent_to_lab']), 'depends': ['state'], }, 'approve': { 'invisible': Eval('state') != 'under_review', 'depends': ['state'], }, 'reject': { 'invisible': Eval('state') != 'under_review', 'depends': ['state'], }, 'expire': { 'invisible': Eval('state').in_( ['approved', 'rejected', 'expired', 'archived']), 'depends': ['state'], }, 'archive': { 'invisible': ~Eval('state').in_( ['approved', 'rejected', 'expired']), 'depends': ['state'], }, 'reset_to_requested': { 'invisible': Eval('state') != 'archived', 'depends': ['state'], }, }) @staticmethod def default_state(): return 'requested' @staticmethod def default_shelf_life_days(): return 90 @staticmethod def default_requested_date(): return Pool().get('ir.date').today() @classmethod def default_direction(cls): return 'purchase' @classmethod def create(cls, vlist): for values in vlist: cls._set_direction_from_line(values) cls._set_default_expiry_date(values, force=True) return super().create(vlist) @classmethod def write(cls, *args): actions = iter(args) args = [] for records, values in zip(actions, actions): values = values.copy() cls._set_direction_from_line(values) if ( {'requested_date', 'received_date', 'shelf_life_days'} & set(values) and not any( getattr(record, 'expiry_date', None) for record in records)): cls._set_default_expiry_date(values) args.extend((records, values)) super().write(*args) @classmethod def _set_direction_from_line(cls, values): if values.get('purchase_line') and not values.get('direction'): values['direction'] = 'purchase' elif values.get('sale_line') and not values.get('direction'): values['direction'] = 'sale' @classmethod def _set_default_expiry_date(cls, values, force=False): if values.get('expiry_date'): return if not force and not ( {'requested_date', 'received_date', 'shelf_life_days'} & set(values)): return start = ( values.get('received_date') or values.get('requested_date') or cls.default_requested_date()) shelf_life_days = values.get('shelf_life_days') if shelf_life_days is None: shelf_life_days = cls.default_shelf_life_days() if start and shelf_life_days: values['expiry_date'] = ( start + datetime.timedelta(days=int(shelf_life_days))) def get_product(self, name=None): line = self.purchase_line or self.sale_line product = getattr(line, 'product', None) return product.id if product else None def get_party(self, name=None): if self.purchase_line and self.purchase_line.purchase: party = self.purchase_line.purchase.party return party.id if party else None if self.sale_line and self.sale_line.sale: party = self.sale_line.sale.party return party.id if party else None def get_public_origin(self, name=None): line = self.purchase_line or self.sale_line return getattr(line, 'coffee_origin', None) def get_public_coffee_type(self, name=None): line = self.purchase_line or self.sale_line return getattr(line, 'coffee_type', None) def get_is_expired(self, name=None): if self.state == 'expired': return True today = Pool().get('ir.date').today() return bool( self.expiry_date and self.expiry_date < today and self.state not in {'archived', 'rejected'}) def get_expires_soon(self, name=None): today = Pool().get('ir.date').today() limit = today + datetime.timedelta(days=14) return bool( self.expiry_date and today <= self.expiry_date <= limit and self.state not in {'archived', 'expired', 'rejected'}) @classmethod def search_is_expired(cls, name, clause): today = Pool().get('ir.date').today() positive = clause[2] domain = ['OR', ('state', '=', 'expired'), [ ('expiry_date', '<', today), ('state', 'not in', ['archived', 'rejected']), ], ] if positive: return domain return ['NOT', domain] @classmethod def search_expires_soon(cls, name, clause): today = Pool().get('ir.date').today() limit = today + datetime.timedelta(days=14) domain = [ ('expiry_date', '>=', today), ('expiry_date', '<=', limit), ('state', 'not in', ['archived', 'expired', 'rejected']), ] if clause[2]: return domain return ['NOT', domain] def get_rec_name(self, name): if self.reference: return self.reference label = self.sample_type or 'sample' if self.purchase_line and self.purchase_line.purchase: return '%s - %s' % (label, self.purchase_line.purchase.rec_name) if self.sale_line and self.sale_line.sale: return '%s - %s' % (label, self.sale_line.sale.rec_name) return label @classmethod @ModelView.button @Workflow.transition('received') def receive(cls, samples): today = Pool().get('ir.date').today() cls.write( [sample for sample in samples if not sample.received_date], {'received_date': today}) @classmethod @ModelView.button @Workflow.transition('sent_to_lab') def send_to_lab(cls, samples): today = Pool().get('ir.date').today() cls.write(samples, { 'delegated_to_lab': True, 'sent_date': today, }) @classmethod @ModelView.button @Workflow.transition('under_review') def review(cls, samples): today = Pool().get('ir.date').today() cls.write(samples, {'review_date': today}) @classmethod @ModelView.button @Workflow.transition('approved') def approve(cls, samples): today = Pool().get('ir.date').today() cls.write(samples, {'decision_date': today}) @classmethod @ModelView.button @Workflow.transition('rejected') def reject(cls, samples): today = Pool().get('ir.date').today() cls.write(samples, {'decision_date': today}) @classmethod @ModelView.button @Workflow.transition('expired') def expire(cls, samples): pass @classmethod @ModelView.button @Workflow.transition('archived') def archive(cls, samples): pass @classmethod @ModelView.button @Workflow.transition('requested') def reset_to_requested(cls, samples): pass class CoffeeCuppingCriterion(ModelSQL, ModelView): 'Coffee Cupping Criterion' __name__ = 'coffee.cupping.criterion' name = fields.Char('Name', required=True) sequence = fields.Integer('Sequence') active = fields.Boolean('Active') kind = fields.Selection([ ('score', 'Score'), ('defect', 'Defect'), ('text', 'Text'), ], 'Kind', required=True) max_score = fields.Numeric('Max score', digits=(5, 2)) subtract_from_score = fields.Boolean('Subtract from score') description = fields.Text('Description') @staticmethod def default_sequence(): return 10 @staticmethod def default_active(): return True @staticmethod def default_kind(): return 'score' @staticmethod def default_max_score(): return Decimal('10.00') class CoffeeCuppingSession(Workflow, ModelSQL, ModelView): 'Coffee Cupping Session' __name__ = 'coffee.cupping.session' name = fields.Char('Name', required=True) date = fields.Date('Date') responsible = fields.Many2One('party.party', 'Responsible') location = fields.Char('Location') cup_capacity = fields.Integer('Cup capacity') cups_per_sample = fields.Integer('Cups per sample') state = fields.Selection([ ('draft', 'Draft'), ('prepared', 'Prepared'), ('in_progress', 'In progress'), ('done', 'Done'), ('cancelled', 'Cancelled'), ], 'State', readonly=True, required=True) lines = fields.One2Many( 'coffee.cupping.session.sample', 'session', 'Samples') total_cups = fields.Function( fields.Integer('Total cups'), 'get_total_cups') remaining_cups = fields.Function( fields.Integer('Remaining cups'), 'get_remaining_cups') @classmethod def __setup__(cls): super().__setup__() cls._transitions |= set(( ('draft', 'prepared'), ('prepared', 'in_progress'), ('in_progress', 'done'), ('draft', 'cancelled'), ('prepared', 'cancelled'), ('in_progress', 'cancelled'), ('cancelled', 'draft'), )) cls._buttons.update({ 'prepare': { 'invisible': Eval('state') != 'draft', 'depends': ['state'], }, 'start': { 'invisible': Eval('state') != 'prepared', 'depends': ['state'], }, 'done': { 'invisible': Eval('state') != 'in_progress', 'depends': ['state'], }, 'cancel': { 'invisible': Eval('state').in_(['done', 'cancelled']), 'depends': ['state'], }, 'draft': { 'invisible': Eval('state') != 'cancelled', 'depends': ['state'], }, }) @staticmethod def default_state(): return 'draft' @staticmethod def default_date(): return Pool().get('ir.date').today() @staticmethod def default_cup_capacity(): return 24 @staticmethod def default_cups_per_sample(): return 3 @classmethod def _total_requested_cups(cls, lines): total = 0 for line in lines: session = getattr(line, 'session', None) total += ( line.cups_per_sample or getattr(session, 'cups_per_sample', 0) or 0) return total def get_total_cups(self, name=None): return self._total_requested_cups(self.lines or []) def get_remaining_cups(self, name=None): capacity = self.cup_capacity or 0 return capacity - self.get_total_cups() @staticmethod def _coffee_type_order(coffee_type): return { 'arabica': 10, 'blend': 20, 'other': 30, 'robusta': 40, }.get(coffee_type, 30) @classmethod def _line_sort_key(cls, line): sample = getattr(line, 'sample', None) coffee_type = sample.get_public_coffee_type() if sample else None origin = sample.get_public_origin() if sample else '' return ( cls._coffee_type_order(coffee_type), origin or '', getattr(sample, 'reference', '') or '', getattr(sample, 'id', 0) or 0, ) @classmethod @ModelView.button @Workflow.transition('prepared') def prepare(cls, sessions): Cup = Pool().get('coffee.cupping.cup') Line = Pool().get('coffee.cupping.session.sample') for session in sessions: lines = list(session.lines or []) total_cups = cls._total_requested_cups(lines) if session.cup_capacity and total_cups > session.cup_capacity: raise UserError( 'The cupping session exceeds the room cup capacity.') for sequence, line in enumerate( sorted(lines, key=cls._line_sort_key), 1): if line.sequence != sequence: Line.write([line], {'sequence': sequence}) existing_cups = list(line.cups or []) missing = (line.cups_per_sample or session.cups_per_sample or 0) missing -= len(existing_cups) if missing <= 0: continue values = [] start = len(existing_cups) + 1 for cup_no in range(start, start + missing): values.append({ 'session_sample': line.id, 'cup_no': cup_no, 'blind_code': 'S%02d-C%d' % (sequence, cup_no), }) Cup.create(values) @classmethod @ModelView.button @Workflow.transition('in_progress') def start(cls, sessions): pass @classmethod @ModelView.button @Workflow.transition('done') def done(cls, sessions): pass @classmethod @ModelView.button @Workflow.transition('cancelled') def cancel(cls, sessions): pass @classmethod @ModelView.button @Workflow.transition('draft') def draft(cls, sessions): pass class CoffeeCuppingSessionSample(ModelSQL, ModelView): 'Coffee Cupping Session Sample' __name__ = 'coffee.cupping.session.sample' session = fields.Many2One( 'coffee.cupping.session', 'Session', required=True, ondelete='CASCADE') sample = fields.Many2One( 'coffee.sample', 'Sample', required=True, ondelete='CASCADE') sequence = fields.Integer('Order') cups_per_sample = fields.Integer('Cups') decision = fields.Selection([ ('pending', 'Pending'), ('approved', 'Approved'), ('re_cup', 'Re-cup'), ('rejected', 'Rejected'), ], 'Decision', required=True) decision_note = fields.Text('Decision note') public_origin = fields.Function( fields.Char('Origin'), 'get_public_origin') public_coffee_type = fields.Function( fields.Selection(COFFEE_TYPES, 'Coffee type'), 'get_public_coffee_type') average_score = fields.Function( fields.Numeric('Average score', digits=(5, 2)), 'get_average_score') defect_count = fields.Function( fields.Integer('Defect reports'), 'get_defect_count') result_count = fields.Function( fields.Integer('Results'), 'get_result_count') cups = fields.One2Many( 'coffee.cupping.cup', 'session_sample', 'Cups') @staticmethod def default_decision(): return 'pending' def get_public_origin(self, name=None): return self.sample.get_public_origin() if self.sample else None def get_public_coffee_type(self, name=None): return self.sample.get_public_coffee_type() if self.sample else None def _results(self): results = [] for cup in self.cups or []: results.extend(cup.results or []) return results def get_average_score(self, name=None): scores = [ result.get_total_score() for result in self._results() if result.get_total_score() is not None] if not scores: return None return sum(scores, Decimal(0)) / Decimal(len(scores)) def get_defect_count(self, name=None): return sum(1 for result in self._results() if result.has_defect) def get_result_count(self, name=None): return len(self._results()) class CoffeeCuppingCup(ModelSQL, ModelView): 'Coffee Cupping Cup' __name__ = 'coffee.cupping.cup' session_sample = fields.Many2One( 'coffee.cupping.session.sample', 'Session sample', required=True, ondelete='CASCADE') cup_no = fields.Integer('Cup') blind_code = fields.Char('Blind code', required=True) public_origin = fields.Function( fields.Char('Origin'), 'get_public_origin') public_coffee_type = fields.Function( fields.Selection(COFFEE_TYPES, 'Coffee type'), 'get_public_coffee_type') average_score = fields.Function( fields.Numeric('Average score', digits=(5, 2)), 'get_average_score') defect_count = fields.Function( fields.Integer('Defect reports'), 'get_defect_count') results = fields.One2Many( 'coffee.cupping.result', 'cup', 'Results') def get_public_origin(self, name=None): if self.session_sample: return self.session_sample.get_public_origin() def get_public_coffee_type(self, name=None): if self.session_sample: return self.session_sample.get_public_coffee_type() def get_average_score(self, name=None): scores = [ result.get_total_score() for result in self.results or [] if result.get_total_score() is not None] if not scores: return None return sum(scores, Decimal(0)) / Decimal(len(scores)) def get_defect_count(self, name=None): return sum(1 for result in self.results or [] if result.has_defect) class CoffeeCuppingResult(ModelSQL, ModelView): 'Coffee Cupping Result' __name__ = 'coffee.cupping.result' cup = fields.Many2One( 'coffee.cupping.cup', 'Cup', required=True, ondelete='CASCADE') cupper = fields.Many2One('party.party', 'Cupper', required=True) date = fields.Date('Date') has_defect = fields.Boolean('Defect found') total_score = fields.Function( fields.Numeric('Total score', digits=(5, 2)), 'get_total_score') lines = fields.One2Many( 'coffee.cupping.result.line', 'result', 'Criteria') notes = fields.Text('Notes') @staticmethod def default_date(): return Pool().get('ir.date').today() def get_total_score(self, name=None): total = Decimal(0) has_score = False for line in self.lines or []: score = line.score if score is None: continue has_score = True if line.criterion and line.criterion.subtract_from_score: total -= Decimal(str(score)) else: total += Decimal(str(score)) return total if has_score else None class CoffeeCuppingResultLine(ModelSQL, ModelView): 'Coffee Cupping Result Line' __name__ = 'coffee.cupping.result.line' result = fields.Many2One( 'coffee.cupping.result', 'Result', required=True, ondelete='CASCADE') criterion = fields.Many2One( 'coffee.cupping.criterion', 'Criterion', required=True) score = fields.Numeric('Score', digits=(5, 2)) defect_found = fields.Boolean('Defect found') notes = fields.Text('Notes') class CoffeeLineMixin: active_coffee_compatibility = fields.Function( fields.Boolean('Active coffee compatibility'), 'on_change_with_active_coffee_compatibility') coffee_samples = fields.One2Many( 'coffee.sample', None, 'Coffee samples') coffee_quality_status = fields.Function(fields.Selection([ (None, ''), ('none', 'No sample'), ('pending', 'Pending'), ('approved', 'Approved'), ('rejected', 'Rejected'), ('expired', 'Expired'), ], 'Coffee quality'), 'get_coffee_quality_status') coffee_origin = fields.Char('Coffee origin') coffee_type = fields.Selection(COFFEE_TYPES, 'Coffee type') coffee_process = fields.Selection(COFFEE_PROCESS_TYPES, 'Coffee process') coffee_variety = fields.Char('Coffee variety') coffee_crop_year = fields.Char('Crop year') coffee_screen_size = fields.Char('Screen size') coffee_moisture_max = fields.Numeric('Moisture max %', digits=(5, 2)) coffee_defect_max = fields.Integer('Defects max') coffee_cup_score_min = fields.Numeric( 'Cup score min', digits=(5, 2)) def _coffee_samples(self): return list(getattr(self, 'coffee_samples', None) or []) def on_change_with_active_coffee_compatibility(self, name=None): return coffee_compatibility_active() def get_coffee_quality_status(self, name=None): samples = self._coffee_samples() if not samples: return 'none' if any(sample.state == 'rejected' for sample in samples): return 'rejected' if any(sample.state == 'expired' or sample.get_is_expired() for sample in samples): return 'expired' if any(sample.state == 'approved' for sample in samples): return 'approved' return 'pending' class PurchaseLine(CoffeeLineMixin, metaclass=PoolMeta): __name__ = 'purchase.line' coffee_samples = fields.One2Many( 'coffee.sample', 'purchase_line', 'Coffee samples') class SaleLine(CoffeeLineMixin, metaclass=PoolMeta): __name__ = 'sale.line' coffee_samples = fields.One2Many( 'coffee.sample', 'sale_line', 'Coffee samples')