Task: storage service rework (#30)

k1nho · web-flow · commit cc4fe6811577 · 2025-04-30T10:52:10.000-04:00
- Add finish_campaign endpoint to service (signal end of campaign)
- upload stateful file when end of campaign is signaled
diff --git a/Dockerfile.storage_service b/Dockerfile.storage_service
@@ -3,8 +3,9 @@ FROM python:3.10-slim
 
 WORKDIR /usr/src/storage_service
 
-RUN python -m pip install python-dotenv boto3==1.35.99
+RUN python -m pip install python-dotenv boto3==1.35.99 pyyaml
 
+COPY ./config/config_storage.yaml /app/config_storage.yaml
 COPY ./services/storage_service.py ./storage_service.py
 COPY ./services/constants.py ./constants.py
 
diff --git a/clients/realtime_client.py b/clients/realtime_client.py
@@ -21,7 +21,7 @@
     default_intersect_lifecycle_loop,
 )
 
-from schema import FileType, TransitionData, NextTemperature
+from schema import FileType, TransitionData, NextTemperature, FinishCampaignMsg
 
 logging.basicConfig(level=logging.INFO)
 
@@ -71,7 +71,28 @@ class SampleOrchestrator:
     def __init__(self) -> None:
         """ "Load all gsa files to simulate a stream of data coming in"""
         self.message_stack = []
-        id_campaign = uuid4()
+        id_campaign = str(uuid4())
+
+        # bragg data
+        with open(DATA_PATH) as f:
+            for line in f:
+                filepath = line.strip()
+                with open(filepath, "rb") as file:
+                    msg = FileType(
+                        filename=os.path.basename(filepath),
+                        file=base64.b64encode(file.read()),
+                    )
+                    # wait 5 seconds for each message
+                    self.message_stack.append(
+                        (
+                            IntersectDirectMessageParams(
+                                destination="nsdf.cloud.diffraction.dashboard.dashboard-service",
+                                operation="NSDFDashboard.get_bragg_data",
+                                payload=msg,
+                            ),
+                            2.0,
+                        )
+                    )
         for i in range(len(transition_data)):
             data = transition_data[i]
             nextpoint = (
@@ -87,7 +108,7 @@ def __init__(self) -> None:
                         destination="nsdf.cloud.diffraction.dashboard.dashboard-service",
                         operation="NSDFDashboard.get_transition_data_single",
                         payload=TransitionData(
-                            id=str(id_campaign),
+                            id=id_campaign,
                             temp=data[0],
                             ylist=data[1:],
                         ),
@@ -102,35 +123,27 @@ def __init__(self) -> None:
                         destination="nsdf.cloud.diffraction.dashboard.dashboard-service",
                         operation="NSDFDashboard.get_next_temperature",
                         payload=NextTemperature(
-                            id=str(id_campaign),
+                            id=id_campaign,
                             data=nextpoint,
                             timestamp=int(time.time()),
                         ),
                     ),
                     2.0,
                 )
             )
-
-        # bragg data
-        with open(DATA_PATH) as f:
-            for line in f:
-                filepath = line.strip()
-                with open(filepath, "rb") as file:
-                    msg = FileType(
-                        filename=os.path.basename(filepath),
-                        file=base64.b64encode(file.read()),
-                    )
-                    # wait 5 seconds for each message
-                    self.message_stack.append(
-                        (
-                            IntersectDirectMessageParams(
-                                destination="nsdf.cloud.diffraction.dashboard.dashboard-service",
-                                operation="NSDFDashboard.get_bragg_data",
-                                payload=msg,
-                            ),
-                            2.0,
-                        )
-                    )
+        # finish campaign
+        self.message_stack.append(
+            (
+                IntersectDirectMessageParams(
+                    destination="nsdf.cloud.diffraction.dashboard.dashboard-service",
+                    operation="NSDFDashboard.finish_campaign",
+                    payload=FinishCampaignMsg(
+                        id=id_campaign,
+                    ),
+                ),
+                2.0,
+            )
+        )
         self.message_stack.reverse()
 
     def client_callback(
diff --git a/clients/schema.py b/clients/schema.py
@@ -47,3 +47,14 @@ class NextTemperature(BaseModel):
     id: str
     data: float
     timestamp: int
+
+
+class FinishCampaignMsg(BaseModel):
+    """
+    Represents a message to signal the end of a campaign
+
+    Attributes:
+        id(str): The campaign id to finish
+    """
+
+    id: str
diff --git a/clients/short_list_of_data.txt b/clients/short_list_of_data.txt
@@ -4,33 +4,3 @@
 ./GSAS/NOM168366tof.gsa
 ./GSAS/NOM168367tof.gsa
 ./GSAS/NOM168368tof.gsa
-./GSAS/NOM168369tof.gsa
-./GSAS/NOM168370tof.gsa
-./GSAS/NOM168371tof.gsa
-./GSAS/NOM168372tof.gsa
-./GSAS/NOM168373tof.gsa
-./GSAS/NOM168374tof.gsa
-./GSAS/NOM168375tof.gsa
-./GSAS/NOM168376tof.gsa
-./GSAS/NOM168377tof.gsa
-./GSAS/NOM168378tof.gsa
-./GSAS/NOM168379tof.gsa
-./GSAS/NOM168380tof.gsa
-./GSAS/NOM168381tof.gsa
-./GSAS/NOM168382tof.gsa
-./GSAS/NOM168383tof.gsa
-./GSAS/NOM168384tof.gsa
-./GSAS/NOM168385tof.gsa
-./GSAS/NOM168386tof.gsa
-./GSAS/NOM168387tof.gsa
-./GSAS/NOM168388tof.gsa
-./GSAS/NOM168389tof.gsa
-./GSAS/NOM168390tof.gsa
-./GSAS/NOM168391tof.gsa
-./GSAS/NOM168392tof.gsa
-./GSAS/NOM168393tof.gsa
-./GSAS/NOM168394tof.gsa
-./GSAS/NOM168395tof.gsa
-./GSAS/NOM168396tof.gsa
-./GSAS/NOM168397tof.gsa
-./GSAS/NOM168398tof.gsa
diff --git a/clients/transition_client.py b/clients/transition_client.py
@@ -23,7 +23,7 @@
     default_intersect_lifecycle_loop,
 )
 
-from schema import TransitionData, NextTemperature
+from schema import TransitionData, NextTemperature, FinishCampaignMsg
 
 logging.basicConfig(level=logging.INFO)
 
@@ -145,6 +145,19 @@ def generate_campaign(
         campaign.append(generate_transition_msg(id, temps[i], ylen))
         campaign.append(generate_next_temp_msg(id, next_temp))
 
+    campaign.append(
+        (
+            IntersectDirectMessageParams(
+                destination="nsdf.cloud.diffraction.dashboard.dashboard-service",
+                operation="NSDFDashboard.finish_campaign",
+                payload=FinishCampaignMsg(
+                    id=id,
+                ),
+            ),
+            2.0,
+        )
+    )
+
     return campaign
 
 
diff --git a/compose.yaml b/compose.yaml
@@ -1,9 +1,10 @@
-# Simple docker-compose file which sets up the broker, dashboard, and dashboard service
+# Simple docker-compose file which sets up the broker, dashboard, dashboard service, and storage service
 # Note that for the broker, the default credentials are ` intersect_username / intersect_password `
 
 services:
   broker:
     image: "bitnami/rabbitmq:3.13.3"
+    container_name: rabbitmq
     ports:
       - "1883:1883" # MQTT port
       - "5672:5672" # AMQP port
@@ -24,47 +25,52 @@ services:
 
   dashboard:
     image: "ghcr.io/nsdf-fabric/intersect-dashboard:latest"
+    container_name: dashboard
     ports:
       - "10042:10042"
     depends_on:
       - broker
     environment:
       BOKEH_ALLOW_WS_ORIGIN: "*"
+      INTERSECT_DASHBOARD_CONFIG: "/app/config_dashboard_default.yaml"
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:10042/"]
       interval: 10s
       retries: 5
       start_period: 10s
       timeout: 10s
     volumes:
-      - intersect_bragg_data:/usr/src/dashboard/bragg_data/
-      - intersect_transition_data:/usr/src/dashboard/transition_data/
-      - intersect_next_temperature_data:/usr/src/dashboard/next_temperature_data/
+      - intersect_bragg_volume:/usr/src/dashboard/bragg_volume/
+      - intersect_transition_volume:/usr/src/dashboard/transition_volume/
+      - intersect_andie_volume:/usr/src/dashboard/andie_volume/
       - intersect_scientist_cloud_volume:/usr/src/dashboard/scientist_cloud_volume
   dashboard_service:
     image: "ghcr.io/nsdf-fabric/intersect-service:latest"
+    container_name: dashboard_service
     ports:
       - "10043:10043"
+    environment:
+      INTERSECT_SERVICE_CONFIG: "/app/config_default.yaml"
     depends_on:
       broker:
         condition: service_healthy
     volumes:
-      - intersect_bragg_data:/usr/src/dashboard_service/bragg_data/
-      - intersect_transition_data:/usr/src/dashboard_service/transition_data/
-      - intersect_next_temperature_data:/usr/src/dashboard_service/next_temperature_data/
+      - intersect_bragg_volume:/usr/src/dashboard_service/bragg_volume/
+      - intersect_transition_volume:/usr/src/dashboard_service/transition_volume/
+      - intersect_andie_volume:/usr/src/dashboard_service/andie_volume/
       - intersect_scientist_cloud_volume:/usr/src/dashboard_service/scientist_cloud_volume
   storage_service:
     image: "ghcr.io/nsdf-fabric/intersect-storage:latest"
-    env_file:
-      - .env
+    container_name: storage_service
     ports:
       - "10044:10044"
+    environment:
+      INTERSECT_SERVICE_CONFIG: "/app/config_storage.yaml"
+    env_file: ".env"
     volumes:
       - intersect_scientist_cloud_volume:/usr/src/storage_service/scientist_cloud_volume
-      - intersect_retry_volume:/usr/src/storage_service/retry_volume
 volumes:
-  intersect_bragg_data:
-  intersect_transition_data:
-  intersect_next_temperature_data:
+  intersect_bragg_volume:
+  intersect_transition_volume:
+  intersect_andie_volume:
   intersect_scientist_cloud_volume:
-  intersect_retry_volume:
diff --git a/config/config_storage.yaml b/config/config_storage.yaml
@@ -0,0 +1,8 @@
+volumes:
+  scientist_cloud_volume: "scientist_cloud_volume"
+scan_period: 30
+sci_cloud:
+  bucket_prefix: "utk"
+  bragg_prefix: "bragg"
+  transition_prefix: "transition"
+  andie_prefix: "andie"
diff --git a/services/constants.py b/services/constants.py
@@ -23,8 +23,8 @@
 NEXT_TEMPERATURE_DATA_PATH = "andie_data"
 
 # DASHBOARD
-MAX_BANKS = 6
 INTERSECT_DASHBOARD_CONFIG = "INTERSECT_DASHBOARD_CONFIG"  # env variable holds the path to dashboard file configuration
-
 # SERVICE
 INTERSECT_SERVICE_CONFIG = "INTERSECT_SERVICE_CONFIG"  # env variable holds the path to service file configuration
+# STORAGE
+INTERSECT_STORAGE_CONFIG = "INTERSECT_STORAGE_CONFIG"  # env variable holds the path to the storage file configuration
diff --git a/services/dashboard.py b/services/dashboard.py
@@ -16,10 +16,7 @@
 import numpy as np
 import yaml
 from gsa_loader import load_gsa_file
-from constants import (
-    MAX_BANKS,
-    INTERSECT_DASHBOARD_CONFIG
-)
+from constants import INTERSECT_DASHBOARD_CONFIG
 
 
 logger = logging.getLogger(__name__)
@@ -54,22 +51,11 @@ def __init__(self):
         self.bragg_data_dict = dict(
             data=[],
             layout=go.Layout(
-                title=dict(text="Bragg Data", font=dict(size=22, weight="bold")),
-                xaxis=dict(title=dict(text="d-Spacing", font=dict(size=16)), tickfont=dict(size=14)),
-                yaxis=dict(title=dict(text="Intensity", font=dict(size=16)), tickfont=dict(size=14)),
+                title=dict(text="Bragg Data", font=dict(size=26, weight="bold")),
+                xaxis=dict(title=dict(text="d-Spacing", font=dict(size=22)), tickfont=dict(size=18)),
+                yaxis=dict(title=dict(text="Intensity", font=dict(size=22)), tickfont=dict(size=18)),
             ),
         )
-        self.bragg_data_by_bank_dict = [
-            dict(
-                data=[],
-                layout=go.Layout(
-                    title=dict(text=f"Bank {i+1}", font=dict(size=22, weight="bold")),
-                    xaxis=dict(title=dict(text="d-Spacing", font=dict(size=16)), tickfont=dict(size=14)),
-                    yaxis=dict(title=dict(text="Intensity", font=dict(size=16)), tickfont=dict(size=14)),
-                ),
-            )
-            for i in range(MAX_BANKS)
-        ]
         self.transition_data_dict = dict(
             data=[
                 go.Scatter(
@@ -81,20 +67,20 @@ def __init__(self):
                 )
             ],
             layout=go.Layout(
-                title=dict(text="Transition Plot", font=dict(size=22, weight="bold")),
-                xaxis=dict(title=dict(text="Temperature (K)", font=dict(size=16)), tickfont=dict(size=14)),
-                yaxis=dict(title=dict(text="d-Spacing", font=dict(size=16)), tickfont=dict(size=14)),
-                legend=dict(font=dict(size=16))
+                title=dict(text="Transition Plot", font=dict(size=26, weight="bold")),
+                xaxis=dict(title=dict(text="Temperature (K)", font=dict(size=22)), tickfont=dict(size=18)),
+                yaxis=dict(title=dict(text="d-Spacing", font=dict(size=22)), tickfont=dict(size=18)),
+                legend=dict(font=dict(size=22))
             ),
         )
         self.stateful_plot_data_dict = dict(
             data=[],
             layout=go.Layout(
                 title=dict(
-                    text="Bragg Data Stateful Plot", font=dict(size=22, weight="bold")
+                    text="Bragg Data Stateful Plot", font=dict(size=26, weight="bold")
                 ),
-                xaxis=dict(title=dict(text="d-Spacing", font=dict(size=16)), tickfont=dict(size=14)),
-                yaxis=dict(title=dict(text="Intensity", font=dict(size=16)), tickfont=dict(size=14)),
+                xaxis=dict(title=dict(text="d-Spacing", font=dict(size=22)), tickfont=dict(size=18)),
+                yaxis=dict(title=dict(text="Intensity", font=dict(size=22)), tickfont=dict(size=18)),
 
             ),
         )
@@ -116,7 +102,7 @@ def __init__(self):
 
         self.by_bank_tab = pn.Column(
             pn.pane.Markdown("<h1>By Bank</h1>"),
-            pn.GridBox(*self.bragg_data_by_bank_plots, ncols=3))
+            pn.FlexBox(*self.bragg_data_by_bank_plots))
 
         self.select_bragg_file = pn.widgets.AutocompleteInput(
             name="Bragg File Timestamp",
@@ -296,10 +282,10 @@ def _render_bragg_plot(self):
                     pn.pane.Plotly(dict(
                         data=scatter_line,
                         layout=go.Layout(
-                            title=dict(text=f"Bank {wksp_index}", font=dict(size=22, weight="bold")),
-                            xaxis=dict(title="d-Spacing"),
-                            yaxis=dict(title="Intensity"))
-                    )))
+                            title=dict(text=f"Bank {wksp_index}", font=dict(size=26, weight="bold")),
+                            xaxis=dict(title=dict(text="d-Spacing", font=dict(size=22)), tickfont=dict(size=18)),
+                            yaxis=dict(title=dict(text="Intensity", font=dict(size=22)), tickfont=dict(size=18)),)
+                    ), sizing_mode="stretch_both"))
 
                 traces.append(scatter_line)
 
@@ -584,7 +570,7 @@ def App() -> MaterialTemplate:
     )
     pn.state.add_periodic_callback(
         callback=app_state.poll_transition_and_andie,
-        period=app_state.config['scan_period']['transition_scan_period']  * 1000,
+        period=app_state.config['scan_period']['transition_scan_period'] * 1000,
     )
 
     pn.state.add_periodic_callback(
diff --git a/services/dashboard_service.py b/services/dashboard_service.py
diff --git a/services/storage_service.py b/services/storage_service.py
diff --git a/tests/unit/test_dashboard.py b/tests/unit/test_dashboard.py
