add telegram alerts sender + changed worker + add migration

.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
+MODEL_PATH=/Users/user/models/model.json
+CORS_ORIGINS=
+RUL_WARNING=75
+RUL_CRITICAL=30
+RUL_CLIP=125
+BOT_TOKEN=
+CHAT_ID=

.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

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 ###

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"))

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
 

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()),)
 

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()

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()