KB8PMY
/
SkywarnPlus
mirror of https://github.com/Mason10198/SkywarnPlus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

v0.4.2 update

Browse Source
pull/140/head
Mason10198 2 years ago
parent 4b8884ea04
commit 92d956e3d8
5 changed files with 126 additions and 57 deletions
Show Stats Download Patch File Download Diff File

2
SkyControl.py
Unescape View File

@ -1,7 +1,7 @@
#!/usr/bin/python3
"""
SkyControl v0.4.0 by Mason Nelson
SkyControl v0.4.2 by Mason Nelson
==================================
A Control Script for SkywarnPlus

53
SkyDescribe.py
Unescape View File

@ -1,7 +1,7 @@
#!/usr/bin/python3
"""
SkyDescribe v0.4.0 by Mason Nelson
SkyDescribe v0.4.2 by Mason Nelson
==================================================
Text to Speech conversion for Weather Descriptions
@ -45,25 +45,25 @@ CONFIG_PATH = os.path.join(BASE_DIR, "config.yaml")
# Open and read configuration file
with open(CONFIG_PATH, "r") as config_file:
config = YAML.load(config_file)
CONFIG = YAML.load(config_file)
# Define tmp_dir
TMP_DIR = config.get("DEV", []).get("TmpDir", "/tmp/SkywarnPlus")
TMP_DIR = CONFIG.get("DEV", []).get("TmpDir", "/tmp/SkywarnPlus")
# Define VoiceRSS settings
# get api key, fellback 150
API_KEY = config.get("SkyDescribe", []).get("APIKey", "")
LANGUAGE = config.get("SkyDescribe", []).get("Language", "en-us")
SPEED = config.get("SkyDescribe", []).get("Speed", 0)
VOICE = config.get("SkyDescribe", []).get("Voice", "John")
MAX_WORDS = config.get("SkyDescribe", []).get("MaxWords", 150)
API_KEY = CONFIG.get("SkyDescribe", []).get("APIKey", "")
LANGUAGE = CONFIG.get("SkyDescribe", []).get("Language", "en-us")
SPEED = CONFIG.get("SkyDescribe", []).get("Speed", 0)
VOICE = CONFIG.get("SkyDescribe", []).get("Voice", "John")
MAX_WORDS = CONFIG.get("SkyDescribe", []).get("MaxWords", 150)
# Path to the data file
DATA_FILE = os.path.join(TMP_DIR, "data.json")
# Define logger
LOGGER = logging.getLogger(__name__)
if config.get("Logging", []).get("Debug", False):
if CONFIG.get("Logging", []).get("Debug", False):
LOGGER.setLevel(logging.DEBUG)
else:
LOGGER.setLevel(logging.INFO)
@ -286,13 +286,42 @@ def main(index_or_title):
)
else:
alert, alert_data = alerts[index]
description = alert_data[0]["description"]
# Count the unique instances of the alert
unique_instances = len(
set((data["description"], data["end_time_utc"]) for data in alert_data)
)
# Modify the description
if unique_instances == 1:
description = alert_data[0]["description"]
else:
description = "{} unique instances of this alert exist. Describing the first instance. {}".format(
unique_instances, alert_data[0]["description"]
)
else:
# Argument is not an index, assume it's a title
title = index_or_title
for alert, alert_data in alerts:
if alert == title: # Assuming alert is a title
description = alert_data[0]["description"]
# Count the unique instances of the alert
unique_instances = len(
set(
(data["description"], data["end_time_utc"])
for data in alert_data
)
)
# Modify the description
if unique_instances == 1:
description = alert_data[0]["description"]
else:
description = "There are {} unique instances of {}. Describing the first instance. {}".format(
unique_instances,
alert,
alert_data[0]["description"]
)
break
else:
LOGGER.error("SkyDescribe: No alert with title %s found.", title)
@ -322,7 +351,7 @@ def main(index_or_title):
duration = frames / float(rate)
LOGGER.debug("SkyDescribe: Length of the audio file in seconds: %s", duration)
nodes = config["Asterisk"]["Nodes"]
nodes = CONFIG["Asterisk"]["Nodes"]
for node in nodes:
LOGGER.info("SkyDescribe: Broadcasting description on node %s.", node)
command = "/usr/sbin/asterisk -rx 'rpt localplay {} {}'".format(

99
SkywarnPlus.py
Unescape View File

@ -1,7 +1,7 @@
#!/usr/bin/python3
"""
SkywarnPlus v0.4.0 by Mason Nelson
SkywarnPlus v0.4.2 by Mason Nelson
===============================================================================
SkywarnPlus is a utility that retrieves severe weather alerts from the National
Weather Service and integrates these alerts with an Asterisk/app_rpt based
@ -386,35 +386,50 @@ def get_alerts(countyCodes):
LOGGER.debug("getAlerts: DEV Alert Injection Enabled")
injected_alerts = config["DEV"].get("INJECTALERTS", [])
LOGGER.debug("getAlerts: Injecting alerts: %s", injected_alerts)
if injected_alerts is None:
injected_alerts = []
county_codes_cycle = itertools.cycle(
countyCodes
) # Create an iterator that returns elements from the iterable in a cyclic manner
max_counties = len(countyCodes) # Assuming countyCodes is a list of counties
county_codes_cycle = itertools.cycle(countyCodes)
counter = 0
for i, event in enumerate(injected_alerts):
last_word = event.split()[-1]
for alert_info in injected_alerts:
if isinstance(alert_info, dict):
alert_title = alert_info.get("Title", "")
specified_counties = alert_info.get("CountyCodes", [])
else:
continue # Ignore if not a dictionary
last_word = alert_title.split()[-1]
severity = severity_mapping_words.get(last_word, 0)
description = "This alert was manually injected as a test."
end_time_str = alert_info.get("EndTime")
county_data = []
for j in range(
i + 1
): # Here we increase the number of county codes for each alert
end_time_utc = current_time + timedelta(hours=counter + 1)
for county in specified_counties:
if county not in countyCodes:
LOGGER.error(
"Specified county code '%s' is not defined in the config. Using next available county code from the config.",
county,
)
county = next(county_codes_cycle)
end_time = (
datetime.strptime(end_time_str, "%Y-%m-%dT%H:%M:%SZ")
if end_time_str
else current_time + timedelta(hours=1)
)
county_data.append(
{
"county_code": next(county_codes_cycle),
"county_code": county,
"severity": severity,
"description": description,
"end_time_utc": end_time_utc.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
"end_time_utc": end_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
}
)
counter += 1 # Increase counter here
alerts[event] = county_data # Add the list of dictionaries to the alert
alerts[
alert_title
] = county_data # Add the list of dictionaries to the alert
# We limit the number of alerts to the maximum defined constant.
alerts = OrderedDict(list(alerts.items())[:MAX_ALERTS])
@ -689,10 +704,7 @@ def say_alerts(alerts):
)
alert_count = 0
for (
alert,
counties,
) in alerts.items(): # Now we loop over both alert name and its associated counties
for alert, counties in alerts.items():
if alert in filtered_alerts:
try:
descriptions = [county["description"] for county in counties]
@ -722,15 +734,18 @@ def say_alerts(alerts):
)
alert_count += 1
# Add county names if they exist
added_county_codes = set()
for county in counties:
# if its the first county, word_space is 600ms of silence. else it is 400ms
if counties.index(county) == 0:
word_space = AudioSegment.silent(duration=600)
else:
word_space = AudioSegment.silent(duration=400)
county_code = county["county_code"]
if COUNTY_WAVS and county_code in COUNTY_CODES:
if (
COUNTY_WAVS
and county_code in COUNTY_CODES
and county_code not in added_county_codes
):
index = COUNTY_CODES.index(county_code)
county_name_file = COUNTY_WAVS[index]
LOGGER.debug(
@ -739,10 +754,17 @@ def say_alerts(alerts):
county_name_file,
alert,
)
combined_sound += word_space + AudioSegment.from_wav(
os.path.join(SOUNDS_PATH, county_name_file)
)
# if this is the last county name, add 600ms of silence after the county name
try:
combined_sound += word_space + AudioSegment.from_wav(
os.path.join(SOUNDS_PATH, county_name_file)
)
except FileNotFoundError:
LOGGER.error(
"sayAlert: County audio file not found: %s",
os.path.join(SOUNDS_PATH, county_name_file),
)
added_county_codes.add(county_code)
if counties.index(county) == len(counties) - 1:
combined_sound += AudioSegment.silent(duration=600)
@ -750,7 +772,7 @@ def say_alerts(alerts):
LOGGER.error("sayAlert: Alert not found: %s", alert)
except FileNotFoundError:
LOGGER.error(
"sayAlert: Audio file not found: %s/ALERTS/SWP_%s.wav",
"sayAlert: Alert audio file not found: %s/ALERTS/SWP_%s.wav",
SOUNDS_PATH,
ALERT_INDEXES[index],
)
@ -920,15 +942,16 @@ def build_tailmessage(alerts):
descriptions = [county["description"] for county in counties]
end_times = [county["end_time_utc"] for county in counties]
if len(set(descriptions)) > 1 or len(set(end_times)) > 1:
LOGGER.debug(
"buildTailMessage: Found multiple unique instances of the alert %s",
alert,
)
multiples_sound = AudioSegment.from_wav(
os.path.join(SOUNDS_PATH, "ALERTS", "SWP_149.wav")
)
combined_sound += AudioSegment.silent(duration=200) + multiples_sound
if config["Alerting"]["WithMultiples"]:
if len(set(descriptions)) > 1 or len(set(end_times)) > 1:
LOGGER.debug(
"buildTailMessage: Found multiple unique instances of the alert %s",
alert,
)
multiples_sound = AudioSegment.from_wav(
os.path.join(SOUNDS_PATH, "ALERTS", "SWP_149.wav")
)
combined_sound += AudioSegment.silent(duration=200) + multiples_sound
# Add county names if they exist
if county_identifiers:

2
UpdateSWP.py
Unescape View File

@ -1,7 +1,7 @@
#!/usr/bin/python3
"""
SkywarnPlus Updater v0.4.0 by Mason Nelson
SkywarnPlus Updater v0.4.2 by Mason Nelson
===============================================================================
Script to update SkywarnPlus to the latest version. This script will download
the latest version of SkywarnPlus from GitHub, and then merge the existing

27
config.yaml
Unescape View File

@ -1,4 +1,4 @@
# SkywarnPlus v0.4.0 Configuration File
# SkywarnPlus v0.4.2 Configuration File
# Author: Mason Nelson (N5LSN/WRKF394)
# Please edit this file according to your specific requirements.
@ -75,6 +75,10 @@ Alerting:
# Specify the WAV file in the SOUNDS/ALERTS directory to use as the alert seperator sound effect
AlertSeperator: Woodblock.wav
# Enable audio tagging an alert as having "multiples" if there is more than one unique instance of that alert type
# If enabled, and there are 2x different Severe Thunderstorm Warnings in your area, the audio will be: "Severe Thunderstorm Warning, with multiples"
WithMultiples: true
# Limit the maximum number of alerts to process in case of multiple alerts.
# SkywarnPlus fetches all alerts, orders them by severity, and processes only the 'n' most severe alerts, where 'n' is the MaxAlerts value.
MaxAlerts: 99
@ -409,8 +413,21 @@ DEV:
# Enable test alert injection instead of calling the NWS API by setting 'INJECT' to 'True'.
INJECT: false
# List the test alerts to inject. Use a case-sensitive list. One alert per line.
# List the test alerts to inject. Alert titles are case sensitive.
# Optionally specify the CountyCodes and/or EndTime for each alert.
# CountyCodes used here must be defined at the top of this configuration file.
# Example:
# INJECTALERTS:
# - Title: "Tornado Warning"
# CountyCodes: ["ARC119", "ARC120"]
# - Title: "Tornado Watch"
# CountyCodes: ["ARC125"]
# EndTime: "2023-08-01T12:00:00Z"
# - Title: "Severe Thunderstorm Warning"
INJECTALERTS:
- Tornado Warning
- Tornado Watch
- Severe Thunderstorm Warning
- Title: "Tornado Warning"
CountyCodes: ["ARC119", "ARC120"]
- Title: "Tornado Watch"
CountyCodes: ["ARC125"]
- Title: "Severe Thunderstorm Warning"
CountyCodes: ["ARC085"]
Write Preview
Loading…
Cancel
Save

Powered by TurnKey Linux.