350 lines
14 KiB
Python
350 lines
14 KiB
Python
from flask import Blueprint, request, redirect, url_for, flash, render_template, session, jsonify
|
|
from datetime import datetime
|
|
import re
|
|
from typing import Optional
|
|
from app.models import Calibration, Channel, WorkOrder, Standard, User
|
|
from app.supabase_client import get_supabase
|
|
|
|
calibrations_bp = Blueprint('calibrations', __name__, url_prefix='/calibrations')
|
|
|
|
@calibrations_bp.route('/create', methods=['POST'])
|
|
def create_calibrations():
|
|
# Validate form data
|
|
work_order_id = request.form.get('work_order_id')
|
|
calibrated_by = request.form.get('calibrated_by')
|
|
std_used = request.form.get('std_used')
|
|
date_str = request.form.get('date')
|
|
|
|
if not all([work_order_id, calibrated_by, std_used, date_str]):
|
|
flash('Missing required fields', 'error')
|
|
return redirect(request.referrer)
|
|
|
|
try:
|
|
date = datetime.strptime(str(date_str), '%Y-%m-%d').date()
|
|
except ValueError:
|
|
flash('Invalid date format', 'error')
|
|
return redirect(request.referrer)
|
|
|
|
# Get channel serials and calibration data
|
|
channel_serials = request.form.getlist('channel_serial[]')
|
|
scales = request.form.getlist('scale[]')
|
|
offsets = request.form.getlist('offset[]')
|
|
deviation_highs = request.form.getlist('deviation_high[]')
|
|
deviation_mids = request.form.getlist('deviation_mid[]')
|
|
deviation_lows = request.form.getlist('deviation_low[]')
|
|
set_highs = request.form.getlist('set_high[]')
|
|
set_mids = request.form.getlist('set_mid[]')
|
|
set_lows = request.form.getlist('set_low[]')
|
|
passed_list = request.form.getlist('passed[]')
|
|
|
|
# Validate serial numbers
|
|
serial_pattern = re.compile(r'^[0-9A-F]{16}$')
|
|
for serial in channel_serials:
|
|
if not serial_pattern.match(serial):
|
|
flash(f'Invalid serial number format: {serial}', 'error')
|
|
return redirect(request.referrer)
|
|
|
|
# Get standard calibration dates
|
|
supabase = get_supabase()
|
|
standard = supabase.table('standards').select('calibrated_on, calibration_due').eq('id', std_used).execute()
|
|
if not standard.data:
|
|
flash('Invalid standard selected', 'error')
|
|
return redirect(request.referrer)
|
|
|
|
std_cal_date = datetime.strptime(standard.data[0]['calibrated_on'], '%Y-%m-%d').date()
|
|
std_cal_due = datetime.strptime(standard.data[0]['calibration_due'], '%Y-%m-%d').date()
|
|
|
|
# Process each calibration
|
|
for i in range(len(channel_serials)):
|
|
# Get channel ID from serial with detailed error handling
|
|
try:
|
|
print(f"Looking up channel with serial: {channel_serials[i]}")
|
|
channel = supabase.table('channels').select('id').eq('serial_number', channel_serials[i]).execute()
|
|
if not channel.data:
|
|
error_msg = f'Channel not found: {channel_serials[i]}'
|
|
flash(error_msg, 'error')
|
|
print(error_msg)
|
|
continue
|
|
print(f"Found channel ID: {channel.data[0]['id']}")
|
|
except Exception as e:
|
|
error_msg = f'Error looking up channel {channel_serials[i]}: {str(e)}'
|
|
flash(error_msg, 'error')
|
|
print(error_msg)
|
|
continue
|
|
|
|
calibration_data = {
|
|
'channel_id': channel.data[0]['id'],
|
|
'work_order_id': work_order_id,
|
|
'calibrated_by': calibrated_by,
|
|
'std_used': std_used,
|
|
'std_cal_date': std_cal_date.isoformat(),
|
|
'std_cal_due': std_cal_due.isoformat(),
|
|
'date': date.isoformat(),
|
|
'scale': float(scales[i]),
|
|
'offset': float(offsets[i]),
|
|
'deviation_high': float(deviation_highs[i]),
|
|
'deviation_mid': float(deviation_mids[i]),
|
|
'deviation_low': float(deviation_lows[i]),
|
|
'set_high': float(set_highs[i]),
|
|
'set_mid': float(set_mids[i]),
|
|
'set_low': float(set_lows[i]),
|
|
'passed': bool(passed_list[i] if i < len(passed_list) else False)
|
|
}
|
|
|
|
# Insert calibration with error handling
|
|
try:
|
|
result = supabase.table('calibrations').insert(calibration_data).execute()
|
|
if not result.data:
|
|
flash(f'Failed to create calibration for {channel_serials[i]}', 'error')
|
|
print(f"Supabase insert failed for channel {channel_serials[i]}: {result}")
|
|
else:
|
|
print(f"Successfully created calibration for channel {channel_serials[i]}: {result.data}")
|
|
except Exception as e:
|
|
flash(f'Error creating calibration for {channel_serials[i]}: {str(e)}', 'error')
|
|
print(f"Exception creating calibration for {channel_serials[i]}: {str(e)}")
|
|
continue
|
|
|
|
flash('Calibrations processed', 'success')
|
|
return redirect(url_for('calibrations.index'))
|
|
|
|
@calibrations_bp.route('/review/<calibration_id>', methods=['GET', 'POST'])
|
|
def review_calibration(calibration_id):
|
|
supabase = get_supabase()
|
|
|
|
if request.method == 'POST':
|
|
# Verify reviewer is logged in
|
|
reviewer_id = request.form.get('reviewer_id')
|
|
signature = request.form.get('signature')
|
|
|
|
if not reviewer_id or not signature:
|
|
flash('Reviewer ID and signature are required', 'error')
|
|
return redirect(request.referrer)
|
|
|
|
# Save signature to user profile
|
|
supabase.table('users').update({
|
|
'signature_image': signature
|
|
}).eq('id', reviewer_id).execute()
|
|
|
|
# Update calibration with review info
|
|
result = supabase.table('calibrations').update({
|
|
'reviewed_by': reviewer_id,
|
|
'reviewed_at': datetime.now().isoformat()
|
|
}).eq('id', calibration_id).execute()
|
|
|
|
if not result.data:
|
|
flash('Failed to update calibration review', 'error')
|
|
else:
|
|
# Log audit event
|
|
supabase.table('calibration_audit').insert({
|
|
'calibration_id': calibration_id,
|
|
'user_id': reviewer_id,
|
|
'action': 'review',
|
|
'new_values': {
|
|
'reviewed_by': reviewer_id,
|
|
'reviewed_at': datetime.now().isoformat()
|
|
},
|
|
'ip_address': request.remote_addr
|
|
}).execute()
|
|
|
|
flash('Calibration reviewed successfully', 'success')
|
|
|
|
return redirect(url_for('calibrations.index'))
|
|
|
|
# Get calibration details with related data
|
|
calibration = supabase.table('calibrations').select('''
|
|
*,
|
|
channels:channel_id(serial_number),
|
|
calibrated_by:calibrated_by(name),
|
|
work_orders:work_order_id(order_number)
|
|
''').eq('id', calibration_id).execute()
|
|
|
|
if not calibration.data:
|
|
flash('Calibration not found', 'error')
|
|
return redirect(url_for('calibrations.index'))
|
|
|
|
return render_template('calibration_review.html',
|
|
calibration=calibration.data[0],
|
|
user=session)
|
|
|
|
@calibrations_bp.route('/new')
|
|
def new_calibration():
|
|
supabase = get_supabase()
|
|
|
|
# Get non-completed work orders with customer names
|
|
work_orders = supabase.table('work_orders').select('''
|
|
id,
|
|
order_number,
|
|
customers:customer_id(name)
|
|
''').neq('status', 'Completed').execute()
|
|
|
|
# Get all standards
|
|
standards = supabase.table('standards').select('id, make, model').execute()
|
|
|
|
# Get all channels (just serial numbers for now)
|
|
channels = supabase.table('channels').select('serial_number').execute()
|
|
|
|
# Get all probe models (using exact column name)
|
|
probe_models = supabase.table('probe_models').select('id, model_name').execute()
|
|
|
|
# Get all parameters (using correct column name)
|
|
parameters = supabase.table('parameters').select('id, parameter_name').execute()
|
|
|
|
return render_template('calibration_form.html',
|
|
work_orders=work_orders.data,
|
|
standards=standards.data,
|
|
channels=channels.data,
|
|
probe_models=probe_models.data,
|
|
parameters=parameters.data,
|
|
user=session)
|
|
|
|
@calibrations_bp.route('/filtered-channels')
|
|
def get_filtered_channels():
|
|
supabase = get_supabase()
|
|
|
|
probe_model_id = request.args.get('probe_model')
|
|
parameter_id = request.args.get('parameter')
|
|
|
|
if not probe_model_id or not parameter_id:
|
|
return jsonify([])
|
|
|
|
# Get channels filtered by both probe model and parameter
|
|
channels = supabase.table('channels').select('''
|
|
serial_number,
|
|
probes:probe_id(model_id),
|
|
parameter_id
|
|
''').eq('parameter_id', parameter_id).execute()
|
|
|
|
# Filter channels by probe model (client-side fallback)
|
|
filtered_channels = []
|
|
for c in channels.data:
|
|
try:
|
|
if c['probes']['model_id'] == probe_model_id:
|
|
filtered_channels.append({
|
|
'serial_number': c['serial_number']
|
|
})
|
|
except (KeyError, TypeError):
|
|
continue
|
|
|
|
print(f"Filtered channels for model {probe_model_id} and parameter {parameter_id}: {filtered_channels}")
|
|
|
|
return jsonify(filtered_channels)
|
|
|
|
@calibrations_bp.route('/')
|
|
def index():
|
|
supabase = get_supabase()
|
|
|
|
# Get summary statistics
|
|
total_result = supabase.table('calibrations').select('*').execute()
|
|
passed_result = supabase.table('calibrations').select('*').eq('passed', True).execute()
|
|
total = len(total_result.data) if total_result.data else 0
|
|
passed = len(passed_result.data) if passed_result.data else 0
|
|
failed = total - passed
|
|
|
|
# Get recent calibrations (last 10)
|
|
recent_calibrations = supabase.table('calibrations').select('''
|
|
*,
|
|
channels:channel_id(serial_number),
|
|
standards:std_used(make, model),
|
|
calibrated_by:calibrated_by(name)
|
|
''').order('date', desc=True).limit(10).execute()
|
|
|
|
return render_template('calibration_dashboard.html',
|
|
summary={
|
|
'total_calibrations': total,
|
|
'passed_calibrations': passed,
|
|
'failed_calibrations': failed
|
|
},
|
|
recent_calibrations=recent_calibrations.data,
|
|
user=session)
|
|
|
|
@calibrations_bp.route('/probe/<probe_id>')
|
|
def calibrations_for_probe(probe_id):
|
|
"""List all calibrations for a specific probe"""
|
|
supabase = get_supabase()
|
|
|
|
# Get all channels for this probe
|
|
channels = supabase.table('channels').select('id').eq('probe_id', probe_id).execute()
|
|
channel_ids = [c['id'] for c in channels.data]
|
|
|
|
# Get all calibrations for these channels
|
|
calibrations = supabase.table('calibrations').select('''
|
|
*,
|
|
channels:channel_id(serial_number),
|
|
standards:std_used(make, model),
|
|
calibrated_by:calibrated_by(name),
|
|
reviewed_by:reviewed_by(name)
|
|
''').in_('channel_id', channel_ids).order('date', desc=True).execute()
|
|
|
|
# Get probe details
|
|
probe = supabase.table('probes').select('*').eq('id', probe_id).execute()
|
|
|
|
return render_template('calibration_history.html',
|
|
calibrations=calibrations.data,
|
|
probe=probe.data[0] if probe.data else None,
|
|
user=session)
|
|
|
|
@calibrations_bp.route('/probe/<probe_id>/trends')
|
|
def calibration_trends(probe_id):
|
|
"""Get calibration trend data for a specific probe"""
|
|
supabase = get_supabase()
|
|
|
|
# Get all channels for this probe
|
|
channels = supabase.table('channels').select('id, serial_number').eq('probe_id', probe_id).execute()
|
|
channel_ids = [c['id'] for c in channels.data]
|
|
|
|
# Get all calibrations for these channels
|
|
calibrations = supabase.table('calibrations').select('''
|
|
id,
|
|
channel_id,
|
|
date,
|
|
deviation_high,
|
|
deviation_mid,
|
|
deviation_low,
|
|
channels:channel_id(serial_number)
|
|
''').in_('channel_id', channel_ids).order('date').execute()
|
|
|
|
# Organize data by channel for charting
|
|
trend_data = {}
|
|
for cal in calibrations.data:
|
|
channel_id = cal['channel_id']
|
|
if channel_id not in trend_data:
|
|
trend_data[channel_id] = {
|
|
'serial': cal['channels']['serial_number'],
|
|
'dates': [],
|
|
'high': [],
|
|
'mid': [],
|
|
'low': []
|
|
}
|
|
trend_data[channel_id]['dates'].append(cal['date'])
|
|
trend_data[channel_id]['high'].append(cal['deviation_high'])
|
|
trend_data[channel_id]['mid'].append(cal['deviation_mid'])
|
|
trend_data[channel_id]['low'].append(cal['deviation_low'])
|
|
|
|
return {
|
|
'probe_id': probe_id,
|
|
'trends': trend_data
|
|
}
|
|
|
|
@calibrations_bp.route('/<calibration_id>')
|
|
def view_calibration(calibration_id):
|
|
"""View details of a single calibration"""
|
|
supabase = get_supabase()
|
|
|
|
# Get calibration details with related data
|
|
calibration = supabase.table('calibrations').select('''
|
|
*,
|
|
channels:channel_id(serial_number, probe_id),
|
|
calibrated_by:calibrated_by(name),
|
|
reviewed_by:reviewed_by(name),
|
|
work_orders:work_order_id(order_number),
|
|
standards:std_used(make, model, description)
|
|
''').eq('id', calibration_id).execute()
|
|
|
|
if not calibration.data:
|
|
flash('Calibration not found', 'error')
|
|
return redirect(url_for('calibrations.index'))
|
|
|
|
return render_template('calibration_view.html',
|
|
calibration=calibration.data[0],
|
|
user=session)
|