diff --git a/app.py b/app.py index 5007831..dc1e1fe 100644 --- a/app.py +++ b/app.py @@ -1,61 +1,122 @@ import os -import warnings import logging import time +import hashlib +from datetime import datetime, timedelta, timezone + from flask import Flask, render_template, jsonify, request from flask_socketio import SocketIO from influxdb import InfluxDBClient -from datetime import datetime, timedelta, timezone -import config -import hashlib +from werkzeug.exceptions import HTTPException -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') -logger = logging.getLogger('werkzeug') +import config + +logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") +logger = logging.getLogger("werkzeug") logger.setLevel(logging.INFO) app = Flask(__name__) -app.config['SECRET_KEY'] = config.FLASK_CONFIG['secret_key'] -app.config['SEND_FILE_MAX_AGE_DEFAULT'] = config.FLASK_CONFIG.get('static_cache_timeout', 60) -# Bez Eventlet (deprecated) – prosty tryb wątkowy. -socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading', logger=False) +app.config["SECRET_KEY"] = config.FLASK_CONFIG["secret_key"] +app.config["SEND_FILE_MAX_AGE_DEFAULT"] = config.FLASK_CONFIG.get("static_cache_timeout", 60) + +# Bez Eventlet – tryb wątkowy +socketio = SocketIO(app, cors_allowed_origins="*", async_mode="threading", logger=False) + + +# -------------------- +# Helpers +# -------------------- + +def is_api_request() -> bool: + return request.path.startswith("/api/") + + +def api_error(code: int, error: str, message: str = "", **extra): + payload = {"error": error, "code": code} + if message: + payload["message"] = message + if extra: + payload.update(extra) + return jsonify(payload), code + + +def parse_range(range_str: str): + """ + Akceptuje np. '24h', '7d'. Zwraca timedelta. + """ + if not range_str: + raise ValueError("empty") + s = range_str.replace(" ", "") + if len(s) < 2: + raise ValueError("too_short") + unit = s[-1] + if unit not in ("h", "d"): + raise ValueError("bad_unit") + digits = "".join(filter(str.isdigit, s[:-1])) + if not digits: + raise ValueError("no_number") + num = int(digits) + if num <= 0: + raise ValueError("non_positive") + return timedelta(hours=num) if unit == "h" else timedelta(days=num) -def get_file_hash(filename): - full_path = os.path.join(app.static_folder, filename) - try: - with open(full_path, "rb") as f: - return hashlib.md5(f.read()).hexdigest()[:8] - except FileNotFoundError: - return "1" -# Klient InfluxDB v1 def get_influx_client(): client = InfluxDBClient( - host=config.INFLUXDB_CONFIG['host'], - port=config.INFLUXDB_CONFIG['port'], - database=config.INFLUXDB_CONFIG['database'] + host=config.INFLUXDB_CONFIG["host"], + port=config.INFLUXDB_CONFIG["port"], + database=config.INFLUXDB_CONFIG["database"], ) - if config.INFLUXDB_CONFIG['username']: - client.switch_user(config.INFLUXDB_CONFIG['username'], config.INFLUXDB_CONFIG['password']) + if config.INFLUXDB_CONFIG.get("username"): + client.switch_user(config.INFLUXDB_CONFIG["username"], config.INFLUXDB_CONFIG.get("password", "")) return client -# --- LOGIKA --- -def get_current_voltage(phase_id): +def get_current_voltage(phase_id: int): client = get_influx_client() - entity_id = config.PHASES[phase_id]['entity_id'] - query = f'SELECT "value" FROM "{config.MEASUREMENT}" WHERE "entity_id" = \'{entity_id}\' ORDER BY time DESC LIMIT 1' try: + entity_id = config.PHASES[phase_id]["entity_id"] + query = ( + f'SELECT "value" FROM "{config.MEASUREMENT}" ' + f'WHERE "entity_id" = \'{entity_id}\' ORDER BY time DESC LIMIT 1' + ) result = client.query(query) points = list(result.get_points()) - if points and points[0].get('value') is not None: - return {'voltage': round(float(points[0]['value']), 2), 'timestamp': points[0]['time']} + if points and points[0].get("value") is not None: + return {"voltage": round(float(points[0]["value"]), 2), "timestamp": points[0]["time"]} + return {"voltage": 0, "timestamp": None} except Exception as e: - print(f"Current Error: {e}") + app.logger.warning(f"Influx current_voltage error: {e}") + return {"voltage": 0, "timestamp": None} finally: - client.close() - return {'voltage': 0, 'timestamp': None} + try: + client.close() + except Exception: + pass -# --- ENDPOINTY --- + +# -------------------- +# Error handlers (40x/50x) +# -------------------- + +@app.errorhandler(HTTPException) +def handle_http_exception(e: HTTPException): + if is_api_request(): + return api_error(e.code or 500, "http_error", e.description or e.name, name=e.name) + return e + + +@app.errorhandler(Exception) +def handle_unhandled_exception(e: Exception): + app.logger.exception("Unhandled error") + if is_api_request(): + return api_error(500, "server_error", "Internal Server Error") + raise + + +# -------------------- +# Static cache busting +# -------------------- @app.context_processor def inject_static_version(): @@ -66,114 +127,144 @@ def inject_static_version(): v = hashlib.md5(f.read()).hexdigest()[:8] except Exception: v = "1" - return f"{request.script_root}/static/{filename}?v={v}" + return dict(static_v=static_v) + @app.after_request def add_header(response): - if request.path.startswith('/static/'): + if request.path.startswith("/static/"): response.cache_control.max_age = 31536000 response.cache_control.public = True - response.headers.pop('Content-Disposition', None) - + response.headers.pop("Content-Disposition", None) else: response.cache_control.no_cache = True response.cache_control.no_store = True - #response.cache_control.must_revalidate = True return response -@app.route('/favicon.ico') + +# -------------------- +# Routes +# -------------------- + +@app.route("/favicon.ico") def favicon(): - return '', 204 + return "", 204 -@app.route('/') + +@app.route("/") def index(): - return render_template('index.html', phases=config.PHASES, time_ranges=config.TIME_RANGES, - default_range=config.DEFAULT_TIME_RANGE, footer=config.FOOTER) + return render_template( + "index.html", + phases=config.PHASES, + time_ranges=config.TIME_RANGES, + default_range=config.DEFAULT_TIME_RANGE, + footer=config.FOOTER, + ) -@app.route('/api/timeseries/') + +@app.route("/api/timeseries/") def get_timeseries(phase_id): - if phase_id not in config.PHASES: - return jsonify({'error': 'Invalid phase'}), 400 - - client = get_influx_client() - range_param = request.args.get('range', config.DEFAULT_TIME_RANGE) - start_param = request.args.get('start') - end_param = request.args.get('end') - entity_id = config.PHASES[phase_id]['entity_id'] + if phase_id not in config.PHASES: + return api_error(400, "invalid_phase", "Invalid phase") + + range_param = request.args.get("range", config.DEFAULT_TIME_RANGE) + start_param = request.args.get("start") + end_param = request.args.get("end") + entity_id = config.PHASES[phase_id]["entity_id"] + + if (start_param and not end_param) or (end_param and not start_param): + return api_error(400, "bad_range", "Provide both start and end") if start_param and end_param: + try: + datetime.fromisoformat(start_param.replace("Z", "+00:00")) + datetime.fromisoformat(end_param.replace("Z", "+00:00")) + except Exception: + return api_error(400, "bad_datetime", "start/end must be ISO 8601") time_filter = f"time >= '{start_param}' AND time <= '{end_param}'" interval = config.DEFAULT_INTERVAL else: clean_range = range_param.replace(" ", "") + try: + parse_range(clean_range) + except Exception: + return api_error(400, "bad_range", "range must be like 24h or 7d") time_filter = f"time > now() - {clean_range}" - - if range_param in config.TIME_RANGES: - interval = config.TIME_RANGES[range_param]['interval'] - else: - interval = config.DEFAULT_INTERVAL + interval = config.TIME_RANGES.get(range_param, {}).get("interval", config.DEFAULT_INTERVAL) query = f''' - SELECT mean("value") AS voltage - FROM "{config.MEASUREMENT}" - WHERE "entity_id" = '{entity_id}' - AND {time_filter} + SELECT mean("value") AS voltage + FROM "{config.MEASUREMENT}" + WHERE "entity_id" = '{entity_id}' + AND {time_filter} GROUP BY time({interval}) fill(none) ''' - + + client = get_influx_client() try: result = client.query(query) - data = [{"time": p['time'], "voltage": round(p['voltage'], 2)} - for p in result.get_points() - if p.get('voltage') is not None] + data = [ + {"time": p["time"], "voltage": round(p["voltage"], 2)} + for p in result.get_points() + if p.get("voltage") is not None + ] return jsonify(data) except Exception as e: app.logger.error(f"Timeseries Error: {e} | Query: {query}") - return jsonify([]) + return api_error(503, "backend_unavailable", "InfluxDB unavailable") finally: - client.close() - -@app.route('/api/events') -def get_events(): - client = get_influx_client() - range_p = request.args.get('range', '24h') - start_p = request.args.get('start') - end_p = request.args.get('end') - now_utc = datetime.now(timezone.utc) - - # Blokada dla zakresów powyżej MAX_EVENT_RANGE_DAYS - if not (start_p and end_p) and "d" in range_p: try: - days = int(''.join(filter(str.isdigit, range_p))) - if days > config.MAX_EVENT_RANGE_DAYS: - return jsonify({ - "error": "range_too_large", - "message": f"Zbyt duży zakres. Maksymalnie {config.MAX_EVENT_RANGE_DAYS} dni. Zaznacz obszar na wykresie lub wybierz własny zakres." - }) - except: + client.close() + except Exception: pass + +@app.route("/api/events") +def get_events(): + range_p = request.args.get("range", "24h") + start_p = request.args.get("start") + end_p = request.args.get("end") + now_utc = datetime.now(timezone.utc) + + if (start_p and not end_p) or (end_p and not start_p): + return api_error(400, "bad_range", "Provide both start and end") + if start_p and end_p: - dt_view_start = datetime.fromisoformat(start_p.replace('Z', '+00:00')) - dt_view_end = datetime.fromisoformat(end_p.replace('Z', '+00:00')) + try: + dt_view_start = datetime.fromisoformat(start_p.replace("Z", "+00:00")) + dt_view_end = datetime.fromisoformat(end_p.replace("Z", "+00:00")) + except Exception: + return api_error(400, "bad_datetime", "start/end must be ISO 8601") time_filter = f"time >= '{start_p}' - 24h AND time <= '{end_p}'" else: clean_range = range_p.replace(" ", "") - num = int(''.join(filter(str.isdigit, clean_range))) - unit = clean_range[-1] - delta = timedelta(hours=num) if unit == 'h' else timedelta(days=num) + try: + delta = parse_range(clean_range) + except Exception: + return api_error(400, "bad_range", "range must be like 24h or 7d") + + if clean_range.endswith("d"): + try: + days = int("".join(filter(str.isdigit, clean_range[:-1]))) + if days > config.MAX_EVENT_RANGE_DAYS: + return api_error( + 400, + "range_too_large", + f"Zbyt duży zakres. Maksymalnie {config.MAX_EVENT_RANGE_DAYS} dni.", + ) + except Exception: + pass + dt_view_start = now_utc - delta dt_view_end = now_utc time_filter = f"time > now() - {clean_range} - 24h" all_events = [] + client = get_influx_client() try: for p_id, p_cfg in config.PHASES.items(): - # min()+max() + fill(null): - # - brak danych (np. restart Influxa) NIE udaje 0V - # - "wysokie" wykrywamy po vmax, a "niskie/zanik" po vmin query = f''' SELECT min("value") AS vmin, max("value") AS vmax FROM "{config.MEASUREMENT}" @@ -183,134 +274,168 @@ def get_events(): ''' result = client.query(query) points = list(result.get_points()) - + i = 0 while i < len(points): - vmin_val = points[i].get('vmin') - vmax_val = points[i].get('vmax') + vmin_val = points[i].get("vmin") + vmax_val = points[i].get("vmax") + # GAP (brak danych) if vmin_val is None and vmax_val is None: - start_str = points[i]['time'] - dt_s = datetime.fromisoformat(start_str.replace('Z', '+00:00')) + start_str = points[i]["time"] + dt_s = datetime.fromisoformat(start_str.replace("Z", "+00:00")) j = i - while j + 1 < len(points): - if points[j+1].get('vmin') is None and points[j+1].get('vmax') is None: - j += 1 - else: - break + while j + 1 < len(points) and points[j + 1].get("vmin") is None and points[j + 1].get("vmax") is None: + j += 1 - end_str = points[j]['time'] - dt_e = datetime.fromisoformat(end_str.replace('Z', '+00:00')) + end_str = points[j]["time"] + dt_e = datetime.fromisoformat(end_str.replace("Z", "+00:00")) duration = (dt_e - dt_s).total_seconds() / 60 + 1 if duration >= config.MIN_EVENT_DURATION_MINUTES: if dt_e >= dt_view_start and dt_s <= dt_view_end: - all_events.append({ - "start": start_str, - "end": end_str, - "phase": p_id, - "type": "zanik", - "duration": int(round(duration, 0)), - "source": "gap" - }) + all_events.append( + { + "start": start_str, + "end": end_str, + "phase": p_id, + "type": "zanik", + "duration": int(round(duration, 0)), + "source": "gap", + } + ) i = j + 1 continue vmin = float(vmin_val) if vmin_val is not None else None vmax = float(vmax_val) if vmax_val is not None else None - ev_type = None - - # Używamy progów z config.VOLTAGE_THRESHOLDS - if vmin is not None and vmin < config.VOLTAGE_THRESHOLDS['outage']: - ev_type = "zanik" - elif vmin is not None and config.VOLTAGE_THRESHOLDS['outage'] <= vmin < config.VOLTAGE_THRESHOLDS['min_safe']: - ev_type = "niskie" - elif vmax is not None and vmax > config.VOLTAGE_THRESHOLDS['max_safe']: - ev_type = "wysokie" - # else: ev_type = None (wartość w zakresie bezpiecznym 207-253V) - - if ev_type: - start_str = points[i]['time'] - dt_s = datetime.fromisoformat(start_str.replace('Z', '+00:00')) - j = i - - while j + 1 < len(points): - v_next_min = points[j+1].get('vmin') - v_next_max = points[j+1].get('vmax') - next_type = None + ev_type = None + if vmin is not None and vmin < config.VOLTAGE_THRESHOLDS["outage"]: + ev_type = "zanik" + elif vmin is not None and config.VOLTAGE_THRESHOLDS["outage"] <= vmin < config.VOLTAGE_THRESHOLDS["min_safe"]: + ev_type = "niskie" + elif vmax is not None and vmax > config.VOLTAGE_THRESHOLDS["max_safe"]: + ev_type = "wysokie" + + if ev_type: + start_str = points[i]["time"] + dt_s = datetime.fromisoformat(start_str.replace("Z", "+00:00")) + j = i + + while j + 1 < len(points): + v_next_min = points[j + 1].get("vmin") + v_next_max = points[j + 1].get("vmax") nm = float(v_next_min) if v_next_min is not None else None nx = float(v_next_max) if v_next_max is not None else None - if nm is not None and nm < config.VOLTAGE_THRESHOLDS['outage']: + + next_type = None + if nm is not None and nm < config.VOLTAGE_THRESHOLDS["outage"]: next_type = "zanik" - elif nm is not None and config.VOLTAGE_THRESHOLDS['outage'] <= nm < config.VOLTAGE_THRESHOLDS['min_safe']: + elif nm is not None and config.VOLTAGE_THRESHOLDS["outage"] <= nm < config.VOLTAGE_THRESHOLDS["min_safe"]: next_type = "niskie" - elif nx is not None and nx > config.VOLTAGE_THRESHOLDS['max_safe']: + elif nx is not None and nx > config.VOLTAGE_THRESHOLDS["max_safe"]: next_type = "wysokie" - # else: next_type = None - + if next_type == ev_type: j += 1 else: break - - end_str = points[j]['time'] - dt_e = datetime.fromisoformat(end_str.replace('Z', '+00:00')) - + + end_str = points[j]["time"] + dt_e = datetime.fromisoformat(end_str.replace("Z", "+00:00")) duration = (dt_e - dt_s).total_seconds() / 60 + 1 - - # Używamy progu minimalnego czasu trwania z config + if duration >= config.MIN_EVENT_DURATION_MINUTES: if dt_e >= dt_view_start and dt_s <= dt_view_end: - all_events.append({ - "start": start_str, - "end": end_str, - "phase": p_id, - "type": ev_type, - "duration": int(round(duration, 0)) - }) + all_events.append( + { + "start": start_str, + "end": end_str, + "phase": p_id, + "type": ev_type, + "duration": int(round(duration, 0)), + } + ) i = j i += 1 - - return jsonify(sorted(all_events, key=lambda x: x['start'], reverse=True)) + + return jsonify(sorted(all_events, key=lambda x: x["start"], reverse=True)) except Exception as e: app.logger.error(f"Event Logic Error: {e}") - return jsonify([]) + return api_error(503, "backend_unavailable", "InfluxDB unavailable") finally: - client.close() + try: + client.close() + except Exception: + pass -@app.route('/api/outages/') + +@app.route("/api/outages/") def api_outages(phase_id): + # phase_id tu jest string -> waliduj jak w timeseries + try: + pid = int(phase_id) + except Exception: + return api_error(400, "invalid_phase", "Invalid phase") + + if pid not in config.PHASES: + return api_error(400, "invalid_phase", "Invalid phase") + + t_range = request.args.get("range", "24h") + try: + parse_range(t_range) + except Exception: + return api_error(400, "bad_range", "range must be like 24h or 7d") + + entity_id = config.PHASES[pid]["entity_id"] + query = ( + f'SELECT "value" FROM "{config.MEASUREMENT}" ' + f'WHERE "entity_id" = \'{entity_id}\' ' + f'AND "value" < {config.VOLTAGE_THRESHOLDS["outage_detection"]} ' + f"AND time > now() - {t_range}" + ) + client = get_influx_client() - t_range = request.args.get('range', '24h') - entity_id = config.PHASES[phase_id]['entity_id'] - query = f'SELECT "value" FROM "{config.MEASUREMENT}" WHERE "entity_id" = \'{entity_id}\' AND "value" < {config.VOLTAGE_THRESHOLDS["outage_detection"]} AND time > now() - {t_range}' try: result = client.query(query) return jsonify(list(result.get_points())) + except Exception: + return api_error(503, "backend_unavailable", "InfluxDB unavailable") finally: - client.close() + try: + client.close() + except Exception: + pass + + +# -------------------- +# SocketIO worker +# -------------------- clients = 0 task = None -@socketio.on('connect') + +@socketio.on("connect") def handle_connect(): global clients, task clients += 1 if task is None: task = socketio.start_background_task(background_voltage_update) -@socketio.on('disconnect') + +@socketio.on("disconnect") def handle_disconnect(): global clients clients = max(0, clients - 1) + def background_voltage_update(): global clients last_refresh = 0 refresh_every_s = config.CHART_CONFIG.get("refresh_interval_seconds", 15) - interval_s = config.CHART_CONFIG['update_interval'] / 1000.0 + interval_s = config.CHART_CONFIG["update_interval"] / 1000.0 while True: if clients == 0: @@ -318,19 +443,18 @@ def background_voltage_update(): continue try: - voltages = {'timestamp': None} - + voltages = {"timestamp": None} for pid in config.PHASES.keys(): res = get_current_voltage(pid) - voltages[f'phase{pid}'] = res['voltage'] - if res['timestamp']: - voltages['timestamp'] = res['timestamp'] + voltages[f"phase{pid}"] = res["voltage"] + if res["timestamp"]: + voltages["timestamp"] = res["timestamp"] - socketio.emit('voltage_update', voltages) + socketio.emit("voltage_update", voltages) now = time.time() if now - last_refresh >= refresh_every_s: - socketio.emit('refresh_timeseries', {'ts': int(now)}) + socketio.emit("refresh_timeseries", {"ts": int(now)}) last_refresh = now except Exception as e: @@ -338,9 +462,9 @@ def background_voltage_update(): socketio.sleep(interval_s) -if __name__ == '__main__': - print("\n" + "="*50) + +if __name__ == "__main__": + print("\n" + "=" * 50) print(f"Voltage Monitor API / Port: {config.FLASK_CONFIG['port']}") - print("="*50 + "\n") - - socketio.run(app, host='0.0.0.0', port=config.FLASK_CONFIG['port'], allow_unsafe_werkzeug=True) \ No newline at end of file + print("=" * 50 + "\n") + socketio.run(app, host="0.0.0.0", port=config.FLASK_CONFIG["port"], allow_unsafe_werkzeug=True) \ No newline at end of file