SPACE WEATHER ALERT - EXTREME STORM CONDITIONS (G5) with probability ≥ 40% predicted
+
+
From 11:56 (CET) 18.02.2026 to 17:56 (CET) 18.02.2026, space weather can reach EXTREME STORM CONDITIONS with Kp = 9 with probability ≥ 40%.
+
Current Conditions: QUIET (Observed Kp data available up to 13:00 CET 17.02.2026)
+
+
Bar colours indicate geomagnetic activity levels: green corresponds to quiet conditions (Kp < 3), yellow to moderate activity (3 < Kp ≤ 6), and red to high storm conditions (Kp > 6). The red dashed line shows the official NOAA SWPC Kp forecast. Error bars represent the minimum-maximum spread of forecast Kp values.
+
+
ALERT SUMMARY
+
+
Alert sent at: 14:56 CET 17.02.2026
+
40% Probability of EXTREME STORM CONDITIONS (G5) within next 24 hours
+
+
AURORA WATCH:
+
+
+
Note: Kp ≥ 7 indicate potential auroral activity at Berlin latitudes. Time indicated in UTC.
+
GEOMAGNETIC ACTIVITY SCALE
+
+
+
+
Level
+
Kp Value
+
Description
+
+
+
+
+
Quiet
+
0-3
+
Quiet conditions
+
+
+
Active
+
4
+
Moderate geomagnetic activity
+
+
+
Minor Storm (G1)
+
5
+
Weak power grid fluctuations. For more details see NOAA [G1]
+
+
+
Moderate Storm (G2)
+
6
+
High-latitude power systems affected. For more details see NOAA [G2]
+
+
+
Strong Storm (G3)
+
7
+
Power systems may need voltage corrections. For more details see NOAA [G3]
+
+
+
Severe Storm (G4)
+
8
+
Possible widespread voltage control problems. For more details see NOAA [G4]
+
+
+
Extreme Storm (G5)
+
9
+
Widespread power system voltage control problems. For more details see NOAA [G5]
+
+
+
+
This is an automated alert from the Kp Index Monitoring System using GFZ Space Weather Forecast.
+
+
+
\ No newline at end of file
diff --git a/src/kp_index_monitor.py b/src/kp_index_monitor.py
index 2ec557d..3f4d441 100644
--- a/src/kp_index_monitor.py
+++ b/src/kp_index_monitor.py
@@ -22,6 +22,10 @@
from email.mime.text import MIMEText
from pathlib import Path
from typing import Optional, Tuple
+from zoneinfo import ZoneInfo
+
+# Display times in CET (Europe/Berlin handles CET/CEST)
+CET = ZoneInfo("Europe/Berlin")
import markdown
import numpy as np
@@ -90,9 +94,13 @@ class KpMonitor:
for geomagnetic activity monitoring.
"""
- IMAGE_PATH = "/PAGER/FLAG/data/published/kp_swift_ensemble_LAST.png"
- IMAGE_PATH_SWPC = "/PAGER/FLAG/data/published/kp_swift_ensemble_with_swpc_LAST.png"
- CSV_PATH = "/PAGER/FLAG/data/published/products/Kp/kp_product_file_SWIFT_LAST.csv"
+ IMAGE_PATH = "./assets/kp_swift_ensemble_LAST.png"
+ IMAGE_PATH_SWPC = "./assets/kp_swift_ensemble_with_swpc_LAST.png"
+ CSV_PATH = "./assets/kp_product_file_SWIFT_LAST.csv"
+ VIDEO_PATH_AURORA = "./assets/aurora_forecast.mp4"
+
+ # Caption for the forecast plot (SWPC + Min-Max)
+ FORECAST_IMAGE_CAPTION = "Bar colours indicate geomagnetic activity levels: green corresponds to quiet conditions (Kp < 3), yellow to moderate activity (3 < Kp ≤ 6), and red to high storm conditions (Kp > 6). The red dashed line shows the official NOAA SWPC Kp forecast. Error bars represent the minimum-maximum spread of forecast Kp values."
def __init__(self, config: MonitorConfig, log_suffix: str = "") -> None:
self.last_alert_time = None
@@ -104,6 +112,7 @@ def __init__(self, config: MonitorConfig, log_suffix: str = "") -> None:
self.config.kp_alert_threshold = np.round(self.config.kp_alert_threshold, 2)
self.kp_threshold_str = DECIMAL_TO_KP[self.config.kp_alert_threshold]
self.LOCAL_IMAGE_PATH = self.copy_image()
+ self.LOCAL_AURORA_VIDEO_PATH = None # set when building message with AURORA WATCH
self.current_utc_time = pd.Timestamp(datetime.now(timezone.utc))
self.log_suffix = log_suffix
self.setup_logging()
@@ -121,6 +130,10 @@ def copy_image(self) -> str:
return shutil.copy2(self.IMAGE_PATH_SWPC, "./kp_swift_ensemble_with_swpc_LAST.png")
return shutil.copy2(self.IMAGE_PATH, "./kp_swift_ensemble_LAST.png")
+ def copy_aurora_video(self) -> str:
+ """Copy the aurora video to the current directory for html embedding."""
+ return shutil.copy2(self.VIDEO_PATH_AURORA, "./aurora_forecast.mp4")
+
def setup_logging(self) -> None:
"""
Configure logging to file and console.
@@ -139,7 +152,7 @@ def log_uncaught_exceptions(exc_type, exc_value, exc_traceback):
logging.basicConfig(
level=self.config.log_level,
- format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+ format="[%(levelname)-8s] %(asctime)s - %(name)s:%(lineno)d - %(message)s",
handlers=[
logging.FileHandler(
self.log_folder
@@ -150,29 +163,71 @@ def log_uncaught_exceptions(exc_type, exc_value, exc_traceback):
)
self.logger = logging.getLogger(__name__)
- def fetch_kp_data(self) -> Optional[pd.DataFrame]:
+ def fetch_kp_data(self, test: bool = False) -> Optional[pd.DataFrame]:
"""
Fetch current Kp index forecast data from GFZ website.
+ Parameters
+ ----------
+ test : bool, optional
+ If True, create a test DataFrame with synthetic data instead of reading from CSV, by default False
Returns
-------
pd.DataFrame or None
DataFrame containing forecast data or None if fetch fails
"""
- try:
- df = pd.read_csv(self.CSV_PATH)
+ if not test:
+ try:
+ df = pd.read_csv(self.CSV_PATH)
- df["Time (UTC)"] = pd.to_datetime(df["Time (UTC)"], format="%d-%m-%Y %H:%M", dayfirst=True, utc=True)
- df.index = df["Time (UTC)"]
- self.logger.info(f"Successfully fetched {len(df)} records")
- return df
+ df["Time (UTC)"] = pd.to_datetime(df["Time (UTC)"], format="%d-%m-%Y %H:%M", dayfirst=True, utc=True)
+ df.index = df["Time (UTC)"]
+ self.logger.info(f"Successfully fetched {len(df)} records")
+ return df
- except pd.errors.EmptyDataError:
- self.logger.error("Received empty CSV file")
- return None
- except Exception as e:
- self.logger.error(f"Unexpected error: {e}", exc_info=True)
- return None
+ except pd.errors.EmptyDataError:
+ self.logger.error("Received empty CSV file")
+ return None
+ except Exception as e:
+ self.logger.error(f"Unexpected error: {e}", exc_info=True)
+ return None
+ else:
+ self.logger.info("Running in test mode - generating synthetic Kp data")
+ return self.get_test_data()
+
+ def round_to_step(self, x, step=0.33):
+ return np.round(x / step) * step
+
+ def get_test_data(self) -> pd.DataFrame:
+ time_index = pd.date_range(start=self.current_utc_time, periods=10, freq="3h")
+
+ data = {"Time (UTC)": time_index}
+
+ kp_members = []
+
+ for i in range(20):
+ vals = np.round(np.random.choice(np.linspace(0, 9, 28), len(time_index)), 2)
+ data[f"kp_{i}"] = vals
+ kp_members.append(vals)
+
+ kp_array = np.vstack(kp_members).T
+
+ data["minimum"] = np.min(kp_array, axis=1)
+ data["0.25-quantile"] = np.quantile(kp_array, 0.25, axis=1)
+ data["median"] = np.median(kp_array, axis=1)
+ data["0.75-quantile"] = np.quantile(kp_array, 0.75, axis=1)
+ data["maximum"] = np.max(kp_array, axis=1)
+
+ data["prob 4-5"] = np.mean((kp_array >= 4) & (kp_array < 5), axis=1)
+ data["prob 5-6"] = np.mean((kp_array >= 5) & (kp_array < 6), axis=1)
+ data["prob 6-7"] = np.mean((kp_array >= 6) & (kp_array < 7), axis=1)
+ data["prob 7-8"] = np.mean((kp_array >= 7) & (kp_array < 8), axis=1)
+ data["prob >= 8"] = np.mean(kp_array >= 8, axis=1)
+
+ df = pd.DataFrame(data)
+ df.index = df["Time (UTC)"]
+
+ return df
def analyze_kp_data(self, df: pd.DataFrame) -> AnalysisResults:
"""
@@ -193,7 +248,6 @@ def analyze_kp_data(self, df: pd.DataFrame) -> AnalysisResults:
self.logger.info(f"Current UTC Time: {self.current_utc_time}")
max_values = df[df.index >= self.current_utc_time]["maximum"]
max: float = np.round(max_values.max(), 2)
-
self.ensembles = [col for col in df.columns if re.match(r"kp_\d+", col)]
self.total_ensembles = len(self.ensembles)
probability = np.sum(df[self.ensembles] >= self.config.kp_alert_threshold, axis=1) / self.total_ensembles
@@ -243,7 +297,7 @@ def footer(self) -> str:
def _kp_html_table(self, record: pd.DataFrame, probabilities: pd.DataFrame) -> str:
"""Generate markdown table for Kp index records."""
table = f"""
-| Time (UTC) | Probability (Kp ≥ {self.kp_threshold_str}) | Min Kp Index[1] | Max Kp Index[2] | Median Kp Index[3] | Activity[4][5] |
+| Time (CET) | Probability (Kp ≥ {self.kp_threshold_str}) | Min Kp Index[1] | Max Kp Index[2] | Median Kp Index[3] | Activity[4][5] |
|------------|-------------------------------------------|------------------|------------------|---------------------|------------------|
"""
for _, row in record.iterrows():
@@ -256,7 +310,7 @@ def _kp_html_table(self, record: pd.DataFrame, probabilities: pd.DataFrame) -> s
time_idx = row["Time (UTC)"]
prob = probabilities.loc[time_idx, "Probability"]
- time_str = row["Time (UTC)"].strftime("%Y-%m-%d %H:%M")
+ time_str = row["Time (UTC)"].tz_convert(CET).strftime("%Y-%m-%d %H:%M")
prob_str = f"{prob * 100:.0f}%"
activity_str = f'{level_min} - {level_max}'
@@ -348,7 +402,9 @@ def create_message(self, analysis: AnalysisResults) -> str:
max_kp_at_finite_time = np.round(max_values.max(), 2)
- max_kp_at_finite_time_status, _, _ = self.get_status_level_color(max_kp_at_finite_time)
+ max_kp_at_finite_time_status, max_kp_at_finite_time_level, _ = self.get_status_level_color(
+ max_kp_at_finite_time
+ )
mask = probability_df["Probability"] >= 0.4
if mask.any():
start_time = probability_df.index[mask][0]
@@ -368,39 +424,45 @@ def create_message(self, analysis: AnalysisResults) -> str:
start_time_kp_min_status, _, _ = self.get_status_level_color(high_records.loc[start_time]["minimum"].min())
end_time_kp_max_status, _, _ = self.get_status_level_color(high_records.loc[end_time]["maximum"].max())
+ start_cet = start_time.tz_convert(CET)
+ end_cet = end_time.tz_convert(CET)
if start_time == end_time:
- message_prefix = f"""At {start_time.strftime("%Y-%m-%d %H:%M")} UTC"""
+ message_prefix = f"""At {start_cet.strftime("%H:%M (CET) %d.%m.%Y")} """
else:
message_prefix = (
- f"""From {start_time.strftime("%Y-%m-%d %H:%M")} UTC to {end_time.strftime("%Y-%m-%d %H:%M")} UTC"""
+ f"""From {start_cet.strftime("%H:%M (CET) %d.%m.%Y")} to {end_cet.strftime("%H:%M (CET) %d.%m.%Y")}"""
)
if observed_time != analysis.next_24h_forecast.index[0]:
- obs_message_prefix = f""" (Observed Kp data available up to {datetime.strptime(observed_time.strip(), "%Y-%m-%dT%H:%M:%SZ").strftime("%Y-%m-%d %H:%M")} UTC)"""
+ obs_utc = datetime.strptime(observed_time.strip(), "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
+ obs_message_prefix = (
+ f""" (Observed Kp data available up to {obs_utc.astimezone(CET).strftime("%H:%M CET %d.%m.%Y")})"""
+ )
else:
obs_message_prefix = ""
- message = f"""
SPACE WEATHER ALERT - {threshold_status} ({threshold_level}) Predicted
+ # Use "=" for Kp 9 (maximum value), "≥" for all other values
+ kp_comparison = "=" if max_kp_at_finite_time == 9 else "≥"
+ message = f"""
SPACE WEATHER ALERT - {end_time_kp_max_status} ({max_kp_at_finite_time_level}) with probability ≥ {prob_at_start_time * 100:.0f}% predicted
-### {message_prefix} Kp is expected to be above {self.config.kp_alert_threshold} ({threshold_level}) with ≥ {prob_at_start_time * 100:.0f}% probability with {start_time_kp_min_status.replace("CONDITIONS", "")} to {end_time_kp_max_status}.
-**Current Predicted Conditions:** {status.replace("CONDITIONS", "")}
-**Current Observed Conditions:** {observed_status.replace("CONDITIONS", "")} {obs_message_prefix}
-
-## **ALERT SUMMARY**
+### {message_prefix}, space weather can reach {end_time_kp_max_status} with Kp {kp_comparison} {DECIMAL_TO_KP[max_kp_at_finite_time]} with probability ≥ {prob_at_start_time * 100:.0f}%.
-- **Alert sent at:** {datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M")} UTC
-- **Maximum Kp ≥ {DECIMAL_TO_KP[max_kp_at_finite_time]} with {max_kp_at_finite_time_status} may occur** {max_values.idxmax().strftime("%Y-%m-%d %H:%M")} UTC onwards
-- **{high_prob_value * 100:.0f}% Probability of {threshold_status} ({threshold_level}) within next {prob_at_time} hours**
+**Current Conditions:** {observed_status.replace("CONDITIONS", "")} {obs_message_prefix}
-## **HIGH Kp INDEX PERIODS Predicted (Kp ≥ {threshold_level})**
+
{self.FORECAST_IMAGE_CAPTION}
+
+## **ALERT SUMMARY**
+- **Alert sent at:** {datetime.now(timezone.utc).astimezone(CET).strftime("%H:%M CET %d.%m.%Y ")}
+- **{high_prob_value * 100:.0f}% Probability of {end_time_kp_max_status} ({max_kp_at_finite_time_level}) within next {prob_at_time} hours**
+
"""
- message += self._kp_html_table(high_records, probability_df)
+ # message += self._kp_html_table(high_records, probability_df)
- AURORA_KP = 6.33
+ AURORA_KP = 7
high_records_above_threshold = high_records[
(high_records["minimum"].astype(float) >= AURORA_KP)
| (high_records["median"].astype(float) >= AURORA_KP)
@@ -411,11 +473,16 @@ def create_message(self, analysis: AnalysisResults) -> str:
message += f"""
## **AURORA WATCH:**
-**Note:** Kp ≥ {DECIMAL_TO_KP[AURORA_KP]} indicate potential auroral activity at Berlin latitudes.
+
+
+**Note:** Kp ≥ {DECIMAL_TO_KP[AURORA_KP]} indicate potential auroral activity at Berlin latitudes. Time indicated in UTC.
"""
- message += """## GEOMAGNETIC ACTIVITY SCALE 5"""
+ message += """## GEOMAGNETIC ACTIVITY SCALE"""
message += self.get_storm_level_description_table()
message += "\n"
message += self.footer()
@@ -571,11 +638,15 @@ def should_send_alert(self, analysis: AnalysisResults) -> bool:
return True
- def run_single_check(self) -> bool:
+ def run_single_check(self, test: bool = False) -> bool:
"""
Execute a single monitoring check cycle.
Fetches Kp data, analyzes it, and sends alerts if necessary.
+ Parameters
+ ----------
+ test : bool, optional
+ If True, runs in test mode with synthetic data, by default False
Returns
-------
@@ -583,7 +654,7 @@ def run_single_check(self) -> bool:
True if check completed successfully, False otherwise
"""
self.logger.info("Kp Index check")
- df = self.fetch_kp_data()
+ df = self.fetch_kp_data(test=test)
if df is None:
return False
analysis = self.analyze_kp_data(df)
@@ -594,8 +665,9 @@ def run_single_check(self) -> bool:
message = self.create_message(analysis)
subject = self.create_subject(analysis)
- email_sent = self.send_alert(subject, message)
_ = self.copy_image()
+ self.LOCAL_AURORA_VIDEO_PATH = self.copy_aurora_video()
+ email_sent = self.send_alert(subject, message)
message_for_file = markdown.markdown(
message.replace("cid:forecast_image", self.LOCAL_IMAGE_PATH),
extensions=["tables", "fenced_code", "footnotes", "nl2br"],
@@ -661,6 +733,9 @@ def construct_and_send_email(self, recipients: list[str], subject: str, message:
img.add_header("Content-Disposition", "inline", filename="forecast_image.png")
msg_root.attach(img)
+ # Note: Video attachments are not well supported in email clients
+ # The video will be available in the generated HTML file
+
with smtplib.SMTP("localhost") as smtp:
smtp.send_message(msg_root)
@@ -670,15 +745,16 @@ def basic_html_format(self, message: str) -> str:
@@ -725,24 +801,27 @@ def run_continuous_monitoring(self) -> None:
def main(
once: bool = typer.Option(False, "--once", help="Run single check and exit"),
continuous: bool = typer.Option(False, "--continuous", help="Run continuous monitoring"),
+ test: bool = typer.Option(False, "--test", help="Run test mode with sample data"),
):
"""
Main function with command line interface.
"""
- selected = [flag for flag in (once, continuous) if flag]
+ selected = [flag for flag in (once, continuous, test) if flag]
if len(selected) == 0:
- raise typer.BadParameter("One of --once or --continuous must be specified")
+ raise typer.BadParameter("One of --once, --continuous, or --test must be specified")
if len(selected) > 1:
raise typer.BadParameter(
- "Options --once and --continuous are mutually exclusive i.e., only one can be selected."
+ "Options --once, --continuous, and --test are mutually exclusive i.e., only one can be selected."
)
config = MonitorConfig.from_yaml()
- log_suffix = "once" if once else "continuous"
+ log_suffix = "once" if once else "continuous" if not test else "test"
monitor = KpMonitor(config, log_suffix=log_suffix)
if once:
- monitor.run_single_check()
+ monitor.run_single_check(test=False)
+ elif test:
+ monitor.run_single_check(test=True)
elif continuous:
monitor.run_continuous_monitoring()
