Guide · Python · intermediate
This builds a small service that warns you when lightning is detected within a radius of a fixed point. It starts with a polling loop you can run in a few minutes, then shows the push version for when a polling interval is too long to wait.
You have a point and a radius. You want to know when a flash lands inside it, once per event rather than once per poll. Everything else is bookkeeping: track what you have already alerted on, and decide how long quiet has to last before you call an all-clear.
Request a bounding box that comfortably contains your circle, then filter to the true radius in code. A box is cheap to ask for and the filtering is a few lines.
poll_alert.py
import math, time, requests
API_KEY = "lapi_your_key_here"
LAT, LON = 35.47, -97.52
RADIUS_KM = 16.0
POLL_SECONDS = 60
def km_between(lat1, lon1, lat2, lon2):
r = 6371.0
p1, p2 = math.radians(lat1), math.radians(lat2)
dp = math.radians(lat2 - lat1)
dl = math.radians(lon2 - lon1)
a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
return 2 * r * math.asin(math.sqrt(a))
def bounding_box(lat, lon, km):
dlat = km / 111.0
dlon = km / (111.0 * math.cos(math.radians(lat)))
return lat - dlat, lat + dlat, lon - dlon, lon + dlon
def poll(seen):
min_lat, max_lat, min_lon, max_lon = bounding_box(LAT, LON, RADIUS_KM)
res = requests.get(
"https://api.lightningapi.dev/v1/flashes",
headers={"X-API-Key": API_KEY},
params={
"since_minutes": 15,
"min_lat": round(min_lat, 4), "max_lat": round(max_lat, 4),
"min_lon": round(min_lon, 4), "max_lon": round(max_lon, 4),
},
timeout=15,
)
res.raise_for_status()
for f in res.json().get("flashes", []):
if f["flash_id"] in seen:
continue
seen.add(f["flash_id"])
d = km_between(LAT, LON, f["lat"], f["lon"])
if d <= RADIUS_KM:
print("ALERT", f["flash_timestamp_utc"], round(d, 1), "km")
if __name__ == "__main__":
seen = set()
while True:
try:
poll(seen)
except Exception as exc:
print("poll failed:", exc)
time.sleep(POLL_SECONDS)The seen set grows without bound in this example. In anything long running, evict entries older than your since_minutes window or the process will slowly eat memory.
Once a minute is 43,200 calls in a thirty day month. That is inside the Pay As You Go monthly cap of 100,000 but well past the 5,000 free calls, so at $0.003 per call it is real money for a single watched point. It also means your worst case delay is a full poll interval.
The push version removes both problems. One connection, flashes arrive as they are detected, and there is no interval to tune. Streaming starts on the Ultimate plan.
stream_alert.py
import asyncio, json, math, websockets
API_KEY = "lapi_your_key_here"
LAT, LON = 35.47, -97.52
RADIUS_KM = 16.0
URL = "wss://api.lightningapi.dev/v1/stream?cg_only=true"
def km_between(lat1, lon1, lat2, lon2):
r = 6371.0
p1, p2 = math.radians(lat1), math.radians(lat2)
dp = math.radians(lat2 - lat1)
dl = math.radians(lon2 - lon1)
a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
return 2 * r * math.asin(math.sqrt(a))
async def main():
async for ws in websockets.connect(
URL, additional_headers={"X-API-Key": API_KEY}
):
try:
async for raw in ws:
msg = json.loads(raw)
# The stream sends batches with an envelope, not bare
# flashes. Other message types carry resume bookkeeping.
if msg.get("type") != "flashes":
continue
for f in msg["flashes"]:
d = km_between(LAT, LON, f["lat"], f["lon"])
if d <= RADIUS_KM:
print("ALERT", f["flash_timestamp_utc"], round(d, 1), "km")
except websockets.ConnectionClosed:
continue # the async for reconnects with backoff
asyncio.run(main())If you would rather not run a process at all, define a zone and give it a webhook. The radius check happens server side and you get a signed delivery when something lands inside. Zone deliveries do not count against your monthly call quota.
Frequently asked questions
If you can accept a delay of up to a minute, poll. If you cannot, stream. If you would rather not run a service, use a zone webhook and let the API do the distance check.
Queries take a bounding box. A box slightly larger than your circle costs nothing extra to request, and the distance filter is four lines.
Keep the flash_id of the last flash you handled and pass it as resume_from_flash_id when you reconnect. The server replays what you missed before resuming live. Replayed flashes count toward your quota.
Not necessarily. Zone webhooks are delivered in near real-time without you holding a connection open, and they are free of quota.
For ground safety, usually yes. In-cloud flashes raise the count without changing the risk to people and equipment at the surface.
Related
Last checked 2026-09-20
Coverage areas
Lightning data provided as-is; not for safety-critical use. Commercial use is permitted on every current plan. Read the EULA →