# ============================================================ # MODULE 2: Digitaal Risicomanagement (RI&E & HAZOP) # Phoenix Metals HSEQ VBS — module_rie.py # ============================================================ from datetime import datetime, timedelta from search import get_db import os from flask import jsonify, request, render_template def init_db_rie(): """Ensure RIE tables exist.""" try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() tables = ['rie_sections','risk_types','substance_library','risk_scenarios','controls','exposure_assessments'] for t in tables: c.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{t}'") if not c.fetchone(): print(f"[RIE] Table {t} missing — migration needed") conn.close() except Exception as e: print(f"[WARN] init_db_rie: {e}") def register_rie_routes(app, page_fn, BASE_PATH="/hseq-dashboard"): _page = page_fn # --- API: Dashboard stats --- @app.route(BASE_PATH + '/api/rie/dashboard') def api_rie_dashboard(): db = get_db() try: # Stats per risk type types = db.execute(""" SELECT rt.id, rt.code, rt.name_nl, rt.color, rt.icon, COUNT(rs.id) AS total, SUM(CASE WHEN rs.raw_risk_category='extreme' THEN 1 ELSE 0 END) AS extreme, SUM(CASE WHEN rs.raw_risk_category='high' THEN 1 ELSE 0 END) AS high, SUM(CASE WHEN rs.raw_risk_category='medium' THEN 1 ELSE 0 END) AS medium, SUM(CASE WHEN rs.raw_risk_category='low' THEN 1 ELSE 0 END) AS low FROM risk_types rt LEFT JOIN risk_scenarios rs ON rs.risk_type_id = rt.id GROUP BY rt.id ORDER BY rt.sort_order """).fetchall() # Overall overall = db.execute(""" SELECT COUNT(*) AS total, SUM(CASE WHEN raw_risk_category='extreme' THEN 1 ELSE 0 END) AS extreme, SUM(CASE WHEN raw_risk_category='high' THEN 1 ELSE 0 END) AS high, SUM(CASE WHEN status='controlled' THEN 1 ELSE 0 END) AS controlled, ROUND(AVG(raw_risk_score),1) AS avg_score FROM risk_scenarios """).fetchone() # Matrix data (5x5 counts) matrix = db.execute(""" SELECT raw_likelihood AS L, raw_effect AS E, COUNT(*) AS cnt FROM risk_scenarios WHERE raw_likelihood IS NOT NULL AND raw_effect IS NOT NULL GROUP BY raw_likelihood, raw_effect """).fetchall() mit_matrix = db.execute(""" SELECT mitigated_likelihood AS L, mitigated_effect AS E, COUNT(*) AS cnt FROM risk_scenarios WHERE mitigated_likelihood IS NOT NULL AND mitigated_effect IS NOT NULL GROUP BY mitigated_likelihood, mitigated_effect """).fetchall() return jsonify({ 'types': [dict(r) for r in types], 'overall': dict(overall), 'matrix': [dict(r) for r in matrix], 'mitigated_matrix': [dict(r) for r in mit_matrix] }) finally: db.close() # --- API: List scenarios with filters --- @app.route(BASE_PATH + '/api/rie/scenarios') def api_rie_scenarios(): db = get_db() try: risk_type = request.args.get('type') section = request.args.get('section') category = request.args.get('category') query = """ SELECT rs.*, rt.name_nl AS type_name, rt.color AS type_color, rt.icon AS type_icon, sec.title AS section_title, sl.substance_name, sl.cas_number FROM risk_scenarios rs LEFT JOIN risk_types rt ON rs.risk_type_id = rt.id LEFT JOIN rie_sections sec ON rs.rie_section_id = sec.id LEFT JOIN substance_library sl ON rs.substance_id = sl.id """ conds, params = [], [] if risk_type: conds.append("rs.risk_type_id = ?") params.append(risk_type) if section: conds.append("rs.rie_section_id = ?") params.append(section) if category: conds.append("rs.raw_risk_category = ?") params.append(category) if conds: query += " WHERE " + " AND ".join(conds) query += " ORDER BY rs.raw_risk_score DESC NULLS LAST, rs.created_date DESC" rows = db.execute(query, params).fetchall() return jsonify({'scenarios': [dict(r) for r in rows], 'total': len(rows)}) finally: db.close() # --- API: Single scenario detail --- @app.route(BASE_PATH + '/api/rie/scenarios/') def api_rie_scenario_detail(sid): db = get_db() try: scenario = db.execute(""" SELECT rs.*, rt.name_nl AS type_name, rt.icon AS type_icon, sec.title AS section_title, sl.substance_name, sl.cas_number, sl.h_phrases, sl.p_phrases, sl.pbm_respiratory, sl.pbm_gloves, sl.pbm_eye, sl.pbm_body, sl.first_aid_inhalation, sl.first_aid_skin, sl.first_aid_eye FROM risk_scenarios rs LEFT JOIN risk_types rt ON rs.risk_type_id = rt.id LEFT JOIN rie_sections sec ON rs.rie_section_id = sec.id LEFT JOIN substance_library sl ON rs.substance_id = sl.id WHERE rs.id = ? """, (sid,)).fetchone() if not scenario: return jsonify({'error': 'Not found'}), 404 controls = db.execute(""" SELECT * FROM controls WHERE scenario_id = ? ORDER BY ahs_level_order """, (sid,)).fetchall() exposures = db.execute(""" SELECT ea.*, sl.substance_name, sl.cas_number, sl.oel_twa_8h, sl.oel_stel_15min FROM exposure_assessments ea LEFT JOIN substance_library sl ON ea.substance_id = sl.id WHERE ea.scenario_id = ? """, (sid,)).fetchall() return jsonify({ 'scenario': dict(scenario), 'controls': [dict(c) for c in controls], 'exposures': [dict(e) for e in exposures] }) finally: db.close() # --- API: Create scenario --- @app.route(BASE_PATH + '/api/rie/scenarios', methods=['POST']) def api_rie_scenario_create(): data = request.get_json() if not data or not data.get('title'): return jsonify({'error': 'title required'}), 400 db = get_db() try: now = datetime.utcnow().isoformat() # Auto-calculate risk score (Likelihood x Exposure x Effect) raw_l = data.get('raw_likelihood', 1) raw_e = data.get('raw_effect', 1) raw_x = data.get('raw_exposure', 2) raw_score = raw_l * raw_e * raw_x raw_cat = 'extreme' if raw_score >= 40 else 'high' if raw_score >= 20 else 'medium' if raw_score >= 8 else 'low' # Generate code count = db.execute("SELECT COUNT(*)+1 FROM risk_scenarios").fetchone()[0] code = f'RS-{count:04d}' db.execute("""INSERT INTO risk_scenarios (scenario_code, title, description, rie_section_id, risk_type_id, source_location, equipment_tag, process_unit, substance_id, hazop_node, hazop_parameter, hazop_deviation, hazop_guide_word, hazop_cause, hazop_consequence, raw_likelihood, raw_effect, raw_exposure, raw_risk_score, raw_risk_category, status, priority, owner, notes, created_date, updated_date) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (code, data['title'], data.get('description',''), data.get('rie_section_id'), data.get('risk_type_id',1), data.get('source_location'), data.get('equipment_tag'), data.get('process_unit'), data.get('substance_id'), data.get('hazop_node'), data.get('hazop_parameter'), data.get('hazop_deviation'), data.get('hazop_guide_word'), data.get('hazop_cause'), data.get('hazop_consequence'), raw_l, raw_e, raw_x, raw_score, raw_cat, data.get('status','identified'), data.get('priority','medium'), data.get('owner'), data.get('notes'), now, now)) db.commit() new_id = db.execute("SELECT last_insert_rowid()").fetchone()[0] return jsonify({'success': True, 'id': new_id, 'scenario_code': code, 'raw_risk_score': raw_score, 'raw_risk_category': raw_cat}) except Exception as e: db.rollback() return jsonify({'error': str(e)}), 500 finally: db.close() # --- API: Update scenario (incl mitigated score) --- @app.route(BASE_PATH + '/api/rie/scenarios/', methods=['PUT']) def api_rie_scenario_update(sid): data = request.get_json() db = get_db() try: now = datetime.utcnow().isoformat() # Recalc mitigated score if provided mit_l = data.get('mitigated_likelihood') mit_e = data.get('mitigated_effect') mit_x = data.get('mitigated_exposure') mit_score = None mit_cat = None if mit_l and mit_e and mit_x: mit_score = mit_l * mit_e * mit_x mit_cat = 'extreme' if mit_score >= 40 else 'high' if mit_score >= 20 else 'medium' if mit_score >= 8 else 'low' db.execute("""UPDATE risk_scenarios SET title=COALESCE(?,title), description=COALESCE(?,description), mitigated_likelihood=COALESCE(?,mitigated_likelihood), mitigated_effect=COALESCE(?,mitigated_effect), mitigated_exposure=COALESCE(?,mitigated_exposure), mitigated_risk_score=COALESCE(?,mitigated_risk_score), mitigated_risk_category=COALESCE(?,mitigated_risk_category), status=COALESCE(?,status), priority=COALESCE(?,priority), owner=COALESCE(?,owner), notes=COALESCE(?,notes), updated_date=? WHERE id=?""", (data.get('title'), data.get('description'), mit_l, mit_e, mit_x, mit_score, mit_cat, data.get('status'), data.get('priority'), data.get('owner'), data.get('notes'), now, sid)) db.commit() return jsonify({'success': True, 'mitigated_risk_score': mit_score, 'mitigated_risk_category': mit_cat}) except Exception as e: db.rollback() return jsonify({'error': str(e)}), 500 finally: db.close() # --- API: Add control --- @app.route(BASE_PATH + '/api/rie/controls', methods=['POST']) def api_rie_control_create(): data = request.get_json() if not data or not data.get('scenario_id') or not data.get('title'): return jsonify({'error': 'scenario_id and title required'}), 400 db = get_db() try: ahs_order = {'bron':1,'technisch':2,'organisatorisch':3,'pbm':4,'overig':5} db.execute("""INSERT INTO controls (scenario_id, title, description, ahs_level, ahs_level_order, control_type, implementation_status, effectiveness_score, reliability, linked_document_id, linked_action_id, notes, created_date, updated_date) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (data['scenario_id'], data['title'], data.get('description',''), data.get('ahs_level','organisatorisch'), ahs_order.get(data.get('ahs_level','organisatorisch'),5), data.get('control_type','preventive'), data.get('implementation_status','planned'), data.get('effectiveness_score',3), data.get('reliability','medium'), data.get('linked_document_id'), data.get('linked_action_id'), data.get('notes'), datetime.utcnow().isoformat(), datetime.utcnow().isoformat())) db.commit() return jsonify({'success': True}) except Exception as e: db.rollback() return jsonify({'error': str(e)}), 500 finally: db.close() # --- API: Substance auto-suggest --- @app.route(BASE_PATH + '/api/rie/substance-suggest') def api_rie_substance_suggest(): q = request.args.get('q', '').strip() if len(q) < 2: return jsonify({'results': []}) db = get_db() try: rows = db.execute(""" SELECT * FROM substance_library WHERE substance_name LIKE ? OR cas_number LIKE ? OR chemical_formula LIKE ? AND active = 1 LIMIT 10 """, (f'%{q}%', f'%{q}%', f'%{q}%')).fetchall() return jsonify({'results': [dict(r) for r in rows]}) finally: db.close() # --- API: Substance detail (for auto-fill) --- @app.route(BASE_PATH + '/api/rie/substance/') def api_rie_substance_detail(sid): db = get_db() try: row = db.execute("SELECT * FROM substance_library WHERE id=? AND active=1", (sid,)).fetchone() if not row: return jsonify({'error': 'Not found'}), 404 s = dict(row) # Build auto-suggest response suggestions = { 'substance_name': s['substance_name'], 'cas_number': s['cas_number'], 'h_phrases': s['h_phrases'], 'p_phrases': s['p_phrases'], 'primary_hazards': [], 'pbm_required': {}, 'controls_suggested': [], 'atex_zone': None, 'oel_twa': s['oel_twa_8h'] } if s['is_flammable']: suggestions['primary_hazards'].append('Brandbaar (H226)') suggestions['controls_suggested'].append({'ahs': 'bron', 'title': f"Gesloten systeem voor {s['substance_name']}-transfer"}) suggestions['controls_suggested'].append({'ahs': 'technisch', 'title': 'LEA aan verzamelput / afzuiging'}) if s['is_toxic']: suggestions['primary_hazards'].append(f'Toxisch ({s["h_phrases"]})') suggestions['controls_suggested'].append({'ahs': 'organisatorisch', 'title': f'Werkplekinstructie {s["substance_name"]} handling'}) if s['is_corrosive']: suggestions['primary_hazards'].append('Bijtend / Corrosief') if s['is_flammable'] and s['flash_point'] and s['flash_point'] < 55: suggestions['atex_zone'] = 'Zone 2 (bij verneveling/sproeien)' if s['pbm_respiratory']: suggestions['pbm_required']['respiratory'] = s['pbm_respiratory'] if s['pbm_gloves']: suggestions['pbm_required']['gloves'] = s['pbm_gloves'] if s['pbm_eye']: suggestions['pbm_required']['eye'] = s['pbm_eye'] if s['pbm_body']: suggestions['pbm_required']['body'] = s['pbm_body'] suggestions['controls_suggested'].append({'ahs': 'pbm', 'title': f"PBM: {s['pbm_gloves'] or 'Handschoenen'} + {s['pbm_eye'] or 'Veiligheidsbril'}"}) return jsonify(suggestions) finally: db.close() # --- API: Sections list --- @app.route(BASE_PATH + '/api/rie/sections') def api_rie_sections(): db = get_db() try: rows = db.execute("SELECT * FROM rie_sections WHERE active=1 ORDER BY sort_order").fetchall() return jsonify({'sections': [dict(r) for r in rows]}) finally: db.close() # === PAGE: RI&E Dashboard === @app.route(BASE_PATH + '/rie') def rie_dashboard_page(): bp = os.environ.get('BASE_PATH', '') html = render_template('rie_dashboard.html', BASE_PATH=bp) return _page(html, page_title='RI&E Risicomanagement', page_subtitle='Phoenix Metals \u2014 Digitaal Risicomanagement (Module 2)', active='rie')