# 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 lxml import etree 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']) lab_analyses = fields.One2Many( 'coffee.lab.analysis', 'sample', 'Lab analyses') lab_analysis_status = fields.Function(fields.Selection([ (None, ''), ('none', 'No lab analysis'), ('pending', 'Pending'), ('received', 'Received'), ('accepted', 'Accepted'), ('rejected', 'Rejected'), ], 'Lab analysis'), 'get_lab_analysis_status') 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') cupping_average_score = fields.Function( fields.Numeric('Cupping score', digits=(5, 2)), 'get_cupping_average_score') quality_status = fields.Function(fields.Selection([ (None, ''), ('pending', 'Pending'), ('pass', 'Pass'), ('warning', 'Warning'), ('fail', 'Fail'), ], 'Quality status'), 'get_quality_status') recommended_decision = fields.Function(fields.Selection([ (None, ''), ('pending', 'Pending'), ('approve', 'Approve'), ('re_cup', 'Re-cup'), ('reject', 'Reject'), ], 'Recommended decision'), 'get_recommended_decision') 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'), ('received', 'approved'), ('received', 'rejected'), ('sent_to_lab', 'approved'), ('sent_to_lab', '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').in_([ 'received', 'sent_to_lab', 'under_review']), 'depends': ['state'], }, 'evaluate_quality': { 'invisible': ~Eval('state').in_([ 'received', 'sent_to_lab', '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 _coffee_line(self): return self.purchase_line or self.sale_line def _latest_lab_analysis(self): analyses = [ analysis for analysis in self.lab_analyses or [] if analysis.state not in {'cancelled'}] if not analyses: return None return sorted( analyses, key=lambda analysis: ( analysis.received_date or analysis.sent_date or analysis.requested_date or datetime.date.min, getattr(analysis, 'id', 0) or 0), reverse=True)[0] def get_lab_analysis_status(self, name=None): analysis = self._latest_lab_analysis() if not analysis: return 'none' if analysis.state in {'accepted', 'rejected'}: return analysis.state if analysis.state == 'received': return 'received' return 'pending' def _cupping_scores(self): scores = [] for line in self.cupping_lines or []: score = line.get_average_score() if score is not None: scores.append(score) return scores def get_cupping_average_score(self, name=None): scores = self._cupping_scores() if not scores: return None return sum(scores, Decimal(0)) / Decimal(len(scores)) def _quality_values(self): analysis = self._latest_lab_analysis() return { 'moisture': ( getattr(analysis, 'moisture', None) if analysis and analysis.moisture is not None else self.moisture), 'defect_count': ( getattr(analysis, 'defect_count', None) if analysis and analysis.defect_count is not None else self.defect_count), 'cupping_score': ( getattr(analysis, 'cup_score', None) if analysis and analysis.cup_score is not None else self.get_cupping_average_score()), } def get_quality_status(self, name=None): line = self._coffee_line() values = self._quality_values() checks = [] moisture_max = getattr(line, 'coffee_moisture_max', None) if moisture_max is not None and values['moisture'] is not None: checks.append(values['moisture'] <= moisture_max) defect_max = getattr(line, 'coffee_defect_max', None) if defect_max is not None and values['defect_count'] is not None: checks.append(values['defect_count'] <= defect_max) score_min = getattr(line, 'coffee_cup_score_min', None) if score_min is not None and values['cupping_score'] is not None: checks.append(values['cupping_score'] >= score_min) rejected_cupping = any( cupping.decision == 'rejected' for cupping in self.cupping_lines or []) recup_cupping = any( cupping.decision == 're_cup' for cupping in self.cupping_lines or []) if rejected_cupping: return 'fail' if recup_cupping: return 'warning' if not checks: return 'pending' if all(checks): return 'pass' if any(check is False for check in checks): return 'fail' return 'warning' def get_recommended_decision(self, name=None): status = self.get_quality_status() if status == 'pass': return 'approve' if status == 'fail': return 'reject' if status == 'warning': return 're_cup' return 'pending' 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() LabAnalysis = Pool().get('coffee.lab.analysis') cls.write(samples, { 'delegated_to_lab': True, 'sent_date': today, }) values = [] for sample in samples: if not sample.lab: continue active_analyses = [ analysis for analysis in sample.lab_analyses or [] if analysis.state in {'draft', 'sent', 'received'}] if active_analyses: continue values.append({ 'sample': sample.id, 'lab': sample.lab.id, 'requested_date': today, 'due_date': sample.lab_due_date, }) if values: LabAnalysis.create(values) @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 def evaluate_quality(cls, samples): today = Pool().get('ir.date').today() to_approve = [] to_reject = [] to_review = [] for sample in samples: decision = sample.get_recommended_decision() if decision == 'approve': to_approve.append(sample) elif decision == 'reject': to_reject.append(sample) else: to_review.append(sample) if to_approve: cls.write(to_approve, { 'state': 'approved', 'decision_date': today, }) if to_reject: cls.write(to_reject, { 'state': 'rejected', 'decision_date': today, }) if to_review: cls.write(to_review, {'state': 'under_review'}) @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 CoffeeLabAnalysis(Workflow, ModelSQL, ModelView): 'Coffee Lab Analysis' __name__ = 'coffee.lab.analysis' sample = fields.Many2One( 'coffee.sample', 'Sample', required=True, ondelete='CASCADE') lab = fields.Many2One('party.party', 'Lab', required=True) state = fields.Selection([ ('draft', 'Draft'), ('sent', 'Sent'), ('received', 'Received'), ('accepted', 'Accepted'), ('rejected', 'Rejected'), ('cancelled', 'Cancelled'), ], 'State', readonly=True, required=True) requested_date = fields.Date('Requested Date') sent_date = fields.Date('Sent Date') due_date = fields.Date('Due Date') received_date = fields.Date('Received Date') report_reference = fields.Char('Report reference') moisture = fields.Numeric('Moisture %', digits=(5, 2)) screen_size = fields.Char('Screen size') defect_count = fields.Integer('Defects count') cup_score = fields.Numeric('Cup score', digits=(5, 2)) compliant = fields.Function( fields.Boolean('Compliant'), 'get_compliant') result_summary = fields.Text('Result summary') @classmethod def __setup__(cls): super().__setup__() cls._transitions |= set(( ('draft', 'sent'), ('sent', 'received'), ('received', 'accepted'), ('received', 'rejected'), ('draft', 'cancelled'), ('sent', 'cancelled'), ('received', 'cancelled'), ('cancelled', 'draft'), )) cls._buttons.update({ 'send': { 'invisible': Eval('state') != 'draft', 'depends': ['state'], }, 'receive': { 'invisible': Eval('state') != 'sent', 'depends': ['state'], }, 'accept': { 'invisible': Eval('state') != 'received', 'depends': ['state'], }, 'reject': { 'invisible': Eval('state') != 'received', 'depends': ['state'], }, 'cancel': { 'invisible': Eval('state').in_([ 'accepted', 'rejected', 'cancelled']), 'depends': ['state'], }, 'draft': { 'invisible': Eval('state') != 'cancelled', 'depends': ['state'], }, }) @staticmethod def default_state(): return 'draft' @staticmethod def default_requested_date(): return Pool().get('ir.date').today() def get_compliant(self, name=None): if self.state == 'rejected': return False if self.state == 'accepted': return True sample = self.sample if not sample: return False line = sample._coffee_line() checks = [] moisture_max = getattr(line, 'coffee_moisture_max', None) if moisture_max is not None and self.moisture is not None: checks.append(self.moisture <= moisture_max) defect_max = getattr(line, 'coffee_defect_max', None) if defect_max is not None and self.defect_count is not None: checks.append(self.defect_count <= defect_max) score_min = getattr(line, 'coffee_cup_score_min', None) if score_min is not None and self.cup_score is not None: checks.append(self.cup_score >= score_min) return bool(checks and all(checks)) @classmethod @ModelView.button @Workflow.transition('sent') def send(cls, analyses): today = Pool().get('ir.date').today() cls.write([a for a in analyses if not a.sent_date], { 'sent_date': today, }) samples = [analysis.sample for analysis in analyses if analysis.sample] if samples: Pool().get('coffee.sample').write(samples, { 'delegated_to_lab': True, 'sent_date': today, }) @classmethod @ModelView.button @Workflow.transition('received') def receive(cls, analyses): today = Pool().get('ir.date').today() cls.write([a for a in analyses if not a.received_date], { 'received_date': today, }) samples = [analysis.sample for analysis in analyses if analysis.sample] if samples: Pool().get('coffee.sample').write(samples, { 'review_date': today, 'state': 'under_review', }) @classmethod @ModelView.button @Workflow.transition('accepted') def accept(cls, analyses): pass @classmethod @ModelView.button @Workflow.transition('rejected') def reject(cls, analyses): pass @classmethod @ModelView.button @Workflow.transition('cancelled') def cancel(cls, analyses): pass @classmethod @ModelView.button @Workflow.transition('draft') def draft(cls, analyses): pass class CoffeeLineMixin: _coffee_view_fields = { 'coffee_quality_status', 'coffee_origin', 'coffee_type', 'coffee_process', 'coffee_variety', 'coffee_crop_year', 'coffee_screen_size', 'coffee_moisture_max', 'coffee_defect_max', 'coffee_cup_score_min', 'coffee_samples', } active_coffee_compatibility = fields.Function( fields.Boolean('Active coffee compatibility'), 'get_active_coffee_compatibility') coffee_quality_status = fields.Function(fields.Selection([ (None, ''), ('none', 'No sample'), ('pending', 'Pending'), ('approved', 'Approved'), ('rejected', 'Rejected'), ('expired', 'Expired'), ('warning', 'Warning'), ], '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 get_active_coffee_compatibility(self, name=None): return coffee_compatibility_active() def on_change_with_active_coffee_compatibility(self, name=None): return self.get_active_coffee_compatibility(name) @classmethod def fields_view_get(cls, view_id=None, view_type='form', level=None): result = super().fields_view_get( view_id=view_id, view_type=view_type, level=level) if view_type == 'form' and not coffee_compatibility_active(): cls._remove_coffee_quality_page(result) return result @classmethod def _remove_coffee_quality_page(cls, view): parser = etree.XMLParser(remove_blank_text=True, resolve_entities=False) tree = etree.fromstring(view['arch'], parser) for page in tree.xpath('//page[@id="coffee_quality"]'): parent = page.getparent() if parent is not None: parent.remove(page) view['arch'] = etree.tostring(tree, encoding='unicode') for field_name in cls._coffee_view_fields: view.get('fields', {}).pop(field_name, None) 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' if any(sample.get_quality_status() == 'warning' for sample in samples): return 'warning' 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')