import re from flask import Blueprint, render_template, request, redirect, url_for, flash, session, jsonify from app.models import Probe, ProbeModel, Channel, Parameter, ProbeLocation from app.routes.auth import role_required probes_bp = Blueprint('probes', __name__) @probes_bp.route('/api/check-channel-serials', methods=['POST'], endpoint='check_channel_serials') @role_required('review') def check_channel_serials(): """Check if channel serials already exist in database""" data = request.get_json() if not data or 'serials' not in data: return jsonify({'error': 'Invalid request'}), 400 duplicates = [] for serial in data['serials']: if Channel.get_by_serial(serial): duplicates.append(serial) return jsonify({'duplicates': duplicates}) @probes_bp.route('/') @role_required('review') def list_probes(): """List all probes with optional active filter""" active_only = request.args.get('active_only', 'true').lower() == 'true' probes = Probe.get_all(active_only=active_only) # Get channels for each probe probes_with_channels = [] for probe in probes: probe_dict = probe.__dict__ probe_dict['channels'] = Channel.get_by_probe(probe.id) probes_with_channels.append(probe_dict) user = { 'user_name': session.get('user_name'), 'can_calibrate': session.get('can_calibrate'), 'can_review': session.get('can_review') } return render_template('probe_list.html', probes=probes_with_channels, active_only=active_only, user=user) @probes_bp.route('/new') @role_required('review') def new_probe(): """Display form to create new probe""" probe_models = ProbeModel.get_all() parameters = Parameter.get_all() user = { 'user_name': session.get('user_name'), 'can_calibrate': session.get('can_calibrate'), 'can_review': session.get('can_review') } return render_template('probe_form.html', probe_models=probe_models, parameters=parameters, user=user) @probes_bp.route('/', methods=['POST']) @role_required('review') def create_probe(): """Handle new probe creation with channels""" try: # First validate all channels channel_serials = request.form.getlist('channel_serials[]') parameter_ids = request.form.getlist('parameter_ids[]') if channel_serials and parameter_ids and len(channel_serials) == len(parameter_ids): for i in range(len(channel_serials)): # Validate channel serial format [0-9A-F]{16} serial_upper = channel_serials[i].upper() if not re.fullmatch(r'^[0-9A-F]{16}$', serial_upper): raise ValueError(f'Invalid channel serial format: {channel_serials[i]}. Must be 16 hex characters (0-9, A-F)') # Check for duplicate serial existing_channel = Channel.get_by_serial(serial_upper) if existing_channel: raise ValueError(f'Channel serial already exists: {channel_serials[i]}') # Only create probe if all channels are valid probe = Probe.create( model_id=request.form['model_id'], serial_number='', # Default empty string since field was removed from form description='' # Empty description since field was removed ) # Create channels after successful probe creation if channel_serials and parameter_ids and len(channel_serials) == len(parameter_ids): for i in range(len(channel_serials)): serial_upper = channel_serials[i].upper() Channel.create( probe_id=probe.id, serial_number=serial_upper, parameter_id=parameter_ids[i] ) flash('Probe and channels created successfully', 'success') return redirect(url_for('probes.list_probes')) except ValueError as ve: flash(f'Validation error: {str(ve)}', 'danger') return redirect(url_for('probes.new_probe')) except Exception as e: flash(f'Error creating probe: {str(e)}', 'danger') return redirect(url_for('probes.new_probe')) @probes_bp.route('/') @role_required('review') def view_probe(probe_id): """View details of a single probe""" probe = Probe.get_by_id(probe_id) if not probe: flash('Probe not found', 'danger') return redirect(url_for('probes.list_probes')) # Get associated channels with their parameters channels = Channel.get_by_probe(probe_id) # Get location history locations = ProbeLocation.get_by_probe(probe_id) # Get parameters for channel form parameters = Parameter.get_all() user = { 'user_name': session.get('user_name'), 'can_calibrate': session.get('can_calibrate'), 'can_review': session.get('can_review') } return render_template('probe_view.html', probe=probe, channels=channels, locations=locations, parameters=parameters, user=user) @probes_bp.route('//channels', methods=['POST']) @role_required('review') def add_channel(probe_id): """Add a channel to an existing probe""" try: probe = Probe.get_by_id(probe_id) if not probe: flash('Probe not found', 'danger') return redirect(url_for('probes.list_probes')) if probe.retired_at: flash('Cannot add channels to retired probe', 'danger') return redirect(url_for('probes.view_probe', probe_id=probe_id)) serial = request.form.get('channel_serial', '').upper() parameter_id = request.form.get('parameter_id') # Validate channel serial format [0-9A-F]{16} if not re.fullmatch(r'^[0-9A-F]{16}$', serial): raise ValueError(f'Invalid channel serial format: {serial}. Must be 16 hex characters (0-9, A-F)') # Validate parameter_id exists and is valid if not parameter_id: raise ValueError('Parameter ID is required') if not Parameter.get_by_id(parameter_id): raise ValueError('Invalid parameter ID') # Check for duplicate serial existing_channel = Channel.get_by_serial(serial) if existing_channel: raise ValueError(f'Channel serial already exists: {serial}') # Create channel Channel.create( probe_id=probe_id, serial_number=serial, parameter_id=parameter_id ) flash('Channel added successfully', 'success') return redirect(url_for('probes.view_probe', probe_id=probe_id)) except ValueError as ve: flash(f'Validation error: {str(ve)}', 'danger') return redirect(url_for('probes.view_probe', probe_id=probe_id)) except Exception as e: flash(f'Error adding channel: {str(e)}', 'danger') return redirect(url_for('probes.view_probe', probe_id=probe_id)) @probes_bp.route('//retire', methods=['POST']) @role_required('review') def retire_probe(probe_id): """Mark a probe as retired""" try: probe = Probe.get_by_id(probe_id) if not probe: flash('Probe not found', 'danger') return redirect(url_for('probes.list_probes')) if probe.retired_at: flash('Probe is already retired', 'warning') return redirect(url_for('probes.view_probe', probe_id=probe_id)) probe.retire() flash('Probe retired successfully', 'success') return redirect(url_for('probes.view_probe', probe_id=probe_id)) except Exception as e: flash(f'Error retiring probe: {str(e)}', 'danger') return redirect(url_for('probes.view_probe', probe_id=probe_id)) @probes_bp.route('/', methods=['DELETE']) @role_required('review') def delete_probe(probe_id): """Delete a probe (only if it has no channels)""" try: probe = Probe.get_by_id(probe_id) if not probe: return jsonify({'error': 'Probe not found'}), 404 channels = Channel.get_by_probe(probe_id) if channels: return jsonify({'error': 'Cannot delete probe with channels'}), 400 probe.delete() return jsonify({'message': 'Probe deleted successfully'}), 200 except Exception as e: return jsonify({'error': str(e)}), 500