🐍 Disaster Lab — learn Python by building an alert system for Chiang Mai

# step1_rain.py — one signal, one rule.
# Rain in Chiang Mai, in millimetres per day, for one week.
# If one day has heavy rain, we print a warning.

RAIN = [4, 0, 12, 31, 88, 64, 9]   # mm per day, Monday to Sunday
HEAVY = 50                         # 50 mm in one day is heavy rain

for day, mm in enumerate(RAIN, start=1):
    if mm >= HEAVY:
        print("day", day, "- heavy rain:", mm, "mm. Watch the river.")
    else:
        print("day", day, "- ok:", mm, "mm.")

# step2_river.py — two signals, and a score for each one.
# A score is a number from 0 to 1. 0 means calm. 1 means danger.
# The file is growing, so it gets a shape: functions, and main().

RAIN = [4, 0, 12, 31, 88, 64, 9]              # mm per day
RIVER = [2.4, 2.4, 2.5, 2.9, 3.6, 4.1, 3.9]    # metres at gauge P.1, Nawarat Bridge
CRITICAL = 3.70                                # at 3.70 m the Irrigation Office warns


def ramp(value, low, high):
    """0 below low, 1 above high, a straight line between them."""
    if value <= low:
        return 0.0
    if value >= high:
        return 1.0
    return (value - low) / (high - low)


def rain_score(mm):
    """20 mm is nothing. 80 mm in one day is full danger."""
    return ramp(mm, 20, 80)


def river_score(metres):
    """3.0 m is calm. At the critical level the score is 1."""
    return ramp(metres, 3.0, CRITICAL)


def main():
    for day in range(len(RAIN)):
        rain = rain_score(RAIN[day])
        river = river_score(RIVER[day])
        print(f"day {day + 1}: rain {rain:.2f}   river {river:.2f}")


if __name__ == "__main__":
    main()

# step3_forecast.py — history makes a forecast.
# The river does not jump. It climbs. The climb today tells us about tomorrow.
# New here: forecast() and forecast_score().

RAIN = [4, 0, 12, 31, 88, 64, 9]              # mm per day
RIVER = [2.4, 2.4, 2.5, 2.9, 3.6, 4.1, 3.9]    # metres at gauge P.1, Nawarat Bridge
CRITICAL = 3.70                                # at 3.70 m the Irrigation Office warns


def ramp(value, low, high):
    """0 below low, 1 above high, a straight line between them."""
    if value <= low:
        return 0.0
    if value >= high:
        return 1.0
    return (value - low) / (high - low)


def rain_score(mm):
    """20 mm is nothing. 80 mm in one day is full danger."""
    return ramp(mm, 20, 80)


def river_score(metres):
    """3.0 m is calm. At the critical level the score is 1."""
    return ramp(metres, 3.0, CRITICAL)


def forecast(levels, day):
    """Tomorrow's level, if the river climbs the same as it did today."""
    if day == 0:
        return levels[0]
    climb = levels[day] - levels[day - 1]
    return levels[day] + climb


def forecast_score(metres):
    """Water enters the first streets at 3.9 m, so a forecast of 3.9 m is full danger."""
    return ramp(metres, 3.3, 3.9)


def main():
    for day in range(len(RAIN)):
        tomorrow = forecast(RIVER, day)
        print(f"day {day + 1}: river {RIVER[day]:.2f} m   "
              f"forecast {tomorrow:.2f} m   score {forecast_score(tomorrow):.2f}")


if __name__ == "__main__":
    main()

# step4_shake.py — the earthquake signal.
# A seismometer hears two waves. The P wave is fast and small. The S wave comes
# later and does the damage. The gap between them says how far away the quake is.
# New here: distance_km(), warning_seconds(), intensity() and shake_score().

import math

P_SPEED = 5.0           # km per second, a P wave in granite
S_SPEED = 3.0           # km per second, about 60 percent of the P wave
KM_PER_GAP_SECOND = 7.5  # from the two speeds: 5 x 3 / (5 - 3). Many books round it to 8.


def ramp(value, low, high):
    """0 below low, 1 above high, a straight line between them."""
    if value <= low:
        return 0.0
    if value >= high:
        return 1.0
    return (value - low) / (high - low)


def distance_km(gap_seconds):
    """Seconds between the P wave and the S wave, times 7.5, is the distance."""
    return gap_seconds * KM_PER_GAP_SECOND


def warning_seconds(km):
    """The P wave arrives first. The time before the S wave is the warning time."""
    return km / S_SPEED - km / P_SPEED


def intensity(magnitude, km):
    """How strong the shaking feels, from 1 to 12.
    A simple classroom rule, fitted to the 2025 quake. Real agencies use better ones."""
    value = 1.5 * magnitude - 3 * math.log10(max(km, 1)) + 1.5
    return max(1.0, min(12.0, value))


def shake_score(magnitude, km):
    """Intensity 3 is a weak shake. Intensity 8 breaks buildings."""
    return ramp(intensity(magnitude, km), 3, 8)


def main():
    # 28 March 2025, magnitude 7.7, Sagaing Fault, Myanmar
    magnitude = 7.7
    for city, km in [("Mandalay", 16), ("Chiang Mai", 480), ("Bangkok", 1000)]:
        print(f"{city:11} {km:5} km   shaking {intensity(magnitude, km):4.1f}   "
              f"warning {warning_seconds(km):5.1f} s   "
              f"score {shake_score(magnitude, km):.2f}")

    # A seismometer in Chiang Mai hears the P wave, then the S wave 64 seconds later.
    print("P to S gap 64 s -> about", round(distance_km(64)), "km away")


if __name__ == "__main__":
    main()

# step5_satellite.py — the satellite signal.
# A radar satellite takes a picture before and a picture after. Water looks dark.
# We subtract the two pictures. Every cell that changed a lot is "changed".
# New here: changed_fraction() and satellite_score().

# Two tiny radar pictures, 6 by 6 cells. 9 is bright (dry land), 0 is dark (water).
BEFORE = [
    [7, 8, 7, 6, 8, 7],
    [8, 7, 2, 6, 7, 8],   # the river is the dark 2 in the middle
    [7, 6, 2, 7, 8, 7],
    [8, 7, 2, 6, 7, 6],
    [7, 8, 3, 7, 8, 7],
    [8, 7, 2, 7, 7, 8],
]
AFTER = [
    [7, 8, 7, 6, 8, 7],
    [8, 2, 1, 2, 7, 8],   # the dark water is wider now
    [7, 1, 1, 2, 2, 7],
    [2, 1, 1, 2, 1, 6],
    [7, 2, 1, 1, 2, 7],
    [8, 7, 2, 2, 7, 8],
]
BIG_CHANGE = 4   # a cell that changed by 4 or more is a real change, not noise


def ramp(value, low, high):
    """0 below low, 1 above high, a straight line between them."""
    if value <= low:
        return 0.0
    if value >= high:
        return 1.0
    return (value - low) / (high - low)


def changed_fraction(before, after):
    """What part of the picture changed? 0.0 is nothing, 1.0 is everything."""
    changed = 0
    total = 0
    for row_before, row_after in zip(before, after):
        for old, new in zip(row_before, row_after):
            total += 1
            if abs(new - old) >= BIG_CHANGE:
                changed += 1
    return changed / total


def satellite_score(fraction):
    """5 percent changed is noise. 30 percent changed is a flood."""
    return ramp(fraction, 0.05, 0.30)


def main():
    fraction = changed_fraction(BEFORE, AFTER)
    print(f"changed: {fraction:.0%} of the picture")
    print(f"satellite score: {satellite_score(fraction):.2f}")
    for row_before, row_after in zip(BEFORE, AFTER):
        pairs = zip(row_before, row_after)
        marks = ["#" if abs(new - old) >= BIG_CHANGE else "." for old, new in pairs]
        print(" ".join(marks))


if __name__ == "__main__":
    main()

# step6_archive.py — the history signal.
# An archive of past disasters. We search it with today's words.
# A past day that shares many words with today is a warning.
# New here: words(), search() and archive_score().

# Words that every text shares. They say nothing, so the search ignores them.
STOP_WORDS = {"the", "and", "for", "with", "from"}

ARCHIVE = [
    {"year": 2004, "kind": "tsunami", "place": "Khao Lak and Phuket",
     "text": "26 December 2004. The sea pulled back from the beach. Then big waves came. "
             "About 5,400 people died in Thailand. There was no warning system."},
    {"year": 2005, "kind": "flood", "place": "Chiang Mai",
     "text": "A big flood hit Chiang Mai city. River water entered the streets."},
    {"year": 2011, "kind": "flood", "place": "Central Thailand",
     "text": "Heavy monsoon rain from July to January. 815 people died. 65 provinces "
             "flooded. Factories north of Bangkok were under water."},
    {"year": 2011, "kind": "flood", "place": "Chiang Mai",
     "text": "A big flood hit Chiang Mai city again. The Ping River rose over "
             "its banks."},
    {"year": 2014, "kind": "earthquake", "place": "Mae Lao, Chiang Rai",
     "text": "5 May 2014. Magnitude 6.1. One person died. Temples cracked. "
             "The strongest earthquake ever recorded in Thailand."},
    {"year": 2024, "kind": "flood", "place": "Mae Sai, Chiang Rai",
     "text": "September 2024. Rain from Typhoon Yagi. Mae Sai town flooded. "
             "More than 1,000 homes were damaged."},
    {"year": 2024, "kind": "flood", "place": "Chiang Mai",
     "text": "26 September 2024. Heavy rain for days. The Ping River rose over its "
             "banks at 4.93 metres. Markets and roads flooded."},
    {"year": 2024, "kind": "flood", "place": "Chiang Mai",
     "text": "5 and 6 October 2024. The Ping River reached 5.30 metres at Nawarat "
             "Bridge, the highest in 50 years. Night Bazaar was under a metre of water."},
    {"year": 2025, "kind": "earthquake", "place": "Myanmar and Thailand",
     "text": "28 March 2025. Magnitude 7.7 on the Sagaing Fault. Felt in Chiang Mai. "
             "A tower under construction fell in Bangkok."},
]


def ramp(value, low, high):
    """0 below low, 1 above high, a straight line between them."""
    if value <= low:
        return 0.0
    if value >= high:
        return 1.0
    return (value - low) / (high - low)


def words(text):
    """The words in a text: lower case, three letters or more, and not a stop word."""
    cleaned = "".join(letter.lower() if letter.isalnum() else " " for letter in text)
    kept = (word for word in cleaned.split() if len(word) >= 3)
    return {word for word in kept if word not in STOP_WORDS}


def search(query, archive):
    """Every record with the count of words it shares with the query, best first."""
    query_words = words(query)
    matches = []
    for record in archive:
        record_text = record["text"] + " " + record["place"] + " " + record["kind"]
        shared = query_words & words(record_text)
        matches.append((len(shared), record, sorted(shared)))
    matches.sort(key=lambda match: match[0], reverse=True)
    return matches


def archive_score(shared_words):
    """One shared word is chance. Four shared words is a real match."""
    return ramp(shared_words, 1, 4)


def main():
    query = "heavy rain for days, the river is rising, markets near the river"
    print("query:", query)
    for count, record, shared in search(query, ARCHIVE)[:3]:
        print(f"  {count} shared {record['year']} {record['kind']:10} "
              f"{record['place']:22} {shared}")
    best_count = search(query, ARCHIVE)[0][0]
    print(f"archive score: {archive_score(best_count):.2f}")


if __name__ == "__main__":
    main()

# step7_fuse.py — one class for every signal, and a coordinator that joins them.
# Five signals, five scores. The coordinator does not add them. It multiplies.
# New here: the Signal class, fuse() and decide().

CRITICAL = 3.70   # at 3.70 m the Irrigation Office warns the city
PRIOR = 0.10      # our starting guess for a flood tomorrow, before we look at any signal


def ramp(value, low, high):
    """0 below low, 1 above high, a straight line between them."""
    if value <= low:
        return 0.0
    if value >= high:
        return 1.0
    return (value - low) / (high - low)


def rain_score(mm):
    return ramp(mm, 20, 80)


def river_score(metres):
    return ramp(metres, 3.0, CRITICAL)


def forecast_score(metres):
    return ramp(metres, 3.3, 3.9)


def satellite_score(fraction):
    return ramp(fraction, 0.05, 0.30)


def archive_score(shared_words):
    return ramp(shared_words, 1, 4)


# The kind of a signal picks the scorer. Same class, different job.
SCORERS = {"rain": rain_score, "river": river_score, "forecast": forecast_score,
           "satellite": satellite_score, "archive": archive_score}


class Signal:
    """One signal. The kind says how its value becomes a score."""

    def __init__(self, kind, value):
        self.kind = kind
        self.value = value

    def score(self):
        """0 is calm, 1 is danger. A missing value is 0.5: it says nothing."""
        if self.value is None:
            return 0.5
        return SCORERS[self.kind](self.value)

    def weight(self):
        """How much this signal multiplies the odds. Score 0.5 gives x1: no change."""
        return 10 ** (2 * self.score() - 1)


def fuse(signals, prior):
    """Start with the prior. Multiply the odds by every signal. Give back a chance."""
    odds = prior / (1 - prior)
    for signal in signals:
        odds = odds * signal.weight()
    return odds / (1 + odds)


def decide(chance, threshold):
    if chance >= threshold:
        return "RED - warn people now"
    if chance >= threshold / 2:
        return "YELLOW - watch closely"
    return "GREEN - calm"


def main():
    # The morning of 24 September 2024 in Chiang Mai. No satellite picture yet.
    signals = [Signal("rain", 98), Signal("river", 3.34), Signal("forecast", 3.63),
               Signal("satellite", None), Signal("archive", 2)]
    for signal in signals:
        print(f"{signal.kind:10} value {str(signal.value):>5}   "
              f"score {signal.score():.2f}   odds x{signal.weight():.2f}")
    chance = fuse(signals, PRIOR)
    print(f"chance of a flood tomorrow: {chance:.0%}")
    print(decide(chance, 0.5))


if __name__ == "__main__":
    main()

# disaster_agent.py — the whole agent, from the eight steps of the Disaster Lab.
# Five flood signals go into one coordinator. The earthquake signal is reported on
# its own, because a quake cannot be forecast, only measured.
# New here: thirty days of data from 2024, and evaluate(), which tests the agent.
#
#   python disaster_agent.py

import math

CRITICAL = 3.70      # at 3.70 m at gauge P.1 the Irrigation Office warns
FLOOD_LEVEL = 3.90   # water enters the first streets at +3.90 m
PRIOR = 0.10         # our starting guess for a flood tomorrow, before any signal
P_SPEED = 5.0        # km per second, a P wave in granite
S_SPEED = 3.0        # km per second, about 60 percent of the P wave

# Chiang Mai, 10 September to 9 October 2024. River levels marked * were reported
# in the news (see the page). The other days are filled in to join them, and the
# rain is a teaching series, not a measurement.
DATE = ["10 Sep", "11 Sep", "12 Sep", "13 Sep", "14 Sep", "15 Sep", "16 Sep", "17 Sep",
        "18 Sep", "19 Sep", "20 Sep", "21 Sep", "22 Sep", "23 Sep", "24 Sep", "25 Sep",
        "26 Sep", "27 Sep", "28 Sep", "29 Sep", "30 Sep", "1 Oct", "2 Oct", "3 Oct",
        "4 Oct", "5 Oct", "6 Oct", "7 Oct", "8 Oct", "9 Oct"]
RAIN = [2, 0, 8, 14, 3, 0, 22, 6, 0, 11, 4, 18, 36, 64, 98, 40,
        12, 5, 0, 2, 0, 9, 28, 55, 120, 35, 6, 0, 0, 3]
RIVER = [2.30, 2.28, 2.30, 2.35, 2.33, 2.30, 2.38, 2.40, 2.36, 2.40, 2.42, 2.55,
         2.80, 3.05, 3.34, 4.32,   # 24 Sep 3.34* at 10:00, 25 Sep 4.32* forecast peak
         4.93, 4.40, 3.90, 3.55, 3.30, 3.15, 3.20, 3.60,   # 26 Sep 4.93* over the banks
         4.40, 5.27, 5.30, 4.70, 3.80, 3.40]   # 5 Oct 5.27*, 6 Oct 5.30*, 8 Oct 3.80*

BEFORE = [
    [7, 8, 7, 6, 8, 7],
    [8, 7, 2, 6, 7, 8],
    [7, 6, 2, 7, 8, 7],
    [8, 7, 2, 6, 7, 6],
    [7, 8, 3, 7, 8, 7],
    [8, 7, 2, 7, 7, 8],
]
AFTER = [
    [7, 8, 7, 6, 8, 7],
    [8, 2, 1, 2, 7, 8],
    [7, 1, 1, 2, 2, 7],
    [2, 1, 1, 2, 1, 6],
    [7, 2, 1, 1, 2, 7],
    [8, 7, 2, 2, 7, 8],
]
BIG_CHANGE = 4
# Words that every text shares. They say nothing, so the search ignores them.
STOP_WORDS = {"the", "and", "for", "with", "from"}

ARCHIVE = [
    {"year": 2004, "kind": "tsunami", "place": "Khao Lak and Phuket",
     "text": "26 December 2004. The sea pulled back from the beach. Then big waves came. "
             "About 5,400 people died in Thailand. There was no warning system."},
    {"year": 2005, "kind": "flood", "place": "Chiang Mai",
     "text": "A big flood hit Chiang Mai city. River water entered the streets."},
    {"year": 2011, "kind": "flood", "place": "Central Thailand",
     "text": "Heavy monsoon rain from July to January. 815 people died. 65 provinces "
             "flooded. Factories north of Bangkok were under water."},
    {"year": 2011, "kind": "flood", "place": "Chiang Mai",
     "text": "A big flood hit Chiang Mai city again. The Ping River rose over "
             "its banks."},
    {"year": 2014, "kind": "earthquake", "place": "Mae Lao, Chiang Rai",
     "text": "5 May 2014. Magnitude 6.1. One person died. Temples cracked. "
             "The strongest earthquake ever recorded in Thailand."},
    {"year": 2024, "kind": "flood", "place": "Mae Sai, Chiang Rai",
     "text": "September 2024. Rain from Typhoon Yagi. Mae Sai town flooded. "
             "More than 1,000 homes were damaged."},
    {"year": 2024, "kind": "flood", "place": "Chiang Mai",
     "text": "26 September 2024. Heavy rain for days. The Ping River rose over its "
             "banks at 4.93 metres. Markets and roads flooded."},
    {"year": 2024, "kind": "flood", "place": "Chiang Mai",
     "text": "5 and 6 October 2024. The Ping River reached 5.30 metres at Nawarat "
             "Bridge, the highest in 50 years. Night Bazaar was under a metre of water."},
    {"year": 2025, "kind": "earthquake", "place": "Myanmar and Thailand",
     "text": "28 March 2025. Magnitude 7.7 on the Sagaing Fault. Felt in Chiang Mai. "
             "A tower under construction fell in Bangkok."},
]


def ramp(value, low, high):
    """0 below low, 1 above high, a straight line between them."""
    if value <= low:
        return 0.0
    if value >= high:
        return 1.0
    return (value - low) / (high - low)


# --- the five flood signals ---

def rain_score(mm):
    return ramp(mm, 20, 80)


def river_score(metres):
    return ramp(metres, 3.0, CRITICAL)


def forecast(levels, day):
    """Tomorrow's level, if the river climbs the same as it did today."""
    if day == 0:
        return levels[0]
    climb = levels[day] - levels[day - 1]
    return levels[day] + climb


def forecast_score(metres):
    return ramp(metres, 3.3, 3.9)


def changed_fraction(before, after):
    """What part of the picture changed? 0.0 is nothing, 1.0 is everything."""
    changed = 0
    total = 0
    for row_before, row_after in zip(before, after):
        for old, new in zip(row_before, row_after):
            total += 1
            if abs(new - old) >= BIG_CHANGE:
                changed += 1
    return changed / total


def satellite_score(fraction):
    return ramp(fraction, 0.05, 0.30)


def words(text):
    """The words in a text: lower case, three letters or more, and not a stop word."""
    cleaned = "".join(letter.lower() if letter.isalnum() else " " for letter in text)
    kept = (word for word in cleaned.split() if len(word) >= 3)
    return {word for word in kept if word not in STOP_WORDS}


def search(query, archive):
    """Every record with the count of words it shares with the query, best first."""
    query_words = words(query)
    matches = []
    for record in archive:
        record_text = record["text"] + " " + record["place"] + " " + record["kind"]
        shared = query_words & words(record_text)
        matches.append((len(shared), record, sorted(shared)))
    matches.sort(key=lambda match: match[0], reverse=True)
    return matches


def archive_score(shared_words):
    return ramp(shared_words, 1, 4)


SCORERS = {"rain": rain_score, "river": river_score, "forecast": forecast_score,
           "satellite": satellite_score, "archive": archive_score}


# --- the earthquake signal, reported on its own ---

def warning_seconds(km):
    """The P wave arrives first. The time before the S wave is the warning time."""
    return km / S_SPEED - km / P_SPEED


def intensity(magnitude, km):
    """How strong the shaking feels, from 1 to 12. A simple classroom rule."""
    value = 1.5 * magnitude - 3 * math.log10(max(km, 1)) + 1.5
    return max(1.0, min(12.0, value))


def quake_report(magnitude, km):
    shaking = intensity(magnitude, km)
    if shaking >= 8:
        advice = "RED - buildings can fall"
    elif shaking >= 6:
        advice = "YELLOW - things fall, walls crack"
    else:
        advice = "GREEN - felt, little damage"
    return (f"earthquake M{magnitude}, {km} km away: shaking {shaking:.1f}, "
            f"warning {warning_seconds(km):.0f} s. {advice}")


# --- the coordinator ---

class Signal:
    """One signal. The kind says how its value becomes a score."""

    def __init__(self, kind, value):
        self.kind = kind
        self.value = value

    def score(self):
        """0 is calm, 1 is danger. A missing value is 0.5: it says nothing."""
        if self.value is None:
            return 0.5
        return SCORERS[self.kind](self.value)

    def weight(self):
        """How much this signal multiplies the odds. Score 0.5 gives x1: no change."""
        return 10 ** (2 * self.score() - 1)


def fuse(signals, prior):
    """Start with the prior. Multiply the odds by every signal. Give back a chance."""
    odds = prior / (1 - prior)
    for signal in signals:
        odds = odds * signal.weight()
    return odds / (1 + odds)


def decide(chance, threshold):
    if chance >= threshold:
        return "RED - warn people now"
    if chance >= threshold / 2:
        return "YELLOW - watch closely"
    return "GREEN - calm"


def evaluate(threshold):
    """Replay the thirty days. Each day, decide for tomorrow. Then look at tomorrow."""
    hits = false_alarms = misses = quiet = 0
    for day in range(1, len(RIVER) - 1):
        signals = [Signal("rain", RAIN[day]), Signal("river", RIVER[day]),
                   Signal("forecast", forecast(RIVER, day))]
        alert = fuse(signals, PRIOR) >= threshold
        flood = RIVER[day + 1] >= FLOOD_LEVEL
        if alert and flood:
            hits += 1
        elif alert:
            false_alarms += 1
        elif flood:
            misses += 1
        else:
            quiet += 1
    return hits, false_alarms, misses, quiet


def main():
    print("threshold  hits  false alarms  misses  quiet")
    for threshold in [0.3, 0.5, 0.7, 0.9]:
        hits, false_alarms, misses, quiet = evaluate(threshold)
        print(f"   {threshold:.1f}    {hits:5}  {false_alarms:12}  {misses:6}  {quiet:5}")

    # The morning of 4 October 2024 in Chiang Mai: every signal we have.
    today = DATE.index("4 Oct")
    query = "heavy rain for days, the river is rising, markets near the river"
    best_count = search(query, ARCHIVE)[0][0]
    signals = [Signal("rain", RAIN[today]), Signal("river", RIVER[today]),
               Signal("forecast", forecast(RIVER, today)),
               Signal("satellite", changed_fraction(BEFORE, AFTER)),
               Signal("archive", best_count)]
    print()
    print("4 Oct 2024, morning")
    for signal in signals:
        print(f"  {signal.kind:10} score {signal.score():.2f}   "
              f"odds x{signal.weight():.2f}")
    chance = fuse(signals, PRIOR)
    print(f"  chance of a flood tomorrow: {chance:.0%}  ->  {decide(chance, 0.5)}")

    # 28 March 2025, 13:20 in Chiang Mai: the Sagaing earthquake, 480 km away.
    print()
    print("28 Mar 2025, 13:20")
    print("  " + quake_report(7.7, 480))


if __name__ == "__main__":
    main()

# live_data.py — four free sources of live signals for the Disaster Lab. No API keys.
#
#   from live_data import rain_days, river_days, quakes, events
#
# Every function asks one website and gives back plain Python: lists of (date, value)
# or lists of dictionaries. The notebook scores them with the same functions it uses
# for the sample data.
#
#   rain_days()   Open-Meteo weather: daily rain, the past days and a forecast
#   river_days()  Open-Meteo flood API: river flow from the GloFAS model, in m3/s
#   quakes()      USGS: earthquakes in the last hours, anywhere in the world
#   events()      NASA EONET: floods, storms and quakes that are open or recent
#
#   pip install requests

import datetime

import requests

CHIANG_MAI = (18.79, 98.98)          # latitude, longitude
PING_GLOFAS = (18.70, 99.02)         # a GloFAS grid cell on the Ping below the city
HEADERS = {"User-Agent": "krueng-disaster-lab/1.0 (educational)"}


def get_json(url, params):
    """Ask a website for JSON. Raises an error if the website says no."""
    reply = requests.get(url, params=params, headers=HEADERS, timeout=30)
    reply.raise_for_status()
    return reply.json()


def rain_days(place=CHIANG_MAI, past=7, future=7):
    """Daily rain in mm as (date, mm) pairs: past days measured, future days forecast."""
    lat, lon = place
    daily = get_json("https://api.open-meteo.com/v1/forecast",
                     {"latitude": lat, "longitude": lon, "daily": "precipitation_sum",
                      "past_days": past, "forecast_days": future,
                      "timezone": "Asia/Bangkok"})["daily"]
    return list(zip(daily["time"], daily["precipitation_sum"]))


def rain_history(place, start, end):
    """Daily rain in mm as (date, mm) pairs for past dates (ERA5 reanalysis)."""
    lat, lon = place
    daily = get_json("https://archive-api.open-meteo.com/v1/archive",
                     {"latitude": lat, "longitude": lon, "start_date": start,
                      "end_date": end, "daily": "precipitation_sum",
                      "timezone": "Asia/Bangkok"})["daily"]
    return list(zip(daily["time"], daily["precipitation_sum"]))


def river_days(place=PING_GLOFAS, past=30, future=7):
    """River flow in m3/s as (date, flow) pairs from the GloFAS model, not a gauge."""
    lat, lon = place
    daily = get_json("https://flood-api.open-meteo.com/v1/flood",
                     {"latitude": lat, "longitude": lon, "daily": "river_discharge",
                      "past_days": past, "forecast_days": future})["daily"]
    return list(zip(daily["time"], daily["river_discharge"]))


def quakes(hours=24, min_magnitude=4.5):
    """Earthquakes from the USGS in the last hours, biggest first.
    Each one is a dictionary: time, magnitude, place, lat, lon, depth_km."""
    now = datetime.datetime.now(datetime.timezone.utc)
    since = now - datetime.timedelta(hours=hours)
    found = get_json("https://earthquake.usgs.gov/fdsnws/event/1/query",
                     {"format": "geojson", "starttime": since.strftime("%Y-%m-%dT%H:%M"),
                      "minmagnitude": min_magnitude, "orderby": "magnitude"})
    result = []
    for feature in found["features"]:
        facts = feature["properties"]
        lon, lat, depth = feature["geometry"]["coordinates"]
        seconds = facts["time"] / 1000
        when = datetime.datetime.fromtimestamp(seconds, datetime.timezone.utc)
        result.append({"time": when.strftime("%Y-%m-%d %H:%M UTC"),
                       "magnitude": facts["mag"], "place": facts["place"],
                       "lat": lat, "lon": lon, "depth_km": depth})
    return result


def events(days=30, kinds="floods,earthquakes,severeStorms,landslides"):
    """Natural events from NASA EONET, newest first.
    Each one is a dictionary: title, kind, date, lat, lon."""
    found = get_json("https://eonet.gsfc.nasa.gov/api/v3/events",
                     {"status": "all", "days": days, "category": kinds, "limit": 500})
    result = []
    for event in found["events"]:
        shape = event["geometry"][-1]
        lat, lon = centre(shape)
        result.append({"title": event["title"], "kind": event["categories"][0]["title"],
                       "date": shape["date"][:10], "lat": lat, "lon": lon})
    return result


def centre(shape):
    """(lat, lon) of an EONET shape: the point itself, or the middle of a polygon's ring.
    EONET writes its points as [lon, lat] but its flood polygons as [lat, lon]. Checked on
    1 October 2026 against a flood in Tak and a fire in Brazil."""
    if shape["type"] == "Point":
        lon, lat = shape["coordinates"]
        return lat, lon
    ring = shape["coordinates"][0]
    lats = [point[0] for point in ring]
    lons = [point[1] for point in ring]
    return sum(lats) / len(lats), sum(lons) / len(lons)