Laurenz/gleiswechsel
1
0
Fork 0
mirror of https://github.com/kaaninchen/Gleiswechsel.git synced 2026-09-17 16:52:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

transitous rewrite: only send announcement embed when no voice announcement

Browse Source
This commit is contained in:
Kaaninchen
2026-08-17 17:14:43 +02:00
parent f990a55da9
commit 60d4936dda
5 changed files with 66 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/transitous.py
+2 -2
View File
@@ -120,8 +120,8 @@ def get_random_stop_id() -> str | None:
stop_ids_list = list(stop_ids.keys())
stops_string = ", ".join(stop_ids.values())
if len(stop_ids_list) > 1:
logger(f"No station associated as '{assigned_station}', choosing random from similar named stations:")
logger({stops_string})
logger(f"No station associated as '{assigned_station}', choosing random from similar named station")
logger(f"Run `python run main.py stations` to get exact station names")
chosen_stop_id = random.choice(stop_ids_list)
return chosen_stop_id
src/data/operators.py
+1 -1
View File
@@ -108,7 +108,7 @@ OPERATORS = {
"color": 0x00B451
},
"GVB": {
"logo": "https://upload.wikimedia.org/wikipedia/commons/thumb/a/a3/GVB_Amsterdam_Logo_001.svg/1280px-GVB_Amsterdam_Logo_001.svg.png",
"logo": "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/BSicon_LOGO_GVB.svg/960px-BSicon_LOGO_GVB.svg.png",
"color": 0x2B62AF
}
}
src/dc/embeds.py
+2 -2
View File
@@ -23,7 +23,7 @@ def build_info_embed() -> discord.Embed:
from src.dc.handlers import trip
agency = trip["agency"]
metadata = get_operator_metadata(agency, trip["route_color"])
metadata = get_operator_metadata(agency, trip["route_color"], trip["mode"])
departure = format_timestamp_to_dc(trip["departure"])
arrival = format_timestamp_to_dc(trip["arrival"])
@@ -62,7 +62,7 @@ def build_info_embed() -> discord.Embed:
def build_announcement_embed(msg):
from src.dc.handlers import trip
agency = trip["agency"]
metadata = get_operator_metadata(agency, trip["route_color"])
metadata = get_operator_metadata(agency, trip["route_color"], trip["mode"])
embed = discord.Embed(
title = "Informationen zu ihrer Fahrt",
src/dc/handlers.py
+52 -47
View File
@@ -3,7 +3,7 @@ import asyncio
import random
from datetime import datetime, timedelta, date
from src.utils import logger, channel_formatting, choose_connection, get_sound_path, LOCAL_TZ
from src.utils import logger, channel_formatting, choose_connection, get_sound_path, LOCAL_TZ, get_next_station
from src.config import config
_scheduled_task: asyncio.Task | None = None
@@ -43,6 +43,55 @@ async def rename_vc(bot: discord.Bot, voice_channel, from_scheduler: bool = Fals
_scheduled_task = asyncio.create_task(_schedule_next_transfer(bot, trip["arrival_dt"], voice_channel, trip["to"]))
async def announcer(announcement: str, voice_channel: discord.VoiceChannel, destination = None):
from src.dc.embeds import build_info_embed, build_announcement_embed
announcements_enabled = config.announcements.enabled
voice_announcement_enabled = config.announcements.voice[0].enabled
if announcements_enabled:
if len(voice_channel.members) > 0:
match announcement:
case "ende":
if voice_announcement_enabled:
announcement_status = await voice_announcer(destination, voice_channel)
if announcement_status:
return
embed = build_announcement_embed(
f'Sehr geehrte Fahrgäste,\nIn wenigen Minuten erreichen wir {destination}. Dieser Zug endet dort.\n\nWir wünschen Ihnen eine angenehme Weiterreise.\n\nVielen Dank für ihr Vertrauen und auf Wiedersehen.')
case "umstieg":
embed = build_info_embed()
case _:
logger(f"Unbekanntes Announcements: {announcement}")
embed = None
if embed:
await voice_channel.send(embed=embed)
else:
logger(f"Announcement {announcement} wird geskipped, keiner da")
return
async def voice_announcer(destination: str, voice_channel: discord.VoiceChannel) -> bool:
sound_path = get_sound_path(destination=destination)
if sound_path is None:
return False
logger(f"VC wird betreten, spiele {sound_path}")
vc = await voice_channel.connect(timeout=15, reconnect=True)
audio_source = discord.FFmpegPCMAudio(sound_path)
loop = asyncio.get_running_loop()
if not vc.is_playing():
def after_playing(error):
if error:
logger(f"Player error: {error}", "error")
loop.create_task(vc.disconnect())
logger("VC wird verlassen")
vc.play(audio_source, after=after_playing)
return True
async def _schedule_next_transfer(bot: discord.Bot, arrival_dt: datetime, voice_channel: discord.VoiceChannel, destination: str):
now = datetime.now(LOCAL_TZ)
@@ -62,49 +111,5 @@ async def _schedule_next_transfer(bot: discord.Bot, arrival_dt: datetime, voice_
else:
await asyncio.sleep(wait_seconds)
logger("Zug angekommen, wähle neue Verbindung")
await rename_vc(bot, voice_channel, from_scheduler=True)
async def announcer(announcement: str, voice_channel: discord.VoiceChannel, destination = None):
from src.dc.embeds import build_info_embed, build_announcement_embed
announcements_enabled = config.announcements.enabled
voice_announcement_enabled = config.announcements.voice[0].enabled
if announcements_enabled:
if len(voice_channel.members) > 0:
match announcement:
case "ende":
embed = build_announcement_embed(
f'Sehr geehrte Fahrgäste,\nIn wenigen Minuten erreichen wir {destination}. Dieser Zug endet dort.\n\nWir wünschen Ihnen eine angenehme Weiterreise.\n\nVielen Dank für ihr Vertrauen und auf Wiedersehen.')
if voice_announcement_enabled:
await voice_announcer(destination, voice_channel)
case "umstieg":
embed = build_info_embed()
case _:
logger(f"Unbekanntes Announcements: {announcement}")
embed = None
if embed:
await voice_channel.send(embed=embed)
else:
logger(f"Announcement {announcement} wird geskipped, keiner da")
return
async def voice_announcer(destination: str, voice_channel: discord.VoiceChannel):
sound_path = get_sound_path(destination=destination)
if sound_path is None:
return
logger(f"VC wird betreten, spiele {sound_path}")
vc = await voice_channel.connect(timeout=15, reconnect=True)
audio_source = discord.FFmpegPCMAudio(sound_path)
loop = asyncio.get_running_loop()
if not vc.is_playing():
def after_playing(error):
if error:
logger(f"Player error: {error}", "error")
loop.create_task(vc.disconnect())
logger("VC wird verlassen")
vc.play(audio_source, after=after_playing)
logger("Zug angekommen, wähle neue Verbindung")
await rename_vc(bot, voice_channel, from_scheduler=True)
src/utils.py
+8 -5
View File
@@ -76,7 +76,7 @@ def channel_formatting(mode: str) -> str:
return f"{emoji}{formatting}"
def get_train_name(train_name: str, mode: str) -> str:
if mode == "BUS" or mode == "TRAM" or train_name.isdigit():
if train_name.isdigit():
train = f"{mode.capitalize()} {train_name}"
elif "(" in train_name:
train = train_name.split(" (")[0]
@@ -101,7 +101,7 @@ def _reload_operators_if_changed():
logger("operators.py wurde automatisch neu geladen (Änderungen erkannt)")
def get_operator_metadata(agency: str, route_color: str) -> dict:
def get_operator_metadata(agency: str, route_color: str, mode: str) -> dict:
_reload_operators_if_changed()
op_data = operators.OPERATOR_ALIASES.get(agency) or operators.OPERATORS.get(agency) or operators.OPERATORS["fallback"]
@@ -147,7 +147,7 @@ def get_sound_path(destination) -> str | None:
return sound_path
def get_next_station(stops: dict, train_from :str) -> dict | None:
def get_next_station(stops: dict, train_from: str) -> dict | None:
now = datetime.now(LOCAL_TZ)
for name, info in stops.items():
arrival_dt = info["arrival"]
@@ -157,7 +157,7 @@ def get_next_station(stops: dict, train_from :str) -> dict | None:
else:
return {
"name": name,
"arrival": arrival_dt.strftime("%H:%M")
"arrival": arrival_dt
}
return None
@@ -174,7 +174,10 @@ def format_via_list(stops: dict) -> str:
if trip_to in important_stops:
important_stops.remove(trip_to)
via = f"{', '.join(important_stops[:-1])} und {important_stops[-1]}"
if len(important_stops) > 1:
via = f"{', '.join(important_stops[:-1])} und {important_stops[-1]}"
else:
via = important_stops[0]
return via
return None