auracaster/auracaster-webui
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement a basic suitable ui with a model of what could happen in the backend.

Browse Source
This commit is contained in:
pstruebi
2025-03-10 14:19:17 +01:00
parent a2c8ce1c03
commit acb5046ccd
3 changed files with 152 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1 +1,2 @@
/venv/
*.pyc

89
backend_model.py Normal file
View File

@@ -0,0 +1,89 @@
import time
import threading
from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional, List
class AnnouncementStates(Enum):
IDLE: str = "Ready"
TRANSLATING: str = "Translating"
GENERATING_VOICE: str = "Generating voice synthesis"
ROUTING: str = "Routing to endpoints"
ACTIVE: str = "Broadcasting announcement"
COMPLETE: str = "Complete"
ERROR: str = "Error"
class AnnouncementDetails(BaseModel):
text: str
languages: List[str]
group: str
endpoints: List[str]
start_time: float
class AnnouncementProgress(BaseModel):
current_state: str = AnnouncementStates.IDLE
error: Optional[str] = None
details: AnnouncementDetails
progress: float = Field(default=0.0, ge=0.0, le=1.0)
description: str = Field(default="Ready")
class AnnouncementSystem:
def __init__(self):
self.current_process = AnnouncementProgress(
details=AnnouncementDetails(
text="",
languages=[],
endpoints=[],
group="",
start_time=0.0
)
)
self._thread = None
def start_announcement(self, text, group):
if self.current_process.current_state not in [AnnouncementStates.IDLE, AnnouncementStates.COMPLETE]:
raise Exception("Announcement already in progress")
self.current_process.details.text = text
self.current_process.details.group = group
self._thread = threading.Thread(target=self._run_process)
self._thread.start()
def _run_process(self):
self.current_process.details.start_time = time.time()
try:
self._update_state(AnnouncementStates.TRANSLATING)
time.sleep(2) # Simulate translation
self._update_state(AnnouncementStates.GENERATING_VOICE)
time.sleep(1.5) # Voice synthesis
self._update_state(AnnouncementStates.ROUTING)
time.sleep(len(self.current_process.details.endpoints) * 0.5)
self._update_state(AnnouncementStates.ACTIVE)
time.sleep(3) # Simulate broadcast
self._update_state(AnnouncementStates.COMPLETE)
except Exception as e:
self.current_process.error = str(e)
self._update_state(AnnouncementStates.ERROR)
def _update_state(self, new_state):
self.current_process.current_state = new_state
# Progress based on state transitions
state_progress = {
AnnouncementStates.TRANSLATING: 0.25,
AnnouncementStates.GENERATING_VOICE: 0.5,
AnnouncementStates.ROUTING: 0.75,
AnnouncementStates.ACTIVE: 1.0,
AnnouncementStates.COMPLETE: 1.0,
AnnouncementStates.ERROR: 1.0
}
self.current_process.progress = state_progress[new_state]
# Singleton instance
announcement_system = AnnouncementSystem()

99
frontend.py
View File

@@ -19,20 +19,48 @@ if "endpoint_groups" not in st.session_state:
}
]
# Mock API call function
def announce(text: str, gate: str):
"""Mock function that would make an API call"""
# Find the selected group
group = next(g for g in st.session_state.endpoint_groups if g["name"] == gate)
st.write(f"📢 Announcement made to group {gate}:")
st.write(f"📡 Endpoints: {', '.join(group['endpoints'])}")
st.write(f"🗣️ '{text}'")
st.write(f"🌐 Languages: {', '.join(group['languages'])}")
st.write("(This is a mock - real implementation would:")
st.write("- Call airport PA system API")
st.write(f"- Make announcements in {len(group['languages'])} languages")
st.write("- Handle endpoint group routing)")
from backend_model import announcement_system, AnnouncementStates
import time
def show_announcement_status():
if announcement_system.current_process.current_state != AnnouncementStates.IDLE:
status = st.status("**Airport PA System Status**", expanded=True)
with status:
# Progress elements
progress_bar = st.progress(announcement_system.current_process.progress)
time_col, stage_col = st.columns([1, 3])
# Track last displayed state
last_state = None
# Update loop
while announcement_system.current_process.current_state not in [AnnouncementStates.COMPLETE, AnnouncementStates.ERROR]:
# Update time elapsed continuously
# Only update stage display if state changed
if announcement_system.current_process.current_state != last_state:
stage_col.write(f"**Stage:** {announcement_system.current_process.current_state.value}")
#stage_col.write(f"🌐 Languages: {', '.join(announcement_system.current_process.details.languages)}")
last_state = announcement_system.current_process.current_state
time_col.write(f"⏱️ Time elapsed: {time.time() - announcement_system.current_process.details.start_time:.1f}s")
# Update progress bar
progress_bar.progress(announcement_system.current_process.progress)
time.sleep(0.3)
# Final state
if announcement_system.current_process.current_state == AnnouncementStates.ERROR:
st.error(f"❌ Error: {announcement_system.current_process.error}")
else:
st.success("✅ Announcement completed successfully")
st.write(f"📢 Announcement made to group {group['name']}:")
st.write(f"📡 Endpoints: {', '.join(group['endpoints'])}")
st.write(f"🗣️ '{announcement_system.current_process.details.text}'")
st.write(f"🌐 Languages: {', '.join(group['languages'])}")
# Reset status after completion
announcement_system.current_process.current_state = AnnouncementStates.IDLE
# Page setup
st.set_page_config(page_title="Airport Announcement System", page_icon="✈️")
@@ -48,22 +76,25 @@ with st.container():
col1, col2, col3 = st.columns(3)
with col1:
if st.button("Final Boarding Call"):
st.session_state.last_announcement = {
"message": "This is the final boarding call for flight LX-380 to New York",
"group": "Gate1"
}
group = next(g for g in st.session_state.endpoint_groups if g["name"] == "Gate1")
announcement_system.start_announcement(
"This is the final boarding call for flight LX-380 to New York",
group
)
with col2:
if st.button("Security Reminder"):
st.session_state.last_announcement = {
"message": "Please keep your luggage with you at all times",
"group": "Gate2"
}
group = next(g for g in st.session_state.endpoint_groups if g["name"] == "Gate2")
announcement_system.start_announcement(
"Please keep your luggage with you at all times",
group
)
with col3:
if st.button("Delay Notice"):
st.session_state.last_announcement = {
"message": "We regret to inform you of a 30-minute delay",
"group": "Gate1"
}
group = next(g for g in st.session_state.endpoint_groups if g["name"] == "Gate1")
announcement_system.start_announcement(
"We regret to inform you of a 30-minute delay",
group
)
# Custom announcement
with st.form("custom_announcement"):
@@ -73,10 +104,8 @@ with st.container():
message = st.text_area("Enter announcement text", "")
if st.form_submit_button("Make Announcement"):
st.session_state.last_announcement = {
"message": message,
"group": selected_group
}
group = next(g for g in st.session_state.endpoint_groups if g["name"] == selected_group)
announcement_system.start_announcement(message, group)
# Configuration section in sidebar
with st.sidebar:
@@ -117,10 +146,6 @@ with st.sidebar:
"languages": DEFAULT_LANGUAGES.copy()
})
# Display announcement feedback at the bottom of the page
if "last_announcement" in st.session_state:
st.write("---")
announce(
st.session_state.last_announcement["message"],
st.session_state.last_announcement["group"]
)
# Display announcement status
st.write("---")
show_announcement_status()