Earthquake Logger For Dummies

Quick Start

Install

pip install genro-bag genro-bag-contrib-resolvers

Run the example

# Clone or download the example file, then:
python earthquake_resolver_example.py

Expected output

  [New Quake] nc/75337592.00 M1.45 -- 4 km NNW of The Geysers, CA
  [New Quake] us/6000sm0b.00 M3.2 -- 58 km NNW of Charlotte Amalie, U.S. Virgin Islands
  [New Quake] us/6000sm21.00 M5.8 -- 121 km WNW of Ternate, Indonesia
  [New Quake] ci/41084399.00 M1.03 -- 7 km SSE of Home Gardens, CA
  [New Quake] ew/1742200140.00 M1.64 -- 15 km N of Truckee, CA
  [New Quake] nc/75337587.00 M0.62 -- 8 km W of Cobb, CA
  [New Quake] ak/025c68p6oj.00 M1.7 -- 90 km NW of Yakutat, Alaska
  [New Quake] nc/75337582.00 M1.35 -- 1 km ESE of The Geysers, CA
  [New Quake] nc/75337577.00 M0.59 -- 9 km WNW of The Geysers, CA

Quakes loaded: 9
Listening for new earthquakes (refresh every 60s)...
Press Ctrl+C to stop.

  [New Quake] nc/75337600.00 M2.1 -- 3 km SE of Mammoth Lakes, CA
  [Update Quake] us/6000sm21.01 M5.9 -- 121 km WNW of Ternate, Indonesia

The logger loads the current feed immediately, then checks every 60 seconds. When USGS updates an earthquake (revised magnitude, location), it appears as a new version (01, 02...).

The complete source

import time
from genro_bag import Bag
from genro_bag.resolvers.contrib import EarthquakeResolver

LOG_TEMPLATE = "  [{tag}] {where}/{event_id}.{version} M{mag} -- {place}"


class EarthquakeLogger:
    """Monitors USGS earthquake feed with versioning and logging."""

    def __init__(self, interval: int = 60):
        self.bag = Bag()
        self.bag["quakes"] = Bag()
        self.bag["quakes"].subscribe("logger", insert=self.log_event)
        self.bag["feed"] = EarthquakeResolver(cache_time=-interval)
        self.bag.subscribe("poll_feed", timer=self.process_feed, interval=interval)
        self.process_feed()

    def process_feed(self, **kw):
        """Process raw feed into versioned quakes."""
        features = self.bag["feed.features"]
        quakes = self.bag["quakes"]

        for fid, attrs, updated in features.query("#k,#a,#a.updated"):
            key = f"{fid[:2]}.{fid[2:]}"
            if key not in quakes:
                quakes[key] = Bag()
            elif quakes[key].nodes[-1].attr['updated'] == updated:
                continue
            event_bag = quakes[key]
            version = f"{len(event_bag):02d}"
            event_bag.set_item(version, None, **attrs)

    def log_event(self, node=None, pathlist=None, **_kw):
        """Log new or updated earthquake events."""
        if not node.attr:
            return
        record = dict(node.attr)
        record.setdefault("place", "?")
        record.setdefault("mag", "?")
        record["where"] = pathlist[0]
        record["event_id"] = pathlist[1]
        record["version"] = node.label
        record["tag"] = "New Quake" if node.label == "00" else "Update Quake"
        print(LOG_TEMPLATE.format(**record))


if __name__ == "__main__":
    logger = EarthquakeLogger(interval=60)
    count = len(list(logger.bag["quakes"].query("#k", deep=True, branch=False)))
    print(f"\nQuakes loaded: {count}")
    print("Listening for new earthquakes (refresh every 60s)...")
    print("Press Ctrl+C to stop.\n")
    try:
        while True:
            time.sleep(10)
    except KeyboardInterrupt:
        print("\nStopped.")

Architecture

The Bag Tree

The entire application state lives in a single Bag. Here is its structure:

How the pieces connect

There are three independent mechanisms working together through the Bag:

The reactive chain

When the timer fires:

# 1. Timer fires → process_feed runs
# 2. process_feed reads bag["feed.features"]
#    → Resolver checks cache. If expired, fetches from USGS.
#    → Returns Bag of earthquake features.
# 3. process_feed writes to bag["quakes.us.7000n1b2"]
#    → set_item creates node "01" with earthquake attrs
#    → Insert subscription fires → log_event runs
# 4. log_event formats and prints the earthquake info

The EarthquakeResolver

EarthquakeResolver extends UrlResolver (which extends BagResolver). It fetches the USGS "all earthquakes, past hour" GeoJSON feed and converts it into a Bag.

Inheritance chain

The source code

class EarthquakeResolver(UrlResolver):
    """Fetches USGS earthquake feed with If-Modified-Since conditional requests."""
    class_kwargs = {**UrlResolver.class_kwargs, "url": USGS_FEED}
    class_args = ["url"]

    def init(self):
        self._last_modified = None

    def prepare_headers(self):
        if self._last_modified:
            return {"If-Modified-Since": self._last_modified}
        return {}

    def process_response(self, response):
        if response.status_code == 304:
            return self.cached_value
        response.raise_for_status()
        self._last_modified = response.headers.get("Last-Modified")
        data = response.json()
        result = Bag()
        result["count"] = len(data["features"])
        result["title"] = data["metadata"]["title"]
        features = Bag()
        for feature in data["features"]:
            props = feature["properties"]
            fid = feature["id"]
            features.set_item(fid, None,
                place=props.get("place", "Unknown"),
                mag=props.get("mag", 0),
                time=props.get("time", 0),
                updated=props.get("updated", 0))
        result["features"] = features
        return result

How it works, step by step

Active cache (cache_time=-60, async only)

The resolver is assigned with a negative cache_time:

self.bag["feed"] = EarthquakeResolver(cache_time=-60)

Conditional requests

The If-Modified-Since header tells USGS: "only send data if it changed since my last fetch." If nothing changed, USGS returns HTTP 304 (Not Modified) with no body — saving bandwidth and parse time.

Subscriptions

The Earthquake Logger uses two subscriptions that serve completely different purposes:

Timer subscription: poll_feed

self.bag.subscribe("poll_feed", timer=self.process_feed, interval=60)

The timer subscription creates a background thread that calls process_feed() every 60 seconds. The callback signature is flexible — it receives keyword arguments that process_feed absorbs with **kw.

Insert subscription: logger

self.bag["quakes"].subscribe("logger", insert=self.log_event)

The insert subscription fires every time a new node is added anywhere under bag["quakes"], including nested paths like quakes.us.7000n1b2.01.

The log_event callback

def log_event(self, node=None, pathlist=None, **_kw):
    if not node.attr:
        return
    record = dict(node.attr)
    record.setdefault("place", "?")
    record.setdefault("mag", "?")
    record["where"] = pathlist[0]       # "us"
    record["event_id"] = pathlist[1]   # "7000n1b2"
    record["version"] = node.label      # "00" or "01"
    record["tag"] = "New Quake" if node.label == "00" else "Update Quake"
    print(LOG_TEMPLATE.format(**record))

Why the guard clause?

if not node.attr:
    return

When process_feed creates an intermediate Bag with quakes[key] = Bag(), that insert also triggers the subscription. But intermediate Bags have no attributes. The guard skips those and only logs leaf nodes (the actual earthquake versions with mag, place, etc.).

Versioning

USGS updates earthquake data in-place: magnitude gets revised, location gets refined, timing gets adjusted. The process_feed method turns these updates into an append-only version history.

The process_feed method

def process_feed(self, **kw):
    features = self.bag["feed.features"]
    quakes = self.bag["quakes"]

    for fid, attrs, updated in features.query("#k,#a,#a.updated"):
        key = f"{fid[:2]}.{fid[2:]}"          # "us7000n1b2" → "us.7000n1b2"
        if key not in quakes:
            quakes[key] = Bag()             # first time: create container
        elif quakes[key].nodes[-1].attr['updated'] == updated:
            continue                        # same version: skip
        event_bag = quakes[key]
        version = f"{len(event_bag):02d}"   # 00, 01, 02, ...
        event_bag.set_item(version, None, **attrs)

Key splitting

USGS earthquake IDs follow a convention: the first 2 characters identify the seismic network, and the rest is the event ID. The code splits this into a two-level path:

This creates a natural grouping by network: all USGS events under us/, all Northern California events under nc/, etc. The dot in the key (f"{fid[:2]}.{fid[2:]}") leverages Bag's dot-path access to auto-create the hierarchy.

Deduplication

The updated timestamp from USGS tells us if anything changed. The logic has three paths:

Version numbering

version = f"{len(event_bag):02d}"   # 00, 01, 02, ...

The version label is the current count of nodes in the event Bag, zero-padded to 2 digits. Since we only append (never delete), this gives a monotonically increasing version history:

The query that drives it all

features.query("#k,#a,#a.updated")

This single query call extracts three things from each node in the feed:

The result is an iterable of tuples: (fid, attrs, updated). This eliminates the need to manually iterate nodes and extract attributes.

Cheat Sheet

Everything used in the Earthquake Logger, in "I want to..." format.

Setup and Wiring

Reading Data

Writing Data

Resolver Patterns

Subscription Callbacks

Query Specifiers

Async Version

The same Earthquake Logger can run on an asyncio event loop with minimal changes. The key difference: set_interval detects the async context and uses asyncio.create_task instead of threads.

What changes

The async example

# earthquake_resolver_async_example.py

import asyncio

from genro_bag import Bag
from genro_bag.resolvers.contrib import EarthquakeResolver

LOG_TEMPLATE = "  [{tag}] {where}/{event_id}.{version} M{mag} -- {place}"


class EarthquakeLogger:

    def __init__(self, interval: int = 60):
        self.bag = Bag()

        # Quakes bag + subscription
        self.bag["quakes"] = Bag()
        self.bag["quakes"].subscribe("logger", insert=self.log_event)

        # Feed resolver + timer subscription
        self.bag["feed"] = EarthquakeResolver(cache_time=-interval)
        self.bag.subscribe("poll_feed", timer=self.process_feed, interval=interval)

    def process_feed(self, **kw):
        features = self.bag["feed.features"]
        quakes = self.bag["quakes"]
        for fid, attrs, updated in features.query("#k,#a,#a.updated"):
            key = f"{fid[:2]}.{fid[2:]}"
            if key not in quakes:
                quakes[key] = Bag()
            elif quakes[key].nodes[-1].attr['updated'] == updated:
                continue
            event_bag = quakes[key]
            version = f"{len(event_bag):02d}"
            event_bag.set_item(version, None, **attrs)

    def log_event(self, node=None, pathlist=None, **_kw):
        if not node.attr:
            return
        record = dict(node.attr)
        record.setdefault("place", "?")
        record.setdefault("mag", "?")
        record["where"] = pathlist[0]
        record["event_id"] = pathlist[1]
        record["version"] = node.label
        record["tag"] = "New Quake" if node.label == "00" else "Update Quake"
        print(LOG_TEMPLATE.format(**record))


async def main():
    logger = EarthquakeLogger(interval=10)
    print("Listening for new earthquakes (refresh every 10s)...")
    print("Press Ctrl+C to stop.\n")
    await asyncio.Event().wait()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nStopped.")

Sync vs Async — side by side

The contract

How set_interval adapts