# ============================================================ # MODULE 3: Operationele Veiligheid (PtW & MoC) # Phoenix Metals HSEQ VBS — module_moc_ptw.py # ============================================================ import sqlite3 from datetime import datetime from flask import jsonify, request from search import get_db def _generate_moc_code(db): """Generate next MOC-YYYY-NNN code.""" year = datetime.now().strftime('%Y') row = db.execute( "SELECT moc_code FROM moc_requests WHERE moc_code LIKE ? ORDER BY moc_code DESC LIMIT 1", (f'MOC-{year}-%',) ).fetchone() if row: last_num = int(row['moc_code'].split('-')[-1]) next_num = last_num + 1 else: next_num = 1 return f'MOC-{year}-{next_num:03d}' def _generate_permit_code(db): """Generate next PTW-YYYY-MMDD-NNN code.""" date_str = datetime.now().strftime('%Y%m%d') prefix = f'PTW-{date_str}-' row = db.execute( "SELECT permit_code FROM ptw_permits WHERE permit_code LIKE ? ORDER BY permit_code DESC LIMIT 1", (prefix + '%',) ).fetchone() if row: last_num = int(row['permit_code'].split('-')[-1]) next_num = last_num + 1 else: next_num = 1 return f'{prefix}{next_num:03d}' def ptw_auto_configure(permit_data, substance=None): """Rule engine: auto-configure PtW safety requirements based on permit type and substance.""" overrides = {} # Breaking containment → gasmeting + LOTO + critical if permit_data.get('permit_type') == 'breaking_containment': overrides['gas_testing_required'] = 1 overrides['loto_required'] = 1 overrides['isolation_verified'] = 0 overrides['risk_level'] = 'critical' # HF (CAS 7664-39-3) → full HF protocol if substance and substance.get('cas_number') == '7664-39-3': overrides['hf_antidote_available'] = 1 overrides['continuous_monitoring'] = 1 overrides['gas_test_toxic_unit'] = 'ppm HF' overrides['ppe_respiratory'] = 'SCBA of volgelaatsmasker met HF-filter (min. A2B2E2K2)' overrides['ppe_gloves'] = 'Viton handschoenen, doorbraaktijd >480 min' overrides['ppe_eye'] = 'Chemisch veiligheidsbril + face shield' overrides['ppe_body'] = 'HF-bestendige overall (bijv. Tychem) + zuurbestendig schort' overrides['risk_level'] = 'critical' overrides['emergency_equipment'] = ( 'Calciumgluconate gel (2x16% tube) binnen 3m bereik, oogdouches, ' 'veiligheidsdouches, HF-neutralisatiemiddel' ) # DMC (CAS 616-38-6) → ATEX + gasmeting if substance and substance.get('cas_number') == '616-38-6': overrides['gas_testing_required'] = 1 overrides['atex_zone'] = 'Zone 2 (bij verneveling/sproeien)' overrides['ppe_respiratory'] = 'Halfmasker ABEK bij verhoogde blootstelling' # Brandbaar stof → gasmeting LEL if substance and substance.get('is_flammable'): overrides['gas_testing_required'] = 1 # Heet werk → brandblusser if permit_data.get('permit_type') == 'hot_work': overrides['fire_extinguisher_available'] = 1 if not overrides.get('risk_level'): overrides['risk_level'] = 'high_risk' # Besloten ruimte → gasmeting + continue monitoring if permit_data.get('permit_type') == 'confined_space': overrides['gas_testing_required'] = 1 overrides['continuous_monitoring'] = 1 if not overrides.get('risk_level'): overrides['risk_level'] = 'high_risk' return overrides def _moc_smart_review(db, moc): """Rule-based safety review for a MoC request.""" warnings = [] pssr_steps = [] substance_hazards = [] substance = None risk_scenario = None # Fetch related substance if moc.get('related_substance_id'): substance = db.execute( "SELECT * FROM substance_library WHERE id = ?", (moc['related_substance_id'],) ).fetchone() if substance: substance = dict(substance) for flag, label in [('is_flammable', 'Brandbaar'), ('is_toxic', 'Toxisch'), ('is_corrosive', 'Bijtend'), ('is_reactive', 'Reactief')]: if substance.get(flag): substance_hazards.append(label) # Fetch related risk scenario if moc.get('related_risk_scenario_id'): risk_scenario = db.execute( "SELECT * FROM risk_scenarios WHERE id = ?", (moc['related_risk_scenario_id'],) ).fetchone() if risk_scenario: risk_scenario = dict(risk_scenario) cat = moc.get('change_category', '') scope = moc.get('change_scope', '') # --- Warnings --- if substance and substance.get('is_flammable') and cat in ('equipment', 'materials', 'process'): warnings.append({ 'severity': 'high', 'category': 'ATEX', 'message': 'Brandbaar stof systeem — ATEX-zone classificatie vereist' }) if substance and substance.get('is_toxic'): warnings.append({ 'severity': 'high', 'category': 'Toxiciteit', 'message': 'Toxische stof — PBM-specificatie controleren, OEL-waarden monitoren' }) if substance and substance.get('is_corrosive'): warnings.append({ 'severity': 'high', 'category': 'Corrosie', 'message': 'Corrosieve stof — materiaalcompatibiliteit controleren (316L SS of PTFE)' }) if substance and substance.get('cas_number') == '7664-39-3': warnings.append({ 'severity': 'critical', 'category': 'HF', 'message': 'HF DETECTIE — Calciumgluconate gel verplicht binnen 3m, HF-specifieke EHBO-protocollen, SCBA bij lekkage' }) if substance and substance.get('cas_number') == '21324-40-3': warnings.append({ 'severity': 'critical', 'category': 'HF_Risk', 'message': 'LiPF6 — Bij contact met vocht ontstaat HF, klok achten vochtdichte verbindingen' }) if scope in ('major', 'catastrophic'): warnings.append({ 'severity': 'critical', 'category': 'Seveso', 'message': 'Grote wijziging — Seveso veiligheidsrapport impact assessment vereist' }) if moc.get('atex_zone_affected'): warnings.append({ 'severity': 'high', 'category': 'ATEX', 'message': 'ATEX-zone beïnvloed — EX-certificering en zone-indeling herziening verplicht' }) if risk_scenario and risk_scenario.get('raw_risk_category') == 'extreme': warnings.append({ 'severity': 'critical', 'category': 'Risico', 'message': 'Gekoppeld aan EXTREEM risicoscenario — extra veiligheidsmaatregelen en PSSR verplicht' }) if cat == 'instrumentation': warnings.append({ 'severity': 'medium', 'category': 'Instrumentatie', 'message': 'Instrumentatiewijziging — DCS/SCADA interlock testing verplicht vooraf' }) if cat == 'materials': warnings.append({ 'severity': 'medium', 'category': 'Materialen', 'message': 'Materiaalwijziging — compatibiliteit met procescondities (temp, druk, chemie) te verifiëren' }) # --- PSSR Steps --- if scope in ('major', 'catastrophic'): pssr_steps.append('Pre-Startup Safety Review met volledige checklist') if substance and substance.get('is_flammable'): pssr_steps.append('Gasdetectie systeem test en kalibratie') if substance and substance.get('is_toxic'): pssr_steps.append('Toxische gasdetectie test (HF/CL2 sensoren)') if moc.get('atex_zone_affected'): pssr_steps.append('ATEX-zone inspectie en documentatie controle') if cat in ('equipment', 'process'): pssr_steps.append('P&ID verificatie versus as-built') if cat == 'instrumentation': pssr_steps.append('Interlock en trip functie test') if moc.get('training_required'): pssr_steps.append('Training verificatie voor alle betrokken operators') if moc.get('breaking_containment') or moc.get('loto_points_count', 0) > 0: pssr_steps.append('LOTO verificatie alle energiepunten') # Always add standard steps for step in ['Noodstopfunctie test', 'Evacuatiepaden vrij en gemarkeerd', 'Commtest met controlekamer']: if step not in pssr_steps: pssr_steps.append(step) # Change impact impact_map = {'catastrophic': 'critical', 'major': 'high', 'moderate': 'medium', 'minor': 'low'} change_impact = impact_map.get(scope, 'medium') if not warnings: change_impact = 'low' return { 'status': 'Review Complete', 'moc_id': moc['id'], 'moc_code': moc['moc_code'], 'review_date': datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ'), 'reviewer': 'Kas — AI Process Safety Engineer', 'safety_warnings': warnings, 'suggested_pssr_steps': pssr_steps, 'risk_summary': { 'substance_hazards': substance_hazards, 'change_impact': change_impact, 'pssr_required': scope in ('major', 'catastrophic') or moc.get('pssr_required'), 'atex_affected': bool(moc.get('atex_zone_affected')) } } # ============================================================ # Route registration # ============================================================ def register_moc_ptw_routes(app, page_fn, BASE_PATH="/hseq-dashboard"): """Register all MoC and PtW API routes.""" import os from flask import render_template _page = page_fn @app.route(BASE_PATH + '/moc-ptw') def moc_ptw_dashboard_page(): bp = os.environ.get('BASE_PATH', '') html = render_template('moc_ptw_dashboard.html', BASE_PATH=bp) return _page(html, page_title='MoC & PtW', page_subtitle='Phoenix Metals — Operationele Veiligheid (Module 3)', active='moc-ptw') # ---- MoC Dashboard ---- @app.route(BASE_PATH + '/api/moc/dashboard') def api_moc_dashboard(): db = get_db() try: by_status = db.execute(""" SELECT status, COUNT(*) AS cnt FROM moc_requests GROUP BY status """).fetchall() by_category = db.execute(""" SELECT change_category, COUNT(*) AS cnt FROM moc_requests GROUP BY change_category """).fetchall() by_scope = db.execute(""" SELECT change_scope, COUNT(*) AS cnt FROM moc_requests GROUP BY change_scope """).fetchall() open_actions = db.execute( "SELECT COUNT(*) AS cnt FROM moc_actions WHERE status IN ('open','in_progress')" ).fetchone()['cnt'] pssr_pending = db.execute( "SELECT COUNT(*) AS cnt FROM moc_requests WHERE pssr_required=1 AND status NOT IN ('closed','cancelled')" ).fetchone()['cnt'] return jsonify({ 'by_status': [dict(r) for r in by_status], 'by_category': [dict(r) for r in by_category], 'by_scope': [dict(r) for r in by_scope], 'open_actions': open_actions, 'pssr_pending': pssr_pending }) finally: db.close() # ---- MoC CRUD ---- @app.route(BASE_PATH + '/api/moc/requests', methods=['GET', 'POST']) def api_moc_requests(): db = get_db() try: if request.method == 'GET': q = "SELECT * FROM moc_requests WHERE 1=1" params = [] for field in ('status', 'change_category', 'change_scope', 'priority'): val = request.args.get(field) if val: q += f" AND {field} = ?" params.append(val) q += " ORDER BY created_at DESC" rows = db.execute(q, params).fetchall() return jsonify([dict(r) for r in rows]) # POST — create data = request.get_json() code = _generate_moc_code(db) db.execute(""" INSERT INTO moc_requests (moc_code, title, description, proposed_change, change_category, change_scope, change_reason, requested_by, requested_by_role, department, affected_equipment, affected_area, related_substance_id, related_risk_scenario_id, atex_zone_affected, training_required, breaking_containment, pssr_required, target_date, status, priority) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( code, data.get('title'), data.get('description'), data.get('proposed_change'), data.get('change_category', 'other'), data.get('change_scope', 'minor'), data.get('change_reason'), data.get('requested_by'), data.get('requested_by_role'), data.get('department'), data.get('affected_equipment'), data.get('affected_area'), data.get('related_substance_id'), data.get('related_risk_scenario_id'), 1 if data.get('atex_zone_affected') else 0, 1 if data.get('training_required') else 0, 1 if data.get('breaking_containment') else 0, 1 if data.get('pssr_required') else 0, data.get('target_date'), data.get('status', 'draft'), data.get('priority', 'medium') )) db.commit() return jsonify({'moc_code': code, 'status': 'created'}), 201 finally: db.close() @app.route(BASE_PATH + '/api/moc/requests/', methods=['GET', 'PUT']) def api_moc_request_detail(moc_id): db = get_db() try: moc = db.execute("SELECT * FROM moc_requests WHERE id = ?", (moc_id,)).fetchone() if not moc: return jsonify({'error': 'MoC not found'}), 404 if request.method == 'GET': moc = dict(moc) moc['approvals'] = [dict(r) for r in db.execute( "SELECT * FROM moc_approvals WHERE moc_id = ? ORDER BY reviewed_at", (moc_id,) ).fetchall()] moc['actions'] = [dict(r) for r in db.execute( "SELECT * FROM moc_actions WHERE moc_id = ? ORDER BY due_date", (moc_id,) ).fetchall()] return jsonify(moc) # PUT — update data = request.get_json() fields = [] params = [] for col in ('title', 'description', 'proposed_change', 'change_category', 'change_scope', 'change_reason', 'status', 'priority', 'target_date', 'department', 'affected_equipment', 'affected_area', 'notes', 'related_substance_id', 'related_risk_scenario_id'): if col in data: fields.append(f"{col} = ?") params.append(data[col]) for col in ('atex_zone_affected', 'training_required', 'breaking_containment', 'pssr_required'): if col in data: fields.append(f"{col} = ?") params.append(1 if data[col] else 0) if not fields: return jsonify({'error': 'No fields to update'}), 400 fields.append("updated_at = datetime('now')") params.append(moc_id) db.execute(f"UPDATE moc_requests SET {', '.join(fields)} WHERE id = ?", params) db.commit() return jsonify({'status': 'updated'}) finally: db.close() # ---- MoC Approvals ---- @app.route(BASE_PATH + '/api/moc/requests//approve', methods=['POST']) def api_moc_approve(moc_id): db = get_db() try: moc = db.execute("SELECT * FROM moc_requests WHERE id = ?", (moc_id,)).fetchone() if not moc: return jsonify({'error': 'MoC not found'}), 404 data = request.get_json() stage = data.get('stage') decision = data.get('decision') db.execute(""" INSERT INTO moc_approvals (moc_id, stage, approver_name, approver_role, decision, comments, conditions) VALUES (?,?,?,?,?,?,?) """, (moc_id, stage, data.get('approver_name'), data.get('approver_role'), decision, data.get('comments'), data.get('conditions'))) # Status progression logic moc_dict = dict(moc) new_status = moc_dict['status'] if decision == 'approved': if stage == 'pse_review': new_status = 'pending_approval' elif stage == 'hseq_approval': scope = moc_dict.get('change_scope', 'minor') if scope in ('major', 'catastrophic'): new_status = 'pending_plant_approval' else: new_status = 'approved' elif stage == 'plant_approval': new_status = 'approved' elif stage == 'pssr_signoff': new_status = 'approved' elif decision == 'rejected': new_status = 'rejected' elif decision == 'returned': new_status = 'returned' db.execute("UPDATE moc_requests SET status = ?, updated_at = datetime('now') WHERE id = ?", (new_status, moc_id)) db.commit() return jsonify({'status': new_status, 'stage': stage, 'decision': decision}) finally: db.close() # ---- MoC Actions ---- @app.route(BASE_PATH + '/api/moc/requests//actions', methods=['GET', 'POST']) def api_moc_actions(moc_id): db = get_db() try: if request.method == 'GET': rows = db.execute( "SELECT * FROM moc_actions WHERE moc_id = ? ORDER BY due_date", (moc_id,) ).fetchall() return jsonify([dict(r) for r in rows]) data = request.get_json() db.execute(""" INSERT INTO moc_actions (moc_id, action_text, assigned_to, due_date, status, priority) VALUES (?,?,?,?,?,?) """, (moc_id, data.get('action_text'), data.get('assigned_to'), data.get('due_date'), data.get('status', 'open'), data.get('priority', 'medium'))) db.commit() return jsonify({'status': 'created'}), 201 finally: db.close() @app.route(BASE_PATH + '/api/moc/actions/', methods=['PUT']) def api_moc_action_update(action_id): db = get_db() try: action = db.execute("SELECT * FROM moc_actions WHERE id = ?", (action_id,)).fetchone() if not action: return jsonify({'error': 'Action not found'}), 404 data = request.get_json() fields = [] params = [] for col in ('status', 'completion_date', 'verified_by', 'assigned_to', 'due_date', 'priority', 'action_text'): if col in data: fields.append(f"{col} = ?") params.append(data[col]) if fields: fields.append("updated_at = datetime('now')") params.append(action_id) db.execute(f"UPDATE moc_actions SET {', '.join(fields)} WHERE id = ?", params) db.commit() return jsonify({'status': 'updated'}) finally: db.close() # ---- MoC Smart Review ---- @app.route(BASE_PATH + '/api/moc/requests//smart-review') def api_moc_smart_review(moc_id): db = get_db() try: moc = db.execute("SELECT * FROM moc_requests WHERE id = ?", (moc_id,)).fetchone() if not moc: return jsonify({'error': 'MoC not found'}), 404 moc = dict(moc) # Count LOTO points if breaking containment if moc.get('breaking_containment'): moc['loto_points_count'] = 1 # Flag that LOTO is needed return jsonify(_moc_smart_review(db, moc)) finally: db.close() # ============================================================ # PtW Routes # ============================================================ # ---- PtW Dashboard ---- @app.route(BASE_PATH + '/api/ptw/dashboard') def api_ptw_dashboard(): db = get_db() try: by_status = db.execute( "SELECT status, COUNT(*) AS cnt FROM ptw_permits GROUP BY status" ).fetchall() by_type = db.execute( "SELECT permit_type, COUNT(*) AS cnt FROM ptw_permits GROUP BY permit_type" ).fetchall() active = db.execute( "SELECT COUNT(*) AS cnt FROM ptw_permits WHERE status = 'active'" ).fetchone()['cnt'] expiring = db.execute( "SELECT COUNT(*) AS cnt FROM ptw_permits WHERE status = 'active' AND end_date <= date('now', '+1 day')" ).fetchone()['cnt'] gas_overdue = db.execute(""" SELECT COUNT(DISTINCT p.id) AS cnt FROM ptw_permits p WHERE p.status = 'active' AND p.gas_testing_required = 1 AND NOT EXISTS ( SELECT 1 FROM ptw_gas_tests g WHERE g.permit_id = p.id AND g.tested_at > datetime('now', '-2 hours') ) """).fetchone()['cnt'] return jsonify({ 'by_status': [dict(r) for r in by_status], 'by_type': [dict(r) for r in by_type], 'active': active, 'expiring_soon': expiring, 'gas_tests_overdue': gas_overdue }) finally: db.close() # ---- PtW CRUD ---- @app.route(BASE_PATH + '/api/ptw/permits', methods=['GET', 'POST']) def api_ptw_permits(): db = get_db() try: if request.method == 'GET': q = "SELECT * FROM ptw_permits WHERE 1=1" params = [] for field in ('status', 'permit_type', 'location'): val = request.args.get(field) if val: q += f" AND {field} = ?" params.append(val) if request.args.get('date_from'): q += " AND start_date >= ?" params.append(request.args.get('date_from')) if request.args.get('date_to'): q += " AND end_date <= ?" params.append(request.args.get('date_to')) q += " ORDER BY created_at DESC" rows = db.execute(q, params).fetchall() return jsonify([dict(r) for r in rows]) # POST — create data = request.get_json() code = _generate_permit_code(db) # Fetch substance if referenced substance = None if data.get('related_substance_id'): substance = db.execute( "SELECT * FROM substance_library WHERE id = ?", (data['related_substance_id'],) ).fetchone() if substance: substance = dict(substance) # Auto-configure overrides = ptw_auto_configure(data, substance) # Merge overrides into data for k, v in overrides.items(): data[k] = v # Validate: if status=approved, check required fields if data.get('status') == 'approved': errors = [] if data.get('gas_testing_required') and not data.get('_has_gas_test'): errors.append('Gas testing required but no gas tests recorded') if data.get('loto_required') and not data.get('_has_loto_points'): errors.append('LOTO required but no LOTO points defined') if errors: return jsonify({'error': 'Approval checks failed', 'details': errors}), 422 db.execute(""" INSERT INTO ptw_permits (permit_code, title, description, permit_type, location, area_zone, related_substance_id, requested_by, requested_by_role, contractor, work_description, risk_level, status, start_date, end_date, issuer_name, issuer_signed_at, closer_name, closer_signed_at, gas_testing_required, loto_required, isolation_verified, continuous_monitoring, hf_antidote_available, fire_extinguisher_available, atex_zone, gas_test_toxic_unit, ppe_respiratory, ppe_gloves, ppe_eye, ppe_body, emergency_equipment, special_instructions, notes) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( code, data.get('title'), data.get('description'), data.get('permit_type', 'general'), data.get('location'), data.get('area_zone'), data.get('related_substance_id'), data.get('requested_by'), data.get('requested_by_role'), data.get('contractor'), data.get('work_description'), data.get('risk_level', 'standard'), data.get('status', 'draft'), data.get('start_date'), data.get('end_date'), data.get('issuer_name'), data.get('issuer_signed_at'), data.get('closer_name'), data.get('closer_signed_at'), data.get('gas_testing_required', 0), data.get('loto_required', 0), data.get('isolation_verified', 0), data.get('continuous_monitoring', 0), data.get('hf_antidote_available', 0), data.get('fire_extinguisher_available', 0), data.get('atex_zone'), data.get('gas_test_toxic_unit'), data.get('ppe_respiratory'), data.get('ppe_gloves'), data.get('ppe_eye'), data.get('ppe_body'), data.get('emergency_equipment'), data.get('special_instructions'), data.get('notes') )) db.commit() return jsonify({'permit_code': code, 'status': 'created', 'overrides_applied': list(overrides.keys())}), 201 finally: db.close() @app.route(BASE_PATH + '/api/ptw/permits/', methods=['GET', 'PUT']) def api_ptw_permit_detail(permit_id): db = get_db() try: permit = db.execute("SELECT * FROM ptw_permits WHERE id = ?", (permit_id,)).fetchone() if not permit: return jsonify({'error': 'Permit not found'}), 404 if request.method == 'GET': permit = dict(permit) permit['gas_tests'] = [dict(r) for r in db.execute( "SELECT * FROM ptw_gas_tests WHERE permit_id = ? ORDER BY tested_at", (permit_id,) ).fetchall()] permit['loto_points'] = [dict(r) for r in db.execute( "SELECT * FROM ptw_loto_points WHERE permit_id = ? ORDER BY id", (permit_id,) ).fetchall()] permit['inspections'] = [dict(r) for r in db.execute( "SELECT * FROM ptw_inspections WHERE permit_id = ? ORDER BY inspected_at", (permit_id,) ).fetchall()] return jsonify(permit) # PUT — update data = request.get_json() permit_dict = dict(permit) # Fetch substance for auto-configure substance = None sub_id = data.get('related_substance_id') or permit_dict.get('related_substance_id') if sub_id: substance = db.execute( "SELECT * FROM substance_library WHERE id = ?", (sub_id,) ).fetchone() if substance: substance = dict(substance) # Merge data with existing for auto-configure merge_data = {**permit_dict, **data} overrides = ptw_auto_configure(merge_data, substance) # Apply overrides for k, v in overrides.items(): data[k] = v # Status transition validation new_status = data.get('status', permit_dict['status']) if new_status == 'active' and not data.get('issuer_signed_at') and not permit_dict.get('issuer_signed_at'): return jsonify({'error': 'issuer_signed_at is required before activation'}), 422 if new_status == 'completed' and not data.get('closer_signed_at') and not permit_dict.get('closer_signed_at'): return jsonify({'error': 'closer_signed_at is required before completion'}), 422 # Approval validation if new_status == 'approved': gas_req = data.get('gas_testing_required', permit_dict.get('gas_testing_required', 0)) loto_req = data.get('loto_required', permit_dict.get('loto_required', 0)) errors = [] if gas_req: gt_count = db.execute( "SELECT COUNT(*) AS c FROM ptw_gas_tests WHERE permit_id = ?", (permit_id,) ).fetchone()['c'] if gt_count == 0: errors.append('Gas testing required but no gas tests recorded') if loto_req: loto_count = db.execute( "SELECT COUNT(*) AS c FROM ptw_loto_points WHERE permit_id = ?", (permit_id,) ).fetchone()['c'] if loto_count == 0: errors.append('LOTO required but no LOTO points defined') if errors: return jsonify({'error': 'Approval checks failed', 'details': errors}), 422 fields = [] params = [] skip = {'id', 'permit_code', 'created_at', 'gas_tests', 'loto_points', 'inspections', '_has_gas_test', '_has_loto_points'} for col in ('title', 'description', 'permit_type', 'location', 'area_zone', 'related_substance_id', 'requested_by', 'requested_by_role', 'contractor', 'work_description', 'risk_level', 'status', 'start_date', 'end_date', 'issuer_name', 'issuer_signed_at', 'closer_name', 'closer_signed_at', 'atex_zone', 'gas_test_toxic_unit', 'ppe_respiratory', 'ppe_gloves', 'ppe_eye', 'ppe_body', 'emergency_equipment', 'special_instructions', 'notes'): if col in data: fields.append(f"{col} = ?") params.append(data[col]) for col in ('gas_testing_required', 'loto_required', 'isolation_verified', 'continuous_monitoring', 'hf_antidote_available', 'fire_extinguisher_available'): if col in data: fields.append(f"{col} = ?") params.append(1 if data[col] else 0) if fields: fields.append("updated_at = datetime('now')") if new_status == 'completed': fields.append("completed_at = datetime('now')") params.append(permit_id) db.execute(f"UPDATE ptw_permits SET {', '.join(fields)} WHERE id = ?", params) db.commit() return jsonify({'status': 'updated', 'overrides_applied': list(overrides.keys())}) finally: db.close() # ---- PtW Gas Tests ---- @app.route(BASE_PATH + '/api/ptw/permits//gas-tests', methods=['GET', 'POST']) def api_ptw_gas_tests(permit_id): db = get_db() try: if request.method == 'GET': rows = db.execute( "SELECT * FROM ptw_gas_tests WHERE permit_id = ? ORDER BY tested_at", (permit_id,) ).fetchall() return jsonify([dict(r) for r in rows]) data = request.get_json() db.execute(""" INSERT INTO ptw_gas_tests (permit_id, test_type, reading_value, reading_unit, result, tested_by, instrument_id, notes) VALUES (?,?,?,?,?,?,?,?) """, (permit_id, data.get('test_type'), data.get('reading_value'), data.get('reading_unit'), data.get('result', 'pass'), data.get('tested_by'), data.get('instrument_id'), data.get('notes'))) db.commit() return jsonify({'status': 'created'}), 201 finally: db.close() # ---- PtW LOTO Points ---- @app.route(BASE_PATH + '/api/ptw/permits//loto-points', methods=['GET', 'POST']) def api_ptw_loto_points(permit_id): db = get_db() try: if request.method == 'GET': rows = db.execute( "SELECT * FROM ptw_loto_points WHERE permit_id = ? ORDER BY id", (permit_id,) ).fetchall() return jsonify([dict(r) for r in rows]) data = request.get_json() db.execute(""" INSERT INTO ptw_loto_points (permit_id, energy_type, equipment_tag, isolation_method, lock_number, status, notes) VALUES (?,?,?,?,?,?,?) """, (permit_id, data.get('energy_type'), data.get('equipment_tag'), data.get('isolation_method'), data.get('lock_number'), data.get('status', 'pending'), data.get('notes'))) db.commit() return jsonify({'status': 'created'}), 201 finally: db.close() @app.route(BASE_PATH + '/api/ptw/loto-points/', methods=['PUT']) def api_ptw_loto_point_update(loto_id): db = get_db() try: loto = db.execute("SELECT * FROM ptw_loto_points WHERE id = ?", (loto_id,)).fetchone() if not loto: return jsonify({'error': 'LOTO point not found'}), 404 data = request.get_json() fields = [] params = [] for col in ('status', 'verified_by', 'verified_at', 'energy_type', 'equipment_tag', 'isolation_method', 'lock_number', 'notes'): if col in data: fields.append(f"{col} = ?") params.append(data[col]) if fields: params.append(loto_id) db.execute(f"UPDATE ptw_loto_points SET {', '.join(fields)} WHERE id = ?", params) db.commit() return jsonify({'status': 'updated'}) finally: db.close() # ---- PtW Inspections ---- @app.route(BASE_PATH + '/api/ptw/permits//inspections', methods=['GET', 'POST']) def api_ptw_inspections(permit_id): db = get_db() try: if request.method == 'GET': rows = db.execute( "SELECT * FROM ptw_inspections WHERE permit_id = ? ORDER BY inspected_at", (permit_id,) ).fetchall() return jsonify([dict(r) for r in rows]) data = request.get_json() db.execute(""" INSERT INTO ptw_inspections (permit_id, inspection_type, result, inspector_name, findings, corrective_action) VALUES (?,?,?,?,?,?) """, (permit_id, data.get('inspection_type'), data.get('result', 'satisfactory'), data.get('inspector_name'), data.get('findings'), data.get('corrective_action'))) db.commit() return jsonify({'status': 'created'}), 201 finally: db.close()