pycitizen

Async-first Python SDK for Citizen's public, unauthenticated incident API — geographic incident discovery via vector tiles, incident details, batch retrieval, related incidents, news feeds, and read-only chat history.

Reverse-engineered from Citizen Android 0.1308.0 (sp0n.citizen, build 1140). Every endpoint implemented here was verified reachable without an access token.

Status: experimental. Unofficial SDK, not affiliated with Citizen. The endpoints are undocumented and may change or become restricted at any time. Use politely — keep request rates low, cache aggressively, and honor the rate limiter defaults.

Features

• Geographic discovery — fetch every incident in a bounding box with one request per covering Mapbox vector tile (/v1/tile/incidents/{x}/{y}/{z}.pbf), decoded by a built-in dependency-free MVT parser.
• Incident details — v1/v2/v3 detail endpoints unified into one Incident model; batch retrieval for up to 50 IDs per request.
• News — curated news feed and news briefings per service area.
• Auxiliary — service-area lookup, safety status, reverse geocoding, place search, incident content/map sources, read-only chat history, neighborhood trends (raw).
• IncidentFeed — polling tracker with deduplication, field-level change detection, lifecycle-transition history, and graceful stale/removed handling. Designed to wrap in a DataUpdateCoordinator (e.g. Home Assistant).
• Async-first — aiohttp, injectable sessions, token-bucket rate limiting, bounded concurrency, retries with exponential backoff honoring Retry-After.
• Fully typed — py.typed, dataclass models, enum severity/lifecycle values matching the app's own DTOs.

Installation

pip install pycitizen

Requires Python 3.11+. Only runtime dependency is aiohttp.

Quick start

import asyncio
from pycitizen import CitizenClient

# (west, south, east, north) — e.g. Manhattan
BBOX = (-74.02, 40.70, -73.93, 40.80)

async def main() -> None:
    async with CitizenClient() as client:
        # 1. Discover every incident marker in the bbox via vector tiles
        markers = await client.get_incident_markers(BBOX)
        for m in markers:
            print(m.incident_id, m.title, m.severity.value, m.timestamp)

        # 2. Hydrate full details (batch or single)
        incidents = await client.get_incidents([m.incident_id for m in markers])
        for inc in incidents:
            print(inc.title, inc.location, inc.lifecycle_state.value)

asyncio.run(main())

Tracking a feed

from pycitizen import CitizenClient, IncidentFeed, FeedState

async with CitizenClient() as client:
    feed = IncidentFeed(client, BBOX, expire_after=900)

    while True:
        diff = await feed.update()
        for t in diff.added:
            print("NEW:", t.marker.title)
        for t in diff.updated:
            print("UPDATED:", t.marker.title, "->", t.marker.lifecycle_state.value)
        for t in diff.removed:
            print("GONE:", t.marker.title)
        await asyncio.sleep(60)

TrackedIncident retains first_seen, last_seen, and a lifecycle_history of (timestamp, LifecycleState) transitions (reported → verified → developing → resolved/inactive).

API surface

client.get_json(path, params) is a low-level escape hatch for anything not wrapped.

Examples

Runnable live demos of the whole public surface — see examples/:

python examples/nearby_incidents.py    # the full incident pipeline
python examples/feed_watcher.py        # IncidentFeed diffing
python examples/map_layers.py          # historical / offender / place tiles
python examples/news_and_trends.py     # news + neighborhood trends
python examples/misc_public.py         # users, usernames, stats, settings

What this SDK does not do

• No authentication flows. Citizen's private endpoints (homescreen feed/mapIncidents, search, friends, variable_settings, user endpoints) return 401 without a user token obtained via phone-OTP sign-in. Not implemented.
• No WebSocket. wss://data.sp0n.io/websocket only carries chat and Protect subscription traffic — not the incident feed — and rejects unauthenticated method calls (auth required).
• No push notifications. Alerts arrive via FCM tied to a registered device token; there is no public push channel.
• No mutations. Posting incidents, comments, likes, follows — all auth-gated and out of scope.

Real-time updates = poll the tiles. They are the same source the app's map consumes.

Configuration

from pycitizen import CitizenClient, RateLimiter, RetryPolicy

client = CitizenClient(
    rate_limiter=RateLimiter(rate=2.0, burst=2, max_concurrent=2),
    retry_policy=RetryPolicy(max_attempts=4),
    timeout=15.0,
)

Inject an existing session (e.g. Home Assistant's shared one) with CitizenClient(session=session); close() then becomes a no-op.

Development

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest            # fully offline test suite
ruff check src tests   # lint
mypy            # strict type check (8 modules, zero errors)
python -m build # sdist + wheel

The test suite never touches the network: HTTP is stubbed and the vector-tile path is exercised against a real tile captured from the live API (tests/fixtures/incidents_tile.pbf).

License

MIT — see LICENSE.