diff options
Diffstat (limited to 'addons/maintenance/models')
| -rw-r--r-- | addons/maintenance/models/__init__.py | 3 | ||||
| -rw-r--r-- | addons/maintenance/models/maintenance.py | 435 |
2 files changed, 438 insertions, 0 deletions
diff --git a/addons/maintenance/models/__init__.py b/addons/maintenance/models/__init__.py new file mode 100644 index 00000000..fb9363f4 --- /dev/null +++ b/addons/maintenance/models/__init__.py @@ -0,0 +1,3 @@ +# -*- coding: utf-8 -*- + +from . import maintenance diff --git a/addons/maintenance/models/maintenance.py b/addons/maintenance/models/maintenance.py new file mode 100644 index 00000000..fcb87179 --- /dev/null +++ b/addons/maintenance/models/maintenance.py @@ -0,0 +1,435 @@ +# -*- coding: utf-8 -*- + +import ast + +from datetime import date, datetime, timedelta + +from odoo import api, fields, models, SUPERUSER_ID, _ +from odoo.exceptions import UserError +from odoo.tools import DEFAULT_SERVER_DATE_FORMAT, DEFAULT_SERVER_DATETIME_FORMAT + + +class MaintenanceStage(models.Model): + """ Model for case stages. This models the main stages of a Maintenance Request management flow. """ + + _name = 'maintenance.stage' + _description = 'Maintenance Stage' + _order = 'sequence, id' + + name = fields.Char('Name', required=True, translate=True) + sequence = fields.Integer('Sequence', default=20) + fold = fields.Boolean('Folded in Maintenance Pipe') + done = fields.Boolean('Request Done') + + +class MaintenanceEquipmentCategory(models.Model): + _name = 'maintenance.equipment.category' + _inherit = ['mail.alias.mixin', 'mail.thread'] + _description = 'Maintenance Equipment Category' + + @api.depends('equipment_ids') + def _compute_fold(self): + # fix mutual dependency: 'fold' depends on 'equipment_count', which is + # computed with a read_group(), which retrieves 'fold'! + self.fold = False + for category in self: + category.fold = False if category.equipment_count else True + + name = fields.Char('Category Name', required=True, translate=True) + company_id = fields.Many2one('res.company', string='Company', + default=lambda self: self.env.company) + technician_user_id = fields.Many2one('res.users', 'Responsible', tracking=True, default=lambda self: self.env.uid) + color = fields.Integer('Color Index') + note = fields.Text('Comments', translate=True) + equipment_ids = fields.One2many('maintenance.equipment', 'category_id', string='Equipments', copy=False) + equipment_count = fields.Integer(string="Equipment", compute='_compute_equipment_count') + maintenance_ids = fields.One2many('maintenance.request', 'category_id', copy=False) + maintenance_count = fields.Integer(string="Maintenance Count", compute='_compute_maintenance_count') + alias_id = fields.Many2one( + 'mail.alias', 'Alias', ondelete='restrict', required=True, + help="Email alias for this equipment category. New emails will automatically " + "create a new equipment under this category.") + fold = fields.Boolean(string='Folded in Maintenance Pipe', compute='_compute_fold', store=True) + + def _compute_equipment_count(self): + equipment_data = self.env['maintenance.equipment'].read_group([('category_id', 'in', self.ids)], ['category_id'], ['category_id']) + mapped_data = dict([(m['category_id'][0], m['category_id_count']) for m in equipment_data]) + for category in self: + category.equipment_count = mapped_data.get(category.id, 0) + + def _compute_maintenance_count(self): + maintenance_data = self.env['maintenance.request'].read_group([('category_id', 'in', self.ids)], ['category_id'], ['category_id']) + mapped_data = dict([(m['category_id'][0], m['category_id_count']) for m in maintenance_data]) + for category in self: + category.maintenance_count = mapped_data.get(category.id, 0) + + def unlink(self): + for category in self: + if category.equipment_ids or category.maintenance_ids: + raise UserError(_("You cannot delete an equipment category containing equipments or maintenance requests.")) + return super(MaintenanceEquipmentCategory, self).unlink() + + def _alias_get_creation_values(self): + values = super(MaintenanceEquipmentCategory, self)._alias_get_creation_values() + values['alias_model_id'] = self.env['ir.model']._get('maintenance.request').id + if self.id: + values['alias_defaults'] = defaults = ast.literal_eval(self.alias_defaults or "{}") + defaults['category_id'] = self.id + return values + + +class MaintenanceEquipment(models.Model): + _name = 'maintenance.equipment' + _inherit = ['mail.thread', 'mail.activity.mixin'] + _description = 'Maintenance Equipment' + _check_company_auto = True + + def _track_subtype(self, init_values): + self.ensure_one() + if 'owner_user_id' in init_values and self.owner_user_id: + return self.env.ref('maintenance.mt_mat_assign') + return super(MaintenanceEquipment, self)._track_subtype(init_values) + + def name_get(self): + result = [] + for record in self: + if record.name and record.serial_no: + result.append((record.id, record.name + '/' + record.serial_no)) + if record.name and not record.serial_no: + result.append((record.id, record.name)) + return result + + @api.model + def _name_search(self, name, args=None, operator='ilike', limit=100, name_get_uid=None): + args = args or [] + equipment_ids = [] + if name: + equipment_ids = self._search([('name', '=', name)] + args, limit=limit, access_rights_uid=name_get_uid) + if not equipment_ids: + equipment_ids = self._search([('name', operator, name)] + args, limit=limit, access_rights_uid=name_get_uid) + return equipment_ids + + name = fields.Char('Equipment Name', required=True, translate=True) + company_id = fields.Many2one('res.company', string='Company', + default=lambda self: self.env.company) + active = fields.Boolean(default=True) + technician_user_id = fields.Many2one('res.users', string='Technician', tracking=True) + owner_user_id = fields.Many2one('res.users', string='Owner', tracking=True) + category_id = fields.Many2one('maintenance.equipment.category', string='Equipment Category', + tracking=True, group_expand='_read_group_category_ids') + partner_id = fields.Many2one('res.partner', string='Vendor', check_company=True) + partner_ref = fields.Char('Vendor Reference') + location = fields.Char('Location') + model = fields.Char('Model') + serial_no = fields.Char('Serial Number', copy=False) + assign_date = fields.Date('Assigned Date', tracking=True) + effective_date = fields.Date('Effective Date', default=fields.Date.context_today, required=True, help="Date at which the equipment became effective. This date will be used to compute the Mean Time Between Failure.") + cost = fields.Float('Cost') + note = fields.Text('Note') + warranty_date = fields.Date('Warranty Expiration Date') + color = fields.Integer('Color Index') + scrap_date = fields.Date('Scrap Date') + maintenance_ids = fields.One2many('maintenance.request', 'equipment_id') + maintenance_count = fields.Integer(compute='_compute_maintenance_count', string="Maintenance Count", store=True) + maintenance_open_count = fields.Integer(compute='_compute_maintenance_count', string="Current Maintenance", store=True) + period = fields.Integer('Days between each preventive maintenance') + next_action_date = fields.Date(compute='_compute_next_maintenance', string='Date of the next preventive maintenance', store=True) + maintenance_team_id = fields.Many2one('maintenance.team', string='Maintenance Team', check_company=True) + maintenance_duration = fields.Float(help="Maintenance Duration in hours.") + + @api.depends('effective_date', 'period', 'maintenance_ids.request_date', 'maintenance_ids.close_date') + def _compute_next_maintenance(self): + date_now = fields.Date.context_today(self) + equipments = self.filtered(lambda x: x.period > 0) + for equipment in equipments: + next_maintenance_todo = self.env['maintenance.request'].search([ + ('equipment_id', '=', equipment.id), + ('maintenance_type', '=', 'preventive'), + ('stage_id.done', '!=', True), + ('close_date', '=', False)], order="request_date asc", limit=1) + last_maintenance_done = self.env['maintenance.request'].search([ + ('equipment_id', '=', equipment.id), + ('maintenance_type', '=', 'preventive'), + ('stage_id.done', '=', True), + ('close_date', '!=', False)], order="close_date desc", limit=1) + if next_maintenance_todo and last_maintenance_done: + next_date = next_maintenance_todo.request_date + date_gap = next_maintenance_todo.request_date - last_maintenance_done.close_date + # If the gap between the last_maintenance_done and the next_maintenance_todo one is bigger than 2 times the period and next request is in the future + # We use 2 times the period to avoid creation too closed request from a manually one created + if date_gap > timedelta(0) and date_gap > timedelta(days=equipment.period) * 2 and next_maintenance_todo.request_date > date_now: + # If the new date still in the past, we set it for today + if last_maintenance_done.close_date + timedelta(days=equipment.period) < date_now: + next_date = date_now + else: + next_date = last_maintenance_done.close_date + timedelta(days=equipment.period) + elif next_maintenance_todo: + next_date = next_maintenance_todo.request_date + date_gap = next_maintenance_todo.request_date - date_now + # If next maintenance to do is in the future, and in more than 2 times the period, we insert an new request + # We use 2 times the period to avoid creation too closed request from a manually one created + if date_gap > timedelta(0) and date_gap > timedelta(days=equipment.period) * 2: + next_date = date_now + timedelta(days=equipment.period) + elif last_maintenance_done: + next_date = last_maintenance_done.close_date + timedelta(days=equipment.period) + # If when we add the period to the last maintenance done and we still in past, we plan it for today + if next_date < date_now: + next_date = date_now + else: + next_date = equipment.effective_date + timedelta(days=equipment.period) + equipment.next_action_date = next_date + (self - equipments).next_action_date = False + + @api.depends('maintenance_ids.stage_id.done') + def _compute_maintenance_count(self): + for equipment in self: + equipment.maintenance_count = len(equipment.maintenance_ids) + equipment.maintenance_open_count = len(equipment.maintenance_ids.filtered(lambda x: not x.stage_id.done)) + + @api.onchange('company_id') + def _onchange_company_id(self): + if self.company_id and self.maintenance_team_id: + if self.maintenance_team_id.company_id and not self.maintenance_team_id.company_id.id == self.company_id.id: + self.maintenance_team_id = False + + @api.onchange('category_id') + def _onchange_category_id(self): + self.technician_user_id = self.category_id.technician_user_id + + _sql_constraints = [ + ('serial_no', 'unique(serial_no)', "Another asset already exists with this serial number!"), + ] + + @api.model + def create(self, vals): + equipment = super(MaintenanceEquipment, self).create(vals) + if equipment.owner_user_id: + equipment.message_subscribe(partner_ids=[equipment.owner_user_id.partner_id.id]) + return equipment + + def write(self, vals): + if vals.get('owner_user_id'): + self.message_subscribe(partner_ids=self.env['res.users'].browse(vals['owner_user_id']).partner_id.ids) + return super(MaintenanceEquipment, self).write(vals) + + @api.model + def _read_group_category_ids(self, categories, domain, order): + """ Read group customization in order to display all the categories in + the kanban view, even if they are empty. + """ + category_ids = categories._search([], order=order, access_rights_uid=SUPERUSER_ID) + return categories.browse(category_ids) + + def _create_new_request(self, date): + self.ensure_one() + self.env['maintenance.request'].create({ + 'name': _('Preventive Maintenance - %s', self.name), + 'request_date': date, + 'schedule_date': date, + 'category_id': self.category_id.id, + 'equipment_id': self.id, + 'maintenance_type': 'preventive', + 'owner_user_id': self.owner_user_id.id, + 'user_id': self.technician_user_id.id, + 'maintenance_team_id': self.maintenance_team_id.id, + 'duration': self.maintenance_duration, + 'company_id': self.company_id.id or self.env.company.id + }) + + @api.model + def _cron_generate_requests(self): + """ + Generates maintenance request on the next_action_date or today if none exists + """ + for equipment in self.search([('period', '>', 0)]): + next_requests = self.env['maintenance.request'].search([('stage_id.done', '=', False), + ('equipment_id', '=', equipment.id), + ('maintenance_type', '=', 'preventive'), + ('request_date', '=', equipment.next_action_date)]) + if not next_requests: + equipment._create_new_request(equipment.next_action_date) + + +class MaintenanceRequest(models.Model): + _name = 'maintenance.request' + _inherit = ['mail.thread.cc', 'mail.activity.mixin'] + _description = 'Maintenance Request' + _order = "id desc" + _check_company_auto = True + + @api.returns('self') + def _default_stage(self): + return self.env['maintenance.stage'].search([], limit=1) + + def _creation_subtype(self): + return self.env.ref('maintenance.mt_req_created') + + def _track_subtype(self, init_values): + self.ensure_one() + if 'stage_id' in init_values: + return self.env.ref('maintenance.mt_req_status') + return super(MaintenanceRequest, self)._track_subtype(init_values) + + def _get_default_team_id(self): + MT = self.env['maintenance.team'] + team = MT.search([('company_id', '=', self.env.company.id)], limit=1) + if not team: + team = MT.search([], limit=1) + return team.id + + name = fields.Char('Subjects', required=True) + company_id = fields.Many2one('res.company', string='Company', + default=lambda self: self.env.company) + description = fields.Text('Description') + request_date = fields.Date('Request Date', tracking=True, default=fields.Date.context_today, + help="Date requested for the maintenance to happen") + owner_user_id = fields.Many2one('res.users', string='Created by User', default=lambda s: s.env.uid) + category_id = fields.Many2one('maintenance.equipment.category', related='equipment_id.category_id', string='Category', store=True, readonly=True) + equipment_id = fields.Many2one('maintenance.equipment', string='Equipment', + ondelete='restrict', index=True, check_company=True) + user_id = fields.Many2one('res.users', string='Technician', tracking=True) + stage_id = fields.Many2one('maintenance.stage', string='Stage', ondelete='restrict', tracking=True, + group_expand='_read_group_stage_ids', default=_default_stage, copy=False) + priority = fields.Selection([('0', 'Very Low'), ('1', 'Low'), ('2', 'Normal'), ('3', 'High')], string='Priority') + color = fields.Integer('Color Index') + close_date = fields.Date('Close Date', help="Date the maintenance was finished. ") + kanban_state = fields.Selection([('normal', 'In Progress'), ('blocked', 'Blocked'), ('done', 'Ready for next stage')], + string='Kanban State', required=True, default='normal', tracking=True) + # active = fields.Boolean(default=True, help="Set active to false to hide the maintenance request without deleting it.") + archive = fields.Boolean(default=False, help="Set archive to true to hide the maintenance request without deleting it.") + maintenance_type = fields.Selection([('corrective', 'Corrective'), ('preventive', 'Preventive')], string='Maintenance Type', default="corrective") + schedule_date = fields.Datetime('Scheduled Date', help="Date the maintenance team plans the maintenance. It should not differ much from the Request Date. ") + maintenance_team_id = fields.Many2one('maintenance.team', string='Team', required=True, default=_get_default_team_id, check_company=True) + duration = fields.Float(help="Duration in hours.") + done = fields.Boolean(related='stage_id.done') + + def archive_equipment_request(self): + self.write({'archive': True}) + + def reset_equipment_request(self): + """ Reinsert the maintenance request into the maintenance pipe in the first stage""" + first_stage_obj = self.env['maintenance.stage'].search([], order="sequence asc", limit=1) + # self.write({'active': True, 'stage_id': first_stage_obj.id}) + self.write({'archive': False, 'stage_id': first_stage_obj.id}) + + @api.onchange('company_id') + def _onchange_company_id(self): + if self.company_id and self.maintenance_team_id: + if self.maintenance_team_id.company_id and not self.maintenance_team_id.company_id.id == self.company_id.id: + self.maintenance_team_id = False + + @api.onchange('equipment_id') + def onchange_equipment_id(self): + if self.equipment_id: + self.user_id = self.equipment_id.technician_user_id if self.equipment_id.technician_user_id else self.equipment_id.category_id.technician_user_id + self.category_id = self.equipment_id.category_id + if self.equipment_id.maintenance_team_id: + self.maintenance_team_id = self.equipment_id.maintenance_team_id.id + + @api.onchange('category_id') + def onchange_category_id(self): + if not self.user_id or not self.equipment_id or (self.user_id and not self.equipment_id.technician_user_id): + self.user_id = self.category_id.technician_user_id + + @api.model + def create(self, vals): + # context: no_log, because subtype already handle this + request = super(MaintenanceRequest, self).create(vals) + if request.owner_user_id or request.user_id: + request._add_followers() + if request.equipment_id and not request.maintenance_team_id: + request.maintenance_team_id = request.equipment_id.maintenance_team_id + request.activity_update() + return request + + def write(self, vals): + # Overridden to reset the kanban_state to normal whenever + # the stage (stage_id) of the Maintenance Request changes. + if vals and 'kanban_state' not in vals and 'stage_id' in vals: + vals['kanban_state'] = 'normal' + res = super(MaintenanceRequest, self).write(vals) + if vals.get('owner_user_id') or vals.get('user_id'): + self._add_followers() + if 'stage_id' in vals: + self.filtered(lambda m: m.stage_id.done).write({'close_date': fields.Date.today()}) + self.activity_feedback(['maintenance.mail_act_maintenance_request']) + if vals.get('user_id') or vals.get('schedule_date'): + self.activity_update() + if vals.get('equipment_id'): + # need to change description of activity also so unlink old and create new activity + self.activity_unlink(['maintenance.mail_act_maintenance_request']) + self.activity_update() + return res + + def activity_update(self): + """ Update maintenance activities based on current record set state. + It reschedule, unlink or create maintenance request activities. """ + self.filtered(lambda request: not request.schedule_date).activity_unlink(['maintenance.mail_act_maintenance_request']) + for request in self.filtered(lambda request: request.schedule_date): + date_dl = fields.Datetime.from_string(request.schedule_date).date() + updated = request.activity_reschedule( + ['maintenance.mail_act_maintenance_request'], + date_deadline=date_dl, + new_user_id=request.user_id.id or request.owner_user_id.id or self.env.uid) + if not updated: + if request.equipment_id: + note = _('Request planned for <a href="#" data-oe-model="%s" data-oe-id="%s">%s</a>') % ( + request.equipment_id._name, request.equipment_id.id, request.equipment_id.display_name) + else: + note = False + request.activity_schedule( + 'maintenance.mail_act_maintenance_request', + fields.Datetime.from_string(request.schedule_date).date(), + note=note, user_id=request.user_id.id or request.owner_user_id.id or self.env.uid) + + def _add_followers(self): + for request in self: + partner_ids = (request.owner_user_id.partner_id + request.user_id.partner_id).ids + request.message_subscribe(partner_ids=partner_ids) + + @api.model + def _read_group_stage_ids(self, stages, domain, order): + """ Read group customization in order to display all the stages in the + kanban view, even if they are empty + """ + stage_ids = stages._search([], order=order, access_rights_uid=SUPERUSER_ID) + return stages.browse(stage_ids) + + +class MaintenanceTeam(models.Model): + _name = 'maintenance.team' + _description = 'Maintenance Teams' + + name = fields.Char('Team Name', required=True, translate=True) + active = fields.Boolean(default=True) + company_id = fields.Many2one('res.company', string='Company', + default=lambda self: self.env.company) + member_ids = fields.Many2many( + 'res.users', 'maintenance_team_users_rel', string="Team Members", + domain="[('company_ids', 'in', company_id)]") + color = fields.Integer("Color Index", default=0) + request_ids = fields.One2many('maintenance.request', 'maintenance_team_id', copy=False) + equipment_ids = fields.One2many('maintenance.equipment', 'maintenance_team_id', copy=False) + + # For the dashboard only + todo_request_ids = fields.One2many('maintenance.request', string="Requests", copy=False, compute='_compute_todo_requests') + todo_request_count = fields.Integer(string="Number of Requests", compute='_compute_todo_requests') + todo_request_count_date = fields.Integer(string="Number of Requests Scheduled", compute='_compute_todo_requests') + todo_request_count_high_priority = fields.Integer(string="Number of Requests in High Priority", compute='_compute_todo_requests') + todo_request_count_block = fields.Integer(string="Number of Requests Blocked", compute='_compute_todo_requests') + todo_request_count_unscheduled = fields.Integer(string="Number of Requests Unscheduled", compute='_compute_todo_requests') + + @api.depends('request_ids.stage_id.done') + def _compute_todo_requests(self): + for team in self: + team.todo_request_ids = team.request_ids.filtered(lambda e: e.stage_id.done==False) + team.todo_request_count = len(team.todo_request_ids) + team.todo_request_count_date = len(team.todo_request_ids.filtered(lambda e: e.schedule_date != False)) + team.todo_request_count_high_priority = len(team.todo_request_ids.filtered(lambda e: e.priority == '3')) + team.todo_request_count_block = len(team.todo_request_ids.filtered(lambda e: e.kanban_state == 'blocked')) + team.todo_request_count_unscheduled = len(team.todo_request_ids.filtered(lambda e: not e.schedule_date)) + + @api.depends('equipment_ids') + def _compute_equipment(self): + for team in self: + team.equipment_count = len(team.equipment_ids) |