diff --git a/.gitignore b/.gitignore index 9f21b54..123e6b8 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ /venv/ +*.pyc \ No newline at end of file diff --git a/backend_model.py b/backend_model.py new file mode 100644 index 0000000..475af6c --- /dev/null +++ b/backend_model.py @@ -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() diff --git a/frontend.py b/frontend.py index a5c4903..b7344fe 100644 --- a/frontend.py +++ b/frontend.py @@ -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()