clinical_laboratory/lims_management/models/stock_lot.py

282 lines
9.8 KiB
Python

# -*- coding: utf-8 -*-
from odoo import models, fields, api
from datetime import datetime
import random
class StockLot(models.Model):
_name = 'stock.lot'
_inherit = ['stock.lot', 'mail.thread', 'mail.activity.mixin']
is_lab_sample = fields.Boolean(string='Es Muestra de Laboratorio')
barcode = fields.Char(
string='Código de Barras',
compute='_compute_barcode',
store=True,
readonly=True,
help="Código de barras único para la muestra en formato YYMMDDNNNNNNC"
)
patient_id = fields.Many2one(
'res.partner',
string='Paciente',
domain="[('is_patient', '=', True)]"
)
request_id = fields.Many2one(
'sale.order',
string='Orden de Laboratorio',
domain="[('is_lab_request', '=', True)]"
)
collection_date = fields.Datetime(string='Fecha de Recolección')
container_type = fields.Selection([
('serum_tube', 'Tubo de Suero'),
('edta_tube', 'Tubo EDTA'),
('swab', 'Hisopo'),
('urine', 'Contenedor de Orina'),
('other', 'Otro')
], string='Tipo de Contenedor (Obsoleto)', help='Campo obsoleto, use sample_type_product_id en su lugar')
sample_type_product_id = fields.Many2one(
'product.template',
string='Tipo de Muestra',
domain="[('is_sample_type', '=', True)]",
help="Producto que representa el tipo de contenedor/muestra"
)
collector_id = fields.Many2one(
'res.users',
string='Recolectado por',
default=lambda self: self.env.user
)
doctor_id = fields.Many2one(
'res.partner',
string='Médico Referente',
domain="[('is_doctor', '=', True)]",
help="Médico que ordenó los análisis"
)
origin = fields.Char(
string='Origen',
help="Referencia a la orden de laboratorio que generó esta muestra"
)
volume_ml = fields.Float(
string='Volumen (ml)',
help="Volumen total de muestra requerido"
)
analysis_names = fields.Char(
string='Análisis',
help="Lista de análisis que se realizarán con esta muestra"
)
state = fields.Selection([
('pending_collection', 'Pendiente de Recolección'),
('collected', 'Recolectada'),
('received', 'Recibida en Laboratorio'),
('in_process', 'En Proceso'),
('analyzed', 'Analizada'),
('stored', 'Almacenada'),
('disposed', 'Desechada'),
('cancelled', 'Cancelada')
], string='Estado', default='collected', tracking=True)
def action_collect(self):
"""Mark sample as collected"""
old_state = self.state
self.write({'state': 'collected', 'collection_date': fields.Datetime.now()})
self.message_post(
body='Muestra recolectada por %s' % self.env.user.name,
subject='Estado actualizado: Recolectada',
message_type='notification'
)
def action_receive(self):
"""Mark sample as received in laboratory"""
old_state = self.state
self.write({'state': 'received'})
self.message_post(
body='Muestra recibida en laboratorio por %s' % self.env.user.name,
subject='Estado actualizado: Recibida',
message_type='notification'
)
def action_start_analysis(self):
"""Start analysis process"""
old_state = self.state
self.write({'state': 'in_process'})
self.message_post(
body='Análisis iniciado por %s' % self.env.user.name,
subject='Estado actualizado: En Proceso',
message_type='notification'
)
def action_complete_analysis(self):
"""Mark analysis as completed"""
old_state = self.state
self.write({'state': 'analyzed'})
self.message_post(
body='Análisis completado por %s' % self.env.user.name,
subject='Estado actualizado: Analizada',
message_type='notification'
)
def action_store(self):
"""Store the sample"""
old_state = self.state
self.write({'state': 'stored'})
self.message_post(
body='Muestra almacenada por %s' % self.env.user.name,
subject='Estado actualizado: Almacenada',
message_type='notification'
)
def action_dispose(self):
"""Dispose of the sample"""
old_state = self.state
self.write({'state': 'disposed'})
self.message_post(
body='Muestra desechada por %s. Motivo de disposición registrado.' % self.env.user.name,
subject='Estado actualizado: Desechada',
message_type='notification'
)
def action_cancel(self):
"""Cancel the sample"""
old_state = self.state
self.write({'state': 'cancelled'})
self.message_post(
body='Muestra cancelada por %s' % self.env.user.name,
subject='Estado actualizado: Cancelada',
message_type='notification'
)
@api.onchange('sample_type_product_id')
def _onchange_sample_type_product_id(self):
"""Synchronize container_type when sample_type_product_id changes"""
if self.sample_type_product_id:
# Try to map product name to legacy container type
product_name = self.sample_type_product_id.name.lower()
if 'suero' in product_name or 'serum' in product_name:
self.container_type = 'serum_tube'
elif 'edta' in product_name:
self.container_type = 'edta_tube'
elif 'hisopo' in product_name or 'swab' in product_name:
self.container_type = 'swab'
elif 'orina' in product_name or 'urine' in product_name:
self.container_type = 'urine'
else:
self.container_type = 'other'
def get_container_name(self):
"""Get container name from product or legacy field"""
if self.sample_type_product_id:
return self.sample_type_product_id.name
elif self.container_type:
return dict(self._fields['container_type'].selection).get(self.container_type)
return 'Unknown'
@api.depends('is_lab_sample', 'create_date')
def _compute_barcode(self):
"""Generate unique barcode for laboratory samples"""
for record in self:
if record.is_lab_sample and not record.barcode:
record.barcode = record._generate_unique_barcode()
elif not record.is_lab_sample:
record.barcode = False
def _generate_unique_barcode(self):
"""Generate a unique barcode in format YYMMDDNNNNNNC
YY: Year (2 digits)
MM: Month (2 digits)
DD: Day (2 digits)
NNNNNN: Sequential number (6 digits)
C: Check digit
"""
self.ensure_one()
now = datetime.now()
date_prefix = now.strftime('%y%m%d')
# Get the highest sequence number for today
domain = [
('is_lab_sample', '=', True),
('barcode', 'like', date_prefix + '%'),
('id', '!=', self.id)
]
max_barcode = self.search(domain, order='barcode desc', limit=1)
if max_barcode and max_barcode.barcode:
# Extract sequence number from existing barcode
try:
sequence = int(max_barcode.barcode[6:12]) + 1
except:
sequence = 1
else:
sequence = 1
# Ensure we don't exceed 6 digits
if sequence > 999999:
# Add prefix based on sample type to allow more barcodes
prefix_map = {
'suero': '1',
'edta': '2',
'orina': '3',
'hisopo': '4',
'other': '9'
}
type_prefix = '9' # default
if self.sample_type_product_id:
name_lower = self.sample_type_product_id.name.lower()
for key, val in prefix_map.items():
if key in name_lower:
type_prefix = val
break
sequence = int(type_prefix + str(sequence % 100000).zfill(5))
# Format sequence with leading zeros
sequence_str = str(sequence).zfill(6)
# Calculate check digit using Luhn algorithm
barcode_without_check = date_prefix + sequence_str
check_digit = self._calculate_luhn_check_digit(barcode_without_check)
final_barcode = barcode_without_check + str(check_digit)
# Verify uniqueness
existing = self.search([
('barcode', '=', final_barcode),
('id', '!=', self.id)
], limit=1)
if existing:
# If collision, add random component and retry
sequence = sequence * 10 + random.randint(0, 9)
sequence_str = str(sequence % 1000000).zfill(6)
barcode_without_check = date_prefix + sequence_str
check_digit = self._calculate_luhn_check_digit(barcode_without_check)
final_barcode = barcode_without_check + str(check_digit)
return final_barcode
def _calculate_luhn_check_digit(self, number_str):
"""Calculate Luhn check digit for barcode validation"""
digits = [int(d) for d in number_str]
odd_sum = sum(digits[-1::-2])
even_sum = sum([sum(divmod(2 * d, 10)) for d in digits[-2::-2]])
total = odd_sum + even_sum
return (10 - (total % 10)) % 10
def _ensure_barcode(self):
"""Ensure all lab samples have a barcode"""
for record in self:
if record.is_lab_sample and not record.barcode:
record.barcode = record._generate_unique_barcode()
return True