Merge branch 'SCRUM-37-frontend' into 'main'

src/pages/Eventportal.py

@@ -1,12 +1,10 @@
-import os
 from datetime import datetime
+from itertools import batched
 
 from bson import ObjectId
 from streamlit import streamlit as st
 
-from dotenv import load_dotenv
-from pymongo import MongoClient
-from itertools import batched
+from services.db_service import init_connection
 
 st.set_page_config(layout="wide")
 
@@ -26,23 +24,16 @@ st.markdown(
 review_options = [":green[:material/done:]", ":red[:material/highlight_off:]"]
 review_filter_options = ["Abgeschlossene Reviews", "Offene Reviews"]
 review_fields = [
-        ("Titel", "title_valid"),
-        ("Kategorie", "category_valid"),
-        ("Subkategorie", "format_valid"),
-        ("Startdatum", "startdate_valid"),
-        ("Startzeit", "starttime_valid"),
-        ("Enddatum", "enddate_valid"),
-        ("Endzeit", "endtime_valid"),
-        ("Adresse", "address_valid"),
-        ("Link", "url_valid")
-    ]
-
-@st.cache_resource
-def init_connection():
-    load_dotenv()
-    uri = f"mongodb+srv://{os.getenv('MONGO_USERNAME')}:{os.getenv('MONGO_PASSWORD')}@{os.getenv('MONGO_HOST')}/?retryWrites=true&w=majority&appName=Cluster0"
-    client = MongoClient(uri)
-    return client.event_data
+    ("Titel", "title_valid"),
+    ("Kategorie", "category_valid"),
+    ("Subkategorie", "format_valid"),
+    ("Startdatum", "startdate_valid"),
+    ("Startzeit", "starttime_valid"),
+    ("Enddatum", "enddate_valid"),
+    ("Endzeit", "endtime_valid"),
+    ("Adresse", "address_valid"),
+    ("Link", "url_valid")
+]
 
 
 @st.cache_resource
@@ -183,9 +174,9 @@ def display_event(event_data):
 
         if comments:
             st.write("**Kommentare**")
-            for comment_id,comment in enumerate(comments):
+            for comment_id, comment in enumerate(comments):
                 st.write(f"💬 {comment}")
-        st.chat_input("Neuer Kommentar" , key="comment_input" + str(event_data["_id"]),
+        st.chat_input("Neuer Kommentar", key="comment_input" + str(event_data["_id"]),
                       on_submit=update_review_comment,
                       args=(str(event_data["_id"]),))
 
@@ -193,8 +184,6 @@ def display_event(event_data):
 def render_review_controls(event_data):
     """Erzeugt dynamisch alle Review-Segmented-Controls für ein Event."""
 
-
-
     # Dynamisch Spalten erzeugen
     cols = st.columns(len(review_fields))
 
@@ -335,13 +324,13 @@ batches = st.session_state.batched_events
 
 # Page Content
 
-col_rev_toggle, col_control1 = st.columns([1,4])
+col_rev_toggle, col_control1 = st.columns([1, 4])
 with col_rev_toggle:
     st.toggle("Review Mode", value=False, on_change=toggle_review_mode)
 if st.session_state.review_mode:
     with col_control1:
-        st.pills("Filter Reviews", review_filter_options, key="filter_reviews_input", label_visibility="collapsed", on_change=update_filter)
-
+        st.pills("Filter Reviews", review_filter_options, key="filter_reviews_input", label_visibility="collapsed",
+                 on_change=update_filter)
 
 st.title("Eventportal")
 

src/pages/Pipeline_Demo.py

@@ -1,43 +1,15 @@
-import os
-import logging
-import threading
-from typing import Callable
-
 import streamlit as st
 from dotenv import load_dotenv
 
-from project import Config, DemoThread
-from project.adapters import InAdapter, OutAdapter, MessageQueueOutAdapterImpl
-
-logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
-logger = logging.getLogger(__name__)
+from project import Config
+from services.mq_service import create_out_adapter
 
 load_dotenv()
 
-
-@st.cache_resource
-def create_project_thread(_config: Config) -> DemoThread:
-    in_adapter_injector: Callable[[Config], InAdapter] = lambda c: InAdapter()
-
-    if config.mq_user == "local":
-        logger.info("Running with local OutAdapter (no MQ connection).")
-        out_adapter_injector: Callable[[Config], OutAdapter] = lambda c: OutAdapter()
-    else:
-        logger.info(f"Using real MessageQueueOutAdapter -> {config.queue_dest}")
-        out_adapter_injector: Callable[[Config], OutAdapter] = lambda c: MessageQueueOutAdapterImpl(c, c.queue_dest)
-
-    return DemoThread(config, None, in_adapter_injector, out_adapter_injector)
-
-
 config = Config()
 config.display()
 
-publish_thread = create_project_thread(config)
-
-if config.is_start_publish():
-    logger.info("Starting MQ consume thread...")
-    consume_thread = threading.Thread(target=publish_thread.run, daemon=True)
-    consume_thread.start()
+out_adapter = create_out_adapter(config, config.queue_dest_pipeline_trigger)
 
 st.title("Pipeline Demo")
 st.markdown("Sende eine Url an die Message Queue um die Pipeline zu starten:")
@@ -47,11 +19,9 @@ user_input = st.text_input("URL eingeben:")
 if st.button("Senden"):
     if user_input:
         try:
-
-            publish_thread._DemoThread__queue_dest.send_message(user_input.strip())
+            out_adapter.send_message(user_input.strip())
             st.info("URL wurde gesendet")
-            logger.info(f"Message sent to MQ: {user_input}")
         except Exception as e:
-            logger.exception("Fehler beim MQ-Senden")
+            st.error(f"Fehler beim senden in die Message Queue: {e}")
     else:
         st.warning("Bitte eine Nachricht eingeben, bevor du sendest.")

src/pages/Test_Evaluation.py

@@ -0,0 +1,256 @@
+import matplotlib.pyplot as plt
+import pandas as pd
+import seaborn as sns
+import streamlit as st
+from bson import ObjectId
+from html_to_markdown import convert
+
+from services import init_connection
+
+st.set_page_config(layout="wide")
+
+st.markdown(
+    """
+    <style>
+        .block-container {
+            width: 80vw;        
+            max-width: 1400px;  
+            margin: 0 auto;     
+        }
+    </style>
+    """,
+    unsafe_allow_html=True,
+)
+
+
+def print_schedule_obj(s):
+    start_date = s.get("start_date")
+    if not start_date:
+        start_date = s.get("startdate")
+    start_date_str = start_date.strftime("%d.%m.%Y") if start_date else ""
+
+    end_date = s.get("end_date")
+    if not end_date:
+        end_date = s.get("enddate")
+    end_date_str = end_date.strftime("%d.%m.%Y") if end_date else ""
+
+    start_time = s.get("start_time")
+    if not start_time:
+        start_time = s.get("starttime")
+    start_time_str = start_time.strftime("%H:%M") if start_time else ""
+
+    end_time = s.get("end_time")
+    if not end_time:
+        end_time = s.get("endtime")
+    end_time_str = end_time.strftime("%H:%M") if end_time else ""
+
+    return f"{start_date_str} - {end_date_str}  |  {start_time_str} - {end_time_str}\n\n"
+
+
+def create_data_metrics_df(overall_metrics: dict) -> pd.DataFrame:
+    field_order = ["page_type", "title", "locations", "schedule", "start_date", "end_date", "start_time", "end_time"]
+
+    rows = {key: "" for key in field_order if key in overall_metrics}
+
+    for field, metrics in overall_metrics.items():
+        cleaned_metrics = metrics.copy()
+        cleaned_metrics = {k: v for k, v in cleaned_metrics.items() if isinstance(v, float)}
+        rows[field] = cleaned_metrics
+        print(rows[field])
+
+    df = pd.DataFrame(rows).T
+    return df
+
+
+def create_confusion_matrix(overall_metrics: dict):
+    page_type_metric = overall_metrics.get("page_type", {})
+    tp_count = page_type_metric.get("tp", 0)
+    tn_count = page_type_metric.get("tn", 0)
+    fp_count = page_type_metric.get("fp", 0)
+    fn_count = page_type_metric.get("fn", 0)
+
+    cm = pd.DataFrame(
+        [[tn_count, fp_count],
+         [fn_count, tp_count]],
+        index=['Expected NO_EVENT', 'Expected EVENT'],
+        columns=['Predicted NO_EVENT', 'Predicted EVENT']
+    )
+
+    fig, ax = plt.subplots()
+    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', ax=ax)
+    ax.set_xlabel('Predicted')
+    ax.set_ylabel('Expected')
+
+    return fig
+
+
+def create_fn_df(record_results: dict):
+    false_negatives = [v.get("data", {}).get("reason") for v in record_results.values() if
+                       v.get("record_metrics", {}).get("page_type", {}).get("fn") == 1]
+
+    fn_counts = pd.Series(false_negatives).value_counts()
+    fn_percent = fn_counts / len(record_results) * 100
+
+    df = pd.DataFrame({
+        'Rejected Reason': fn_counts.index,
+        'Anzahl': fn_counts.values,
+        'Prozent': fn_percent.values
+    })
+
+    total_row = pd.DataFrame([{
+        'Rejected Reason': 'Gesamt',
+        'Anzahl': fn_counts.sum(),
+        'Prozent': fn_percent.sum()
+    }])
+    df = pd.concat([df, total_row], ignore_index=True)
+    df = df.style.format({'Prozent': '{:.1f}%'})
+    return df
+
+
+def create_error_df(overall_metrics: dict, batchsize: int):
+    rows = []
+    sum = 0
+    for k, v in overall_metrics.get("error", {}).items():
+        rows.append({
+            'Error': k.upper(),
+            'Anzahl': v,
+            'Prozent': v / batchsize * 100
+        })
+        sum += v
+    rows.append({
+        'Error': "Gesamt",
+        'Anzahl': sum,
+        'Prozent': sum / batchsize * 100
+    })
+    sum += v
+    df = pd.DataFrame(rows).style.format({'Prozent': '{:.1f}%'})
+    return df
+
+
+def create_detail_table(test: dict):
+    def _stringify(v):
+        if isinstance(v, list):
+            return ", ".join(map(str, v))
+        if isinstance(v, dict):
+            return str(v)
+        return str(v)
+
+    rows = []
+    meta_columns = set()
+
+    for record_id, result in test.get("record_results", {}).items():
+        validation = db.testdata_1.find_one({"_id": ObjectId(record_id)})
+        expected = validation.get("data", {})
+        predicted = result.get("data", {})
+        metrics = result.get("record_metrics", {})
+        meta = result.get("meta", {})
+
+        # Meta columns sammeln
+        meta_columns |= {f"Meta - {k}" for k in meta}
+
+        # Kopfzeile pro Datensatz
+        head = {
+            "Record ID": str(record_id),
+            "Field": "",
+            "Expected": "",
+            "Predicted": "",
+            "Metrics": "",
+            **{f"Meta - {k}": _stringify(meta.get(k, "")) for k in meta}
+        }
+        rows.append(head)
+
+        def add(field, exp, pred):
+            val = metrics.get(field)
+            if isinstance(val, dict):
+                val = " | ".join(f"{k}: {v:.2f}" for k, v in val.items())
+            rows.append({
+                "Record ID": "",
+                "Field": field,
+                "Expected": exp,
+                "Predicted": pred,
+                "Metrics": val if val is not None else "",
+                **{col: "" for col in meta_columns}
+            })
+
+        add("page_type", validation.get("page_type"), result.get("page_type"))
+        add("title", expected.get("title"), predicted.get("title"))
+        add(
+            "schedule",
+            "\n\n".join(print_schedule_obj(s) for s in expected.get("schedule", [])),
+            "\n\n".join(print_schedule_obj(s) for s in predicted.get("schedule", []))
+        )
+        add(
+            "locations",
+            ", ".join(g.get("geolocation", {}).get("formatted", "") for g in expected.get("locations", [])),
+            ", ".join(g.get("geolocation", {}).get("formatted", "") for g in predicted.get("locations", []))
+        )
+
+        rows.append({col: "" for col in ["Record ID", "Field", "Expected", "Predicted", "Metrics", *meta_columns]})
+
+    return pd.DataFrame(rows)
+
+
+@st.dialog("Original Seite", width="medium")
+def show_website(url, html):
+    st.info(f"Link zur Original Website: {url}")
+    md = convert(html)
+    st.write(md)
+
+
+st.title("Test Evaluation")
+db = init_connection()
+tests = list(db.test_evaluation.find({}, {"_id": 1, "status": 1, "created_at": 1}))
+tests_sorted = sorted(tests, key=lambda t: t["created_at"], reverse=True)
+
+options = {str(t["_id"]): f"{t['status']} - {t['created_at'].strftime('%Y-%m-%d %H:%M:%S')}" for t in tests_sorted}
+
+selected_id = st.selectbox("Wähle einen Test aus", options=list(options.keys()), format_func=lambda x: options[x])
+
+if selected_id:
+    test = db.test_evaluation.find_one({"_id": ObjectId(selected_id)})
+    record_results = test.get("record_results", {})
+    batchsize = len(record_results)
+
+    st.write(
+        f"**Test ID:**  {selected_id} | **Status:** {test.get("status")} | **Batchsize:** {batchsize}")
+
+    overall_metrics = test.get("overall_metrics", {})
+    if overall_metrics:
+        df_data_metrics = create_data_metrics_df(overall_metrics)
+        cm_fig = create_confusion_matrix(overall_metrics)
+        df_fn = create_fn_df(record_results)
+        df_error = create_error_df(overall_metrics, batchsize)
+
+        st.write("# Overall Metrics")
+
+        st.write("### Data Metrics")
+        st.bar_chart(df_data_metrics, width=400, stack=False, sort=False)
+
+        col1, col2 = st.columns(2)
+        with col1:
+            st.write("### Confusion Matrix für Page Classification (page_type)")
+            st.pyplot(cm_fig, width=450)
+
+        with col2:
+            st.write("### Falsch abgelehnte Seiten (false negatives fn)")
+            st.dataframe(df_fn)
+
+            st.write("### Fehler in der Pipeline (error)")
+            st.dataframe(df_error)
+
+
+    else:
+        st.info("Der Test läuft noch. Es konnte noch keine Metric erstellt werden")
+
+    st.write(f"# Testergebnisse im Detail")
+    df = create_detail_table(test)
+    st.dataframe(df, height=600)
+
+    record_id = st.text_input(label="Gebe eine Record ID ein um die Original Website anzusehen.", value="")
+    if record_id:
+        record = db.testdata_1.find_one({"_id": ObjectId(record_id)})
+        html = record.get("html")
+        url = record.get("url")
+        if html:
+            html = html.decode("utf-8")
+            show_website(url, html)

src/pages/Testing.py

@@ -0,0 +1,71 @@
+import base64
+import json
+
+import streamlit as st
+from dotenv import load_dotenv
+
+from project import Config
+from services.db_service import init_connection
+from services.mq_service import create_out_adapter
+from services.test_creator import TestCreator
+
+load_dotenv()
+config = Config()
+config.display()
+out_adapter = create_out_adapter(config, config.queue_dest_test_trigger)
+
+db = init_connection()
+test_creator = TestCreator(db)
+
+st.title("Testing")
+st.write(f"Es sind {db.testdata_1.count_documents({})} Testdatensätze vorhanden.")
+
+with st.form("Testkonfiguration"):
+    batch_id = st.selectbox("Testbatch wählen", [0, 1, 2, 3, "Alle"])
+    start_test = st.form_submit_button("Starte Test")
+
+if start_test:
+    if batch_id == "Alle":
+        batch = list(db.testdata_1.find({}))
+    else:
+        batch = list(db.testdata_1.find({"batch_id": batch_id}))
+    record_ids = [str(data["_id"]) for data in batch]
+
+    test_id = test_creator.create(record_ids)
+
+
+    if test_id:
+        st.info(f"Starte Test mit {len(batch)} Testdaten, Test-ID: {str(test_id)}")
+        st.write(f"**Testdaten aus dem Batch:**")
+
+        for testdata in batch:
+            id = str(testdata["_id"])
+            url = testdata.get("url")
+            html = testdata.get("html", None)
+            if html:
+                html = html.decode("utf-8")
+            else: html = ""
+            data = testdata.get("data")
+            st.write(f"Data ID: {id}")
+            st.write(f"URL: {url}")
+            st.write("___")
+
+            try:
+                message = {
+                    "url": url,
+                    "body": base64.b64encode(html.encode("utf-8")).decode("utf-8"),
+                    "meta": [{
+                        "service_name": "demo-service",
+                        "queue_dest": "storehtml-out",
+                        "test_id": str(test_id),
+                        "record_id": id,
+                    }]
+                }
+                encoded = json.dumps(message).encode("utf-8")
+                out_adapter.send_message(encoded)
+                print(f"Sent {url}")
+
+            except Exception as e:
+                st.error(f"Fehler beim senden in die Message Queue: {e}")
+    else:
+        st.error("Fehler beim anlegen des Tests")

src/project/__init__.py

@@ -1,2 +1 @@
 from .config import Config
-from .demo_thread import DemoThread

src/project/adapters/__init__.py

@@ -1,4 +1,2 @@
-from .in_adapter import InAdapter
-from .message_queue_in_adapter_impl import MessageQueueInAdapterImpl
 from .message_queue_out_adapter_impl import MessageQueueOutAdapterImpl
 from .out_adapter import OutAdapter

src/project/adapters/in_adapter.py

@@ -1,12 +0,0 @@
-from abc import abstractmethod
-from typing import Callable
-
-from pika import spec
-from pika.adapters.blocking_connection import BlockingChannel
-
-
-class InAdapter:
-
-    @abstractmethod
-    def start_consuming(self, callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]) -> None:
-        pass

src/project/adapters/message_queue_in_adapter_impl.py

@@ -1,19 +0,0 @@
-from typing import Callable
-
-from pika import spec
-from pika.adapters.blocking_connection import BlockingChannel
-
-from project import Config
-from project.adapters.in_adapter import InAdapter
-from wrapper import MessageQueue
-
-
-class MessageQueueInAdapterImpl(InAdapter):
-
-    def __init__(self, config: Config, queue_name: str) -> None:
-        self.__config = config
-        # initialize queue
-        self.__queue = MessageQueue(config.mq_host, config.mq_user, config.mq_password, queue_name)
-
-    def start_consuming(self, callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]) -> None:
-        self.__queue.start_consuming(callback)

src/project/config.py

@@ -21,7 +21,7 @@ class Config:
         self.mq_password = os.environ.get('MQ_PASSWORD')
 
         # different modes:
-        self.queue_mode = os.environ.get('QUEUE_MODE', MODE_CONSUME).lower()
+        self.queue_mode = os.environ.get('QUEUE_MODE', MODE_PUBLISH).lower()
         print(self.queue_mode)
         # consume: take elements from a queue (alias pull, alias store)
         if self.queue_mode in [MODE_CONSUME, 'pull', 'store', ]:
@@ -33,8 +33,8 @@ class Config:
         if self.queue_mode in [MODE_CONSUME_AND_PUBLISH, 'pull-push', 'store-n-forward', ]:
             self.queue_mode = MODE_CONSUME_AND_PUBLISH
 
-        self.queue_src = os.environ.get('QUEUE_SRC')
-        self.queue_dest = os.environ.get('QUEUE_DEST')
+        self.queue_dest_test_trigger= os.environ.get('QUEUE_DEST_TEST_TRIGGER')
+        self.queue_dest_pipeline_trigger = os.environ.get('QUEUE_DEST_PIPELINE_TRIGGER')
         self.data_directory = os.environ.get('DATA_DIRECTORY', '/storage/data/')
         # if mode is consume-and-publish the data_index file will be used for consume and will have a suffix ".published"
         self.data_index = os.environ.get('DATA_INDEX', '/storage/index')
@@ -43,9 +43,9 @@ class Config:
         self.__assert_variable_is_defined(self.mq_user, "User not specified in MQ_USER")
         self.__assert_variable_is_defined(self.mq_password, "Password not specified in MQ_PASSWORD")
         if self.is_start_publish():
-            self.__assert_variable_is_defined(self.queue_dest, "Destination queue is not specified in QUEUE_DEST")
-        if self.is_start_consume():
-            self.__assert_variable_is_defined(self.queue_src, "Source queue is not specified in QUEUE_SRC")
+            self.__assert_variable_is_defined(self.queue_dest_test_trigger, "Destination queue is not specified in QUEUE_DEST_TEST_TRIGGER")
+            self.__assert_variable_is_defined(self.queue_dest_pipeline_trigger, "Destination queue is not specified in QUEUE_DEST_PIPELINE_TRIGGER")
+
 
     def display(self) -> None:
         for name in sorted(dir(self)):
@@ -58,8 +58,6 @@ class Config:
                 print(f"{name.upper()}={getattr(self, name)}")
         print()
 
-    def is_start_consume(self) -> bool:
-        return self.queue_mode in [MODE_CONSUME, MODE_CONSUME_AND_PUBLISH, ]
 
     def is_start_publish(self) -> bool:
         return self.queue_mode in [MODE_PUBLISH, MODE_CONSUME_AND_PUBLISH, ]

src/project/demo_thread.py

@@ -1,48 +0,0 @@
-import ast
-from typing import Callable, Union
-
-from project.adapters import InAdapter, OutAdapter
-from wrapper import HealthCheck
-from . import Config
-
-
-class DemoThread:
-    def __init__(
-            self,
-            config: Config,
-            health_check: Union[HealthCheck, None],
-            in_adapter_injector: Callable[[Config], InAdapter],
-            out_adapter_injector: Callable[[Config], OutAdapter]
-    ) -> None:
-        """
-        Create BoilerplateRemovalThread instance.
-        :param config: configuration review_options
-        :param health_check: health check
-        :param in_adapter_injector: input adapter, based on config
-        :param out_adapter_injector: output adapter, based on config
-        """
-        self.__config = config
-        self.__health_check = health_check
-        # initialize queues
-        self.__queue_src = in_adapter_injector(config)
-        self.__queue_dest = out_adapter_injector(config)
-
-    def handle_element(self, ch, method, properties, url: bytes) -> None:
-        print(f"CONSUME: Received element {properties}: {len(url)}")
-
-        try:
-
-            self.__queue_dest.send_message(bytes(str(url), encoding="utf-8"))
-            print(f"CONSUME: Send element {properties}")
-
-        except Exception as e:
-            print("Error", e)
-
-    def run(self) -> None:
-        # consume elements
-        # TODO: are these callbacks called only one at a time or in multiple threads?
-        # self.__handle_element("ch", "method", "properties", bytes("localhost/health", "utf-8"))
-        # self.__handle_element("ch", "method", "properties", bytes("localhost/health", "utf-8"))
-        self.__queue_src.start_consuming(self.handle_element)
-        # print("CONSUME iteration")
-        # time.sleep(1)

src/services/__init__.py

@@ -0,0 +1,2 @@
+from .db_service import init_connection
+from .mq_service import create_out_adapter,PrintingOutAdapter

src/services/db_service.py

@@ -0,0 +1,13 @@
+import os
+
+import streamlit as st
+from dotenv import load_dotenv
+from pymongo import MongoClient
+
+
+@st.cache_resource
+def init_connection():
+    load_dotenv()
+    uri = f"mongodb+srv://{os.getenv('MONGO_USERNAME')}:{os.getenv('MONGO_PASSWORD')}@{os.getenv('MONGO_HOST')}/?retryWrites=true&w=majority&appName=Cluster0"
+    client = MongoClient(uri)
+    return client.event_data

src/services/mq_service.py

@@ -0,0 +1,17 @@
+import streamlit as st
+
+from project import Config
+from project.adapters import MessageQueueOutAdapterImpl, OutAdapter
+
+
+class PrintingOutAdapter(OutAdapter):
+    def send_message(self, message):
+        print(f"Publishing message {len(message)}")
+
+
+@st.cache_resource
+def create_out_adapter(_config: Config, queue_dest: str) -> OutAdapter:
+    if _config.mq_user == "local":
+        return PrintingOutAdapter()
+    else:
+        return MessageQueueOutAdapterImpl(_config, queue_dest)

src/services/test_creator.py

@@ -0,0 +1,23 @@
+import datetime
+
+from bson import ObjectId, DatetimeMS
+
+
+class TestCreator:
+    def __init__(self, db):
+        self.db = db
+        self.test_obj = None
+
+    def create(self, record_ids: list[str]) -> ObjectId:
+        record_results = {}
+        for batch_id in record_ids:
+            record_results[batch_id] = {"status": "pending"}
+
+        test_obj = {
+            "status": "pending",
+            "created_at": datetime.datetime.now(datetime.UTC),
+            "pipeline_version": "1.0.0",
+            "record_results": record_results
+        }
+        insert_result = self.db.test_evaluation.insert_one(test_obj)
+        return insert_result.inserted_id

src/wrapper/__init__.py

@@ -1,2 +1 @@
-from .health_check import HealthCheck
 from .message_queue import MessageQueue

src/wrapper/health_check.py

@@ -1,73 +0,0 @@
-import logging
-from typing import Callable, Dict
-
-from flask import Flask
-from flask.typing import ResponseValue
-
-
-# adapted from https://stackoverflow.com/questions/62000942/flask-block-specific-endpoints-from-logging
-class AjaxFilter(logging.Filter):
-    def filter(self, record):
-        return "/health" not in record.getMessage()
-
-
-class HealthCheck:
-    # adapted from https://thelinuxnotes.com/index.php/implementing-a-flask-health-check-and-kubernetes-liveness-probe-in-python-application/
-    def __do_check(self) -> tuple[ResponseValue, int]:
-        # https://docs.python.org/3.9/library/functions.html#all
-        # https://stackoverflow.com/questions/12229064/mapping-over-values-in-a-python-dictionary
-        # loop over checks
-        # execute check method for each key
-        results = {k: v() for k, v in self.__checks.items()}
-        overall_result = all(results.values())
-        json = {
-            "state": "OK" if overall_result else "ERROR",
-            "elements": results
-        }
-        status_code = 200 if overall_result else 500
-        return json, status_code
-
-    def __init__(self, app: Flask) -> None:
-        self.__app = app
-        self.__checks: Dict[str, Callable[[], bool]] = {}
-        self.__log = logging.getLogger('werkzeug')
-        self.disable_logging_filter = AjaxFilter()
-
-        # register route dynamically
-        # https://pytutorial.com/flask-appadd_url_rule-dynamic-url-registration-guide/
-        # @app.route('/health')
-        # @app.route('/health/')
-        self.__app.add_url_rule(
-            '/health',  # URL rule with variable
-            'health_check',  # Endpoint name
-            view_func=self.__do_check,  # View function
-            methods=['GET'],  # Allowed methods
-            strict_slashes=False  # URL trailing slash handling
-        )
-
-    def disable_logging(self) -> None:
-        print("health check logging disabled")
-        self.__log.addFilter(self.disable_logging_filter)
-
-    def enable_logging(self) -> None:
-        print("health check logging enabled")
-        self.__log.removeFilter(self.disable_logging_filter)
-
-    def add_check(self, name, method: Callable[[], bool]) -> None:
-        """method to add named check methods to health check
-        Parameters
-        ----------
-        name : str
-            named key of check method
-        method : function
-            method which returns a bool for the health check
-        """
-        self.__checks.update({f"{name}": method})
-
-    def wrap_exception_handler(self, name: str, method: Callable[[], None]) -> None:
-        """wrap exception handler around method and register simple health check for it"""
-        self.add_check(name, lambda: True)
-        try:
-            method()
-        finally:
-            self.add_check(name, lambda: False)

src/wrapper/message_queue.py

@@ -42,6 +42,7 @@ class MessageQueue:
                 properties=pika.BasicProperties(
                     delivery_mode=pika.DeliveryMode.Persistent
                 ))
+            print(f"[PUBLISH] Queued message in {self.__queue_name}")
 
         # try repeating once, if connection was lost
         retry = True
@@ -53,6 +54,7 @@ class MessageQueue:
                 print("Reconnecting...")
                 if retry:
                     self.connect()
+                    __send()
                     retry = False
                 else:
                     raise e
@@ -60,21 +62,13 @@ class MessageQueue:
         # connection.close()
 
     def __ack_and_call_callback(
-            self,
-            ch: BlockingChannel,
-            method: spec.Basic.Deliver,
-            properties: spec.BasicProperties,
-            body: bytes) -> None:
+        self,
+        ch: BlockingChannel,
+        method: spec.Basic.Deliver,
+        properties: spec.BasicProperties,
+        body: bytes) -> None:
         # acknowledge processing directly
         ch.basic_ack(delivery_tag=method.delivery_tag)
         # print(f" [x] Received {body.decode()}")
         self.__callback(ch, method, properties, body)
         # print(" [x] Done")
-
-    def start_consuming(self, callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]) -> None:
-        """callback method parameter is a method with the following signature:
-        def callback(ch, method, properties, body):"""
-        self.__callback = callback
-        self.__channel.basic_qos(prefetch_count=1)
-        self.__channel.basic_consume(queue=self.__queue_name, on_message_callback=self.__ack_and_call_callback)
-        self.__channel.start_consuming()