-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
146 lines (121 loc) · 6.18 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
import os
import logging
import time
import re
import subprocess
# Külső könyvtárak
import yaml
import schedule
from dotenv import load_dotenv
# Konstansok
POSSIBLE_EVENTS = ["becsengo", "kicsengo", "hirdetes", "podcast"]
FFPLAY_COMMAND = "ffplay -v 0 -nodisp -autoexit"
# Környezeti változók betöltése
load_dotenv()
RUNNING_PATH = os.environ.get("RUNNING_PATH", default="./")
LOGGING_LEVEL = int(os.environ.get("LOGGING_LEVEL", default="20"))
# A logger beállítása. Érdemes így hagyni. Ha túl sok az kimenet, a "level" változót lehet változtatni.
logging.basicConfig(format='%(asctime)s %(levelname)-8s %(message)s', level = LOGGING_LEVEL, datefmt='%Y-%m-%d %H:%M:%S')
# Logging
logging.info("Konfigurációk betöltése...")
# File helyek importálása. A paths.yaml file ezzel a script-tel egyhelyen kell hogy legyen!
PATHS: dict[str] = []
with open(RUNNING_PATH + "paths.yaml") as paths_file:
PATHS = yaml.safe_load(paths_file)
# Ellenőrzés hogy léteznek-e a megadott fájlok illetve mappák.
for path_name, path in PATHS.items():
if not os.path.exists(path):
logging.error("A '%s' kulcshoz megadott '%s' file vagy mappa nem létezik!", path_name, path)
logging.debug("File helyek: %s", PATHS)
# Időpontok importálása egy könyvtárba.
EVENTS: dict[str, str] = []
with open(PATHS["events"]) as events_file:
EVENTS = yaml.safe_load(events_file)
# Az időpontok és eseménytípusok ellenőrzése.
for event in EVENTS:
if not re.match(r"[0-2]\d:[0-5]\d", event["time"]):
logging.error("A '%s' eseményhez megadott '%s' időpont formázása helytelen. Helyes formátum: HH:MM", event["type"], event["time"])
if event["type"] not in POSSIBLE_EVENTS:
logging.error("A '%s' időponthoz rendelt '%s' esemény nem létezik! Lehetséges opciók: '%s'", event["time"], event["type"], POSSIBLE_EVENTS)
logging.debug("Események: %s", EVENTS)
# Funkció a különböző típusú események felismerésére és a megfelelő feladat futtatására.
def run_event(event_type: str) -> None:
match event_type:
case "becsengo":
logging.info("Becsengő lejátszása...")
try:
subprocess.run(FFPLAY_COMMAND + " " + PATHS["sounds-in"])
except Exception:
logging.warning("Becsengő sikertelen!")
else:
logging.debug("Becsengő lejátszva.")
return
case "kicsengo":
logging.info("Kicsengő lejátszása...")
try:
subprocess.run(FFPLAY_COMMAND + " " + PATHS["sounds-out"])
except Exception :
logging.warning("Kicsengő sikertelen!")
else:
logging.debug("Kicsengő lejátszva.")
return
case "hirdetes":
# TODO: .wav és egyéb audio formátumok elfogadása és kezelése
# AZ új hirdetések mappájának ellenőrzése
logging.debug("Hirdetés keresése...")
possible_files = os.listdir(PATHS["announcements-new"])
possible_announcements = [e for e in possible_files if re.match(r".*\.mp3", e)]
if not possible_announcements:
logging.info("Nincs hirdetés!")
return
# Fájlnév illetve helyek előkészítése
TIMESTAMP = str(round(time.time() * 1000))
ANNOUNCEMENT_NAME = possible_announcements[0][:-4]
ANNOUNCEMENT_STARING_PATH = PATHS["announcements-new"] + ANNOUNCEMENT_NAME + ".mp3"
ANNOUNCEMENT_ENGIND_PATH = PATHS["announcements-old"] + ANNOUNCEMENT_NAME + "-" + TIMESTAMP + ".mp3"
# Dallam és hirdetés lejátszása
logging.info("A '%s' nevű hirdetés lejátszása...", ANNOUNCEMENT_NAME)
subprocess.run(FFPLAY_COMMAND + " " + PATHS["sounds-chime"], shell=True)
subprocess.run(FFPLAY_COMMAND + " " + ANNOUNCEMENT_STARING_PATH, shell=True)
os.rename(ANNOUNCEMENT_STARING_PATH, ANNOUNCEMENT_ENGIND_PATH)
logging.debug("Hirdetés lejátszva.")
return
case "podcast":
# TODO: .wav és egyéb audio formátumok elfogadása és kezelése
# AZ új hirdetések mappájának ellenőrzése
logging.debug("Hirdetés keresése...")
possible_files = os.listdir(PATHS["podcasts-new"])
possible_announcements = [e for e in possible_files if re.match(r".*\.mp3", e)]
if not possible_announcements:
logging.info("Nincs podcast!")
return
# Fájlnév illetve helyek előkészítése
TIMESTAMP = str(round(time.time() * 1000))
ANNOUNCEMENT_NAME = possible_announcements[0][:-4]
ANNOUNCEMENT_STARING_PATH = PATHS["podcasts-new"] + ANNOUNCEMENT_NAME + ".mp3"
ANNOUNCEMENT_ENGIND_PATH = PATHS["podcasts-old"] + ANNOUNCEMENT_NAME + "-" + TIMESTAMP + ".mp3"
# Dallam és hirdetés lejátszása
logging.info("A '%s' nevű podcast lejátszása...", ANNOUNCEMENT_NAME)
subprocess.run(FFPLAY_COMMAND + " " + PATHS["sounds-chime"], shell=True)
subprocess.run(FFPLAY_COMMAND + " " + ANNOUNCEMENT_STARING_PATH, shell=True)
os.rename(ANNOUNCEMENT_STARING_PATH, ANNOUNCEMENT_ENGIND_PATH)
logging.debug("Podcast lejátszva.")
return
case _:
logging.warning("A bemeneti file nem megfelelően van formázva!, '%s' esemény nem létezik.", event_type)
return
# A konfigurációs fájlban megadott események beidőzítése hétköznapokra.
logging.info("Események időzítése...")
for event in EVENTS:
schedule.every().monday.at(event["time"]).do(run_event, event["type"])
schedule.every().tuesday.at(event["time"]).do(run_event, event["type"])
schedule.every().wednesday.at(event["time"]).do(run_event, event["type"])
schedule.every().thursday.at(event["time"]).do(run_event, event["type"])
schedule.every().friday.at(event["time"]).do(run_event, event["type"])
# Logging
logging.info("Inicializáció sikeres!")
# A beidőzített események végrehajtása.
logging.info("Várakozás az eseményekre...")
while True:
schedule.run_pending()
time.sleep(1)