# 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()