Send USGS Data to Kafka

src/sasquatchbackpack/cli.py

@@ -1,3 +1,4 @@
+import json
 from datetime import timedelta
 
 import click
@@ -10,6 +11,8 @@
 
 DEFAULT_MAGNITUDE_BOUNDS = (2, 10)
 
+DEFAULT_TEST = False
+
 
 def check_duration(
     ctx: click.Context, param: dict, value: tuple[int, int]
@@ -155,11 +158,20 @@ def main() -> None:
     show_default=True,
     callback=check_magnitude_bounds,
 )
+@click.option(
+    "-t",
+    "--test",
+    help="set to True to echo API results without sending them",
+    default=DEFAULT_TEST,
+    type=bool,
+    show_default=True,
+)
 def usgs_earthquake_data(
     duration: tuple[int, int],
     radius: int,
     coords: tuple[float, float],
     magnitude_bounds: tuple[int, int],
+    test: bool,
 ) -> None:
     """Seaches USGS databases for relevant earthquake data and prints it
     to console
@@ -180,10 +192,74 @@ def usgs_earthquake_data(
             click.echo(result)
         click.echo("------")
     else:
-        click.secho("SUCCESS", fg="orange")
+        click.secho("SUCCESS! (kinda)", fg="yellow")
         click.echo("------")
         click.echo("No results found for the provided criteria :(")
         click.echo("------")
+        return
+
+    if not test:
+        click.echo("Sending data...")
+
+        # sasquatch_rest_proxy_url = (
+        #     "https://data-int.lsst.cloud/sasquatch-rest-proxy"
+        # )
+
+        namespace = "lsst.example"
+        topic_name = "usgs-earthquake-data"
+
+        value_schema = json.dumps(
+            {
+                "namespace": f"{namespace}",
+                "type": "record",
+                "name": f"{topic_name}",
+                "description": "Collection of earthquakes near the "
+                + "summit",
+                "fields": [
+                    {"name": "timestamp", "type": "long"},
+                    {"name": "id", "type": "str"},
+                    {"name": "latitude", "type": "float"},
+                    {"name": "longitude", "type": "float"},
+                    {"name": "depth", "type": "float", "units": "Km"},
+                    {"name": "magnitude", "type": "float"},
+                ],
+            }
+        )
+
+        # url = f"{sasquatch_rest_proxy_url}/topics/{namespace}.{topic_name}"
+
+        # headers = {
+        #     "Content-Type": "application/vnd.kafka.avro.v2+json",
+        #     "Accept": "application/vnd.kafka.v2+json",
+        # }
+
+        records = []
+
+        for result in results:
+            records.append(
+                {
+                    "value": {
+                        "timestamp": result.time.strftime("%s"),
+                        "id": result.id,
+                        "latitude": result.latitude,
+                        "longitude": result.longitude,
+                        "depth": float(result.depth),
+                        "magnitude": float(result.magnitude),
+                    }
+                }
+            )
+
+        payload = {"value_schema": value_schema, "records": records}
+
+        print(payload)
+
+        # response = requests.request("POST",
+        #                             url,
+        #                             json=payload,
+        #                             headers=headers
+        #                             )
+
+        # print(response.text)
 
 
 if __name__ == "__main__":

test.json

@@ -0,0 +1,30 @@
+{
+    "namespace": "lsst.example",
+    "type": "record",
+    "name": "usgs-earthquake-data",
+    "description": "Collection of earthquakes near the summit",
+    "fields": [
+        {
+            "name": "timestamp",
+            "type": "long"
+        },
+        {
+            "name": "latitude",
+            "type": "float"
+        },
+        {
+            "name": "longitude",
+            "type": "float"
+        },
+        {
+            "name": "distance",
+            "type": "float",
+            "description": "Distance from the summit",
+            "units": "Km"
+        },
+        {
+            "name": "magnitude",
+            "type": "float"
+        }
+    ]
+}