diff --git a/.env.example b/.env.example index 5178bc9..b78e297 100644 --- a/.env.example +++ b/.env.example @@ -5,4 +5,10 @@ POSTGRES_PORT=5432 DATABASE_URL=postgresql+psycopg://pdm:pdm@localhost:5432/pdm SIMULATOR_WAIT_TIME=30 WORKER_POLL_INTERVAL=30 -MODEL_PATH=/Users/user/models/model.json \ No newline at end of file +MODEL_PATH=/Users/user/models/model.json +CORS_ORIGINS= +RUL_WARNING=75 +RUL_CRITICAL=30 +RUL_CLIP=125 +BOT_TOKEN= +CHAT_ID= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1d3e0b1..066a9bf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -115,6 +115,9 @@ jobs: MODEL_PATH: /tmp/model.json SIMULATOR_WAIT_TIME: "0" WORKER_POLL_INTERVAL: "5" + RUL_WARNING: 75 + RUL_CRITICAL: 30 + RUL_CLIP: 125 steps: - uses: actions/checkout@v4 diff --git a/backend/alembic/versions/e74d5734bc65_.py b/backend/alembic/versions/e74d5734bc65_.py new file mode 100644 index 0000000..bc76808 --- /dev/null +++ b/backend/alembic/versions/e74d5734bc65_.py @@ -0,0 +1,31 @@ +"""empty message + +Revision ID: e74d5734bc65 +Revises: 6732fa7119a4 +Create Date: 2026-11-04 16:20:31.179328 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = 'e74d5734bc65' +down_revision: str | Sequence[str] | None = '6732fa7119a4' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.add_column('alerts', sa.Column('sent_at', sa.DateTime(), nullable=True)) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column('alerts', 'sent_at') + # ### end Alembic commands ### diff --git a/backend/app/core/config.py b/backend/app/core/config.py index d845c6d..2c4cbd0 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -1,7 +1,11 @@ import os +from dotenv import load_dotenv + +load_dotenv() + CORS_ORIGINS = os.getenv("CORS_ORIGINS", "http://localhost:5173").split(",") -RUL_WARNING = 75 -RUL_CRITICAL = 30 -RUL_CLIP = 125 +RUL_WARNING = int(os.getenv("RUL_WARNING")) +RUL_CRITICAL = int(os.getenv("RUL_CRITICAL")) +RUL_CLIP = int(os.getenv("RUL_CLIP")) diff --git a/backend/app/integrations/telegram.py b/backend/app/integrations/telegram.py index 6ea5eb2..ce50c72 100644 --- a/backend/app/integrations/telegram.py +++ b/backend/app/integrations/telegram.py @@ -22,7 +22,7 @@ def format_message(alert: Alert) -> str: return (f"{emoji} {alert.title} \n" f"{alert.message}\n") -async def send_telegram_alert(bot_token: str, chat_id: str, alert: Alert): +async def send_telegram_alert(bot_token: str, chat_id: str, alert: Alert) -> bool: url = f"https://api.telegram.org/bot{bot_token}/sendMessage" payload = { "chat_id": chat_id, @@ -40,21 +40,28 @@ async def send_telegram_alert(bot_token: str, chat_id: str, alert: Alert): if r.status_code==200: logger.info(f"SENT {alert.severity.upper()} {alert.title}") circuit_breaker.set_closed() + return True except httpx.HTTPStatusError as e: if e.response.is_client_error and e.response.status_code != 429: logger.error(f"error code: {e.response.status_code}, error text {e.response.text}") circuit_breaker.set_closed() + return False elif e.response.is_server_error or e.response.status_code == 429: circuit_breaker.check_half_open() circuit_breaker.errors_counter() logger.error(f"error code: {e.response.status_code}, error text {e.response.text}") + return False else: logger.error(f"error code: {e.response.status_code}, error text {e.response.text}") + return False except httpx.TransportError as e: circuit_breaker.check_half_open() circuit_breaker.errors_counter() logger.error(f"transport error {type(e).__name__}") + return False + + return False diff --git a/backend/app/models/alerts.py b/backend/app/models/alerts.py index 993989f..939e7ca 100644 --- a/backend/app/models/alerts.py +++ b/backend/app/models/alerts.py @@ -22,6 +22,7 @@ class Alert(Base): triggered_at = Column(DateTime, nullable = False, server_default=func.now()) severity = Column(Text, nullable=False) message = Column(Text, nullable=False) + sent_at = Column(DateTime, nullable=True) __table_args__ = (Index("ix_alerts_equipment_triggered", equipment_id, triggered_at.desc()),) diff --git a/backend/app/workers/alert_sender.py b/backend/app/workers/alert_sender.py new file mode 100644 index 0000000..523fbaf --- /dev/null +++ b/backend/app/workers/alert_sender.py @@ -0,0 +1,61 @@ +import asyncio +import logging +import os +import sys +import time + +from app.integrations.telegram import Alert, send_telegram_alert +from dotenv import load_dotenv +from sqlalchemy import create_engine, text + +logger = logging.getLogger("telegram alerts") + + +load_dotenv() + +engine = create_engine(os.getenv("DATABASE_URL")) + +BOT_TOKEN = os.getenv("BOT_TOKEN") +CHAT_ID = os.getenv("CHAT_ID") +INTERVAL_SEC = int(os.getenv("SIMULATOR_WAIT_TIME", "30")) + +def get_rows_without_alert(): + query_first = text(""" + SELECT severity, message, id FROM alerts WHERE (sent_at IS NULL) + """) + + query_second = text(""" + UPDATE alerts SET sent_at = LOCALTIMESTAMP WHERE id = :id + """) + + while True: + with engine.connect() as connection: + res = connection.execute(query_first) + if (res.rowcount == 0): + logger.info("telegram alerts: all processed") + else: + res_str = res.fetchall() + + for i in range(len(res_str)): + title = f"{res_str[i][0]} message" + message = f"{res_str[i][1]}" + severity = f"{res_str[i][0]}" + + alert = Alert(title, message, severity) + flag = asyncio.run(send_telegram_alert(BOT_TOKEN, CHAT_ID, alert)) + + if flag: + connection.execute(query_second, {'id': res_str[i][2]}) + connection.commit() + + time.sleep(INTERVAL_SEC) + + +if __name__ == "__main__": + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + stream=sys.stdout, + ) + + get_rows_without_alert() diff --git a/ml/worker.py b/ml/worker.py index d415fe6..a02a5f2 100644 --- a/ml/worker.py +++ b/ml/worker.py @@ -13,6 +13,9 @@ load_dotenv() +warning_rul = int(os.getenv("RUL_WARNING")) +critical_rul = int(os.getenv("RUL_CRITICAL")) + engine = create_engine(os.getenv("DATABASE_URL")) def get_model() -> xgb.XGBRegressor: @@ -47,9 +50,11 @@ def model_predict(df: pd.DataFrame, model: xgb.XGBRegressor) -> np.ndarray: return arr -def write_to_db(): +def write_to_tables(): model = get_model() + last_id = 0 + query_second = text(""" INSERT INTO ml_predictions (equipment_id, cycle, predicted_rul, defect_probability, residual, model_version) VALUES (:eq_id, :cycle, :rul, 0, 0, :model) @@ -61,7 +66,7 @@ def write_to_db(): df_pred, df = get_data_from_simulator() if df_pred is None or df is None: - logger.info("all processed") + logger.info("ml_predictions: all processed") time.sleep(int(os.getenv("WORKER_POLL_INTERVAL"))) else: t0 = time.perf_counter() @@ -77,15 +82,95 @@ def write_to_db(): elapsed_ms = (time.perf_counter() - t0) * 1000 logger.info("inserted %d rows in %.2f ms", arr.size, elapsed_ms) + flag, last_id = write_to_alerts_table(last_id) + + if flag: + logger.info("alerts: all processed") + except KeyboardInterrupt: logger.warning("interrupted by user") +def write_to_alerts_table(last_id): + query_third = text(""" + SELECT ml_predictions.id, ml_predictions.equipment_id, ml_predictions.cycle, ml_predictions.predicted_rul FROM ml_predictions + WHERE (ml_predictions.predicted_rul <= :warning_rul AND ml_predictions.id > :last_id) + """) + + query_fourth = text(""" + SELECT equipment.name FROM equipment WHERE (id = :eq_id) + """) + + + query_fifth = text(""" + INSERT INTO alerts (equipment_id, ml_prediction_id, cycle, severity, message) + VALUES (:equipment_id, :ml_prediction_id, :cycle, :severity, :message) + ON CONFLICT DO NOTHING + """) + + query_sixth = text(""" + SELECT equipment_id, severity FROM alerts WHERE (severity = :severity AND equipment_id = :eq_id) + """) + + with engine.connect() as connection: + res = connection.execute(query_third, {'warning_rul': warning_rul, 'last_id': last_id}) + + if (res.rowcount == 0): + return True, last_id + + data_ml_pred = res.fetchall() + + t0 = time.perf_counter() + counter = 0 + + for i in range (len(data_ml_pred)): + last_id = max(last_id, data_ml_pred[i][0]) + + eq_name = connection.execute(query_fourth, {'eq_id':data_ml_pred[i][1]}) + + if (eq_name.rowcount == 0): + continue + + eq_name = eq_name.fetchall() + + if data_ml_pred[i][3] > critical_rul: + severity = "warning" + + find_sev_in_alerts = connection.execute(query_sixth, {'severity':severity, 'eq_id': data_ml_pred[i][1]}) + + if find_sev_in_alerts.rowcount == 0: + message = f"warning alert on equipment name {eq_name[0][0]}" + + connection.execute(query_fifth, {'equipment_id': data_ml_pred[i][1], 'ml_prediction_id': data_ml_pred[i][0], 'cycle': data_ml_pred[i][2], 'severity': severity, 'message': message}) + connection.commit() + counter += 1 + + else: pass + + else: + severity = "critical" + + find_sev_in_alerts = connection.execute(query_sixth, {'severity':severity, 'eq_id': data_ml_pred[i][1]}) + + if find_sev_in_alerts.rowcount == 0: + message = f"critical alert on equipment name {eq_name[0][0]}" + + connection.execute(query_fifth, {'equipment_id': data_ml_pred[i][1], 'ml_prediction_id': data_ml_pred[i][0], 'cycle': data_ml_pred[i][2], 'severity': severity, 'message': message}) + connection.commit() + counter += 1 + + else: pass + + elapsed_ms = (time.perf_counter() - t0) * 1000 + logger.info("inserted %d rows in alerts in %.2f ms", counter, elapsed_ms) + + return False, last_id + if __name__ == "__main__": logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", stream=sys.stdout, ) - write_to_db() + write_to_tables()