Code Examples

# Install the requests library
pip install requests

import ssl
import socket
import requests
from requests.adapters import HTTPAdapter

# API endpoints
AUTH_HOST = "keycloak.qairbon.com"
API_HOST = "${API_HOST}"
TOKEN_URL = f"https://{AUTH_HOST}/realms/QairbonAPIDev/protocol/openid-connect/token"
API_URL = f"https://{API_HOST}/api/v1"


def _fetch_server_cert(hostname, port=443):
    """Fetch server certificate for pinning (trust-on-first-use).
    Needed because the API uses a Traefik default cert with no matching SAN."""
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    with socket.create_connection((hostname, port)) as sock:
        with ctx.wrap_socket(sock, server_hostname=hostname) as ssock:
            return ssl.DER_cert_to_PEM_cert(ssock.getpeercert(binary_form=True))


class _PinnedCertAdapter(HTTPAdapter):
    """HTTP adapter that verifies against a specific pinned certificate."""

    def __init__(self, cert_pem, **kwargs):
        self._cert_pem = cert_pem
        super().__init__(**kwargs)

    def init_poolmanager(self, *args, **kwargs):
        ctx = ssl.create_default_context(cadata=self._cert_pem)
        ctx.check_hostname = False
        kwargs["ssl_context"] = ctx
        kwargs["assert_hostname"] = False
        super().init_poolmanager(*args, **kwargs)


# Build a shared session with pinned certs — avoids SSLError on first connection
_AUTH_CERT = _fetch_server_cert(AUTH_HOST)
_API_CERT = _fetch_server_cert(API_HOST)

_session = requests.Session()
_session.mount(f"https://{AUTH_HOST}", _PinnedCertAdapter(_AUTH_CERT))
_session.mount(f"https://{API_HOST}", _PinnedCertAdapter(_API_CERT))


class QAirbonClient:
    """QAirbon API client with Keycloak OIDC authentication."""

    def __init__(self, username, password):
        self.username = username
        self.password = password
        self.access_token = None
        self.refresh_token = None
        self._get_token()

    def _get_token(self):
        response = _session.post(
            TOKEN_URL,
            data={
                "grant_type": "password",
                "client_id": "auth",
                "username": self.username,
                "password": self.password,
            }
        )
        response.raise_for_status()
        data = response.json()
        self.access_token = data["access_token"]
        self.refresh_token = data.get("refresh_token")

    def _refresh(self):
        response = _session.post(
            TOKEN_URL,
            data={
                "grant_type": "refresh_token",
                "client_id": "auth",
                "refresh_token": self.refresh_token,
            }
        )
        response.raise_for_status()
        self.access_token = response.json()["access_token"]

    @property
    def headers(self):
        return {"Authorization": f"Bearer {self.access_token}"}

    def get(self, endpoint, params=None):
        response = _session.get(
            f"{API_URL}{endpoint}",
            headers=self.headers,
            params=params
        )
        if response.status_code == 401:
            self._refresh()
            response = _session.get(
                f"{API_URL}{endpoint}",
                headers=self.headers,
                params=params
            )
        response.raise_for_status()
        return response.json()

    def post(self, endpoint, data):
        response = _session.post(
            f"{API_URL}{endpoint}",
            headers=self.headers,
            json=data
        )
        if response.status_code == 401:
            self._refresh()
            response = _session.post(
                f"{API_URL}{endpoint}",
                headers=self.headers,
                json=data
            )
        response.raise_for_status()
        return response.json()

# Initialize client with your Keycloak credentials
client = QAirbonClient("your@email.com", "YOUR_PASSWORD")

# Token refresh is handled automatically by QAirbonClient on 401 responses
# To refresh manually:
client._refresh()

# Get the root STAC catalog
catalog = client.get("/stac")

print(f"Catalog: {catalog['title']}")
print(f"Description: {catalog['description']}")
print(f"STAC Version: {catalog['stac_version']}")

# List conformance classes
for uri in catalog.get('conformsTo', []):
    print(f"  - {uri}")

# Get first page of sites
items = client.get("/collections/sites/items")

print(f"Total sites: {items['context']['matched']}")
print(f"Returned: {items['context']['returned']}")

for site in items['features']:
    props = site['properties']
    print(f"- {props['title']} ({props['qairbon:industry']}, {props['qairbon:country']})")

site_id = "1d55e91b-65e5-4073-b3ba-d44c394318cc"
site = client.get(f"/collections/sites/items/{site_id}")

props = site['properties']
bbox = site['bbox']  # [west, south, east, north] — geometry may be Point or Polygon
lon = (bbox[0] + bbox[2]) / 2
lat = (bbox[1] + bbox[3]) / 2

print(f"Site: {props['title']}")
print(f"Industry: {props['qairbon:industry']}")
print(f"Status: {props['site:status']}")
print(f"Location: {lat:.5f}, {lon:.5f}")
print(f"Country: {props['qairbon:country']}")

def get_all_sites(client, limit=50):
    """Fetch all sites using cursor-based pagination (token = integer offset)."""
    all_sites = []
    token = None

    while True:
        params = {"limit": limit}
        if token is not None:
            params["token"] = token

        result = client.get("/collections/sites/items", params=params)
        all_sites.extend(result["features"])

        next_link = next((l for l in result.get("links", []) if l["rel"] == "next"), None)
        if not next_link:
            break

        token = next_link["href"].split("token=")[1].split("&")[0]
        print(f"Fetched {len(all_sites)} / {result['context']['matched']} sites...")

    return all_sites

# Get all sites
sites = get_all_sites(client)
print(f"Total sites fetched: {len(sites)}")

# Search for sites near ArcelorMittal Méditerranée, Fos-sur-Mer, France
search_request = {
    "collections": ["sites"],
    "bbox": [4.4, 43.0, 5.4, 43.9],  # [west, south, east, north]
    "limit": 100
}

results = client.post("/search", search_request)

print(f"Found {results['context']['returned']} sites in bounding box")
for site in results['features']:
    print(f"  - {site['properties']['title']}")

# Search with a custom polygon (e.g., ArcelorMittal Méditerranée, Fos-sur-Mer, France)
search_request = {
    "intersects": {
        "type": "Polygon",
        "coordinates": [[
            [4.4, 43.0],
            [5.4, 43.0],
            [5.4, 43.9],
            [4.4, 43.9],
            [4.4, 43.0]  # Close the polygon
        ]]
    },
    "limit": 50
}

results = client.post("/search", search_request)
print(f"Sites in polygon: {results['context']['returned']}")

# Filter by industry using the server-side query parameter (values: see Endpoints › Industries)
steel_sites = client.get("/collections/sites/items", params={"industry": "steel_bf", "limit": 100})
print(f"Steel BF plants: {steel_sites['context']['matched']}")

copper_sites = client.get("/collections/sites/items", params={"industry": "copper_smelter_primary", "limit": 100})
print(f"Primary copper smelters: {copper_sites['context']['matched']}")

from requests.exceptions import HTTPError

try:
    # Attempt to get a site
    site = client.get("/collections/sites/items/invalid-id")
except HTTPError as e:
    if e.response.status_code == 400:
        print(f"Bad Request: {e.response.json().get('detail')}")
    elif e.response.status_code == 401:
        print("Authentication failed - token may be expired")
    elif e.response.status_code == 404:
        print("Resource not found")
    else:
        print(f"API Error: {e}")

# List SAI time series for a site
site_id = "1d55e91b-65e5-4073-b3ba-d44c394318cc"
sai_data = client.get(
    f"/collections/sites/items/{site_id}/sai/items",
    params={"limit": 50}
)

print(f"Total SAI records: {sai_data['context']['matched']}")
for item in sai_data["features"]:
    props = item["properties"]
    print(f"{props['datetime'][:10]}: SAI = {props['sai:value']:.3f}")

# Filter SAI records by date range
sai_data = client.get(
    f"/collections/sites/items/{site_id}/sai/items",
    params={"datetime": "2024-01-01/2024-12-31", "limit": 100}
)
print(f"SAI records in 2024: {sai_data['context']['matched']}")
for item in sai_data["features"]:
    props = item["properties"]
    print(f"{props['datetime'][:10]}: SAI = {props['sai:value']:.3f}")

# Open-ended ranges also work:
# "2024-01-01/.."  → from date onwards
# "../2024-12-31"  → up to date

def get_all_sai(client, site_id):
    """Fetch all SAI records for a site, following pagination links."""
    items, token = [], None
    while True:
        params = {"limit": 100}
        if token:
            params["token"] = token
        data = client.get(f"/collections/sites/items/{site_id}/sai/items", params=params)
        items.extend(data["features"])
        next_link = next((l for l in data["links"] if l["rel"] == "next"), None)
        if not next_link:
            break
        token = next_link["href"].split("token=")[1]
    return items

sai_items = get_all_sai(client, site_id)
print(f"Total SAI records: {len(sai_items)}")

# Monthly production (kt) of a site, with the capacity running each month
site_id = "1d55e91b-65e5-4073-b3ba-d44c394318cc"
production = client.get(
    f"/collections/sites/items/{site_id}/production/items",
    params={"datetime": "2025-01-01/..", "limit": 100}
)

for item in production["features"]:
    props = item["properties"]
    label = "now" if item["id"].endswith("_current") else props["datetime"][:7]
    print(f"{label}: {props['production:value']:.1f} kt / {props['production:active_capacity']:.1f} kt capacity")

# Requires CO2 access
site_id = "1d55e91b-65e5-4073-b3ba-d44c394318cc"
plume_data = client.get(
    f"/collections/sites/items/{site_id}/plume/items",
    params={"limit": 20}
)

for item in plume_data["features"]:
    props = item["properties"]
    print(f"{props['datetime'][:10]}: {props['plume:quantification']:.1f} t/h ± {props['plume:uncertainty']:.1f} t/h")
    raster_href = item["assets"]["raster"]["href"]
    print(f"  Raster: {raster_href}")

# Download the color-mapped RGBA GeoTIFF for the first plume event
first_item = plume_data["features"][0]
plume_id = first_item["id"]
raster_url = (
    f"https://${API_HOST}/api/v1/collections/sites/items/{site_id}"
    f"/plume/items/{plume_id}/assets/raster"
)

import requests
r = requests.get(raster_url, headers={"Authorization": f"Bearer {client.access_token}"})
with open(f"plume_{plume_id}.tif", "wb") as f:
    f.write(r.content)

# Open with rasterio to inspect
import rasterio
with rasterio.open(f"plume_{plume_id}.tif") as src:
    print(f"CRS: {src.crs}")
    print(f"Bounds: {src.bounds}")
    print(f"Size: {src.width} x {src.height}")
    print(f"Bands: {src.count} (RGBA)")

# 1. List steel_bf panels
panels = client.get("/collections/panel/items", params={"industry": "steel_bf"})
for panel in panels["features"]:
    print(f"Panel: {panel['properties']['title']} ({panel['id']})")

# 2. Get QAI for the first steel_bf panel
panel_id = panels["features"][0]["id"]
qai_data = client.get(
    f"/collections/panel/items/{panel_id}/qai/items",
    params={"limit": 100}
)

print(f"QAI records: {qai_data['context']['matched']}")
for item in qai_data["features"]:
    props = item["properties"]
    print(f"{props['datetime'][:7]}: QAI = {props['qai:value']:.3f}")

# 1. Monthly production of a panel (panel_id from the QAI example above)
production = client.get(
    f"/collections/panel/items/{panel_id}/production/items",
    params={"limit": 100}
)
for item in production["features"]:
    props = item["properties"]
    label = "now" if item["id"].endswith("_current") else props["datetime"][:7]
    print(f"{label}: {props['production:value']:.1f} kt over {props['production:site_count']} sites")

# 2. Who produced the last complete month: one asset per site
last_month = [f for f in production["features"] if not f["id"].endswith("_current")][-1]
for site_id, asset in sorted(last_month["assets"].items(), key=lambda kv: -kv[1]["production:weight"]):
    kt = asset.get("production:value")  # absent for a site you cannot access
    shown = f"{kt:.1f} kt" if kt is not None else "—"
    print(f"  {asset['title']}: weight {asset['production:weight']:.1%}, {shown}")

# 1. Confirmed alerts since July 2026
alerts = client.get("/collections/alerts/items", params={"datetime": "2026-07-01/..", "limit": 100})
for a in alerts["features"]:
    p = a["properties"]
    if p["alert:level"] == "confirmed":
        print(p["datetime"][:10], p["title"], p["alert:event"])

# 2. Every alert on one site's blast furnaces
site_id = alerts["features"][0]["properties"]["qairbon:site_id"]
history = client.get(f"/collections/sites/items/{site_id}/alerts/items")
print(f"{history['context']['matched']} alerts on site {site_id}")

from collections import defaultdict

news = client.get("/collections/news/items", params={"datetime": "2026-09-01/..", "limit": 200})
by_site = defaultdict(list)
for item in news["features"]:
    p = item["properties"]
    by_site[p["news:site_name"]].append(p)

for site, articles in by_site.items():
    print(f"{site} ({len(articles)} articles)")
    for p in articles:
        print(f"  {p['datetime'][:10]} [{p['news:source']}] {p['title']}")
        print(f"    {p['description']} — {p['news:url']}")

# Binary response: call the session directly — client.post() parses JSON
site_ids = [s["id"] for s in client.get("/collections/sites/items", params={"industry": "steel_bf", "limit": 10})["features"]]

response = _session.post(
    f"{API_URL}/export/sites",
    json={"site_ids": site_ids, "start_date": "2025-01-01T00:00:00Z", "end_date": "2025-12-31T00:00:00Z"},
    headers=client.headers,
)
response.raise_for_status()
with open("sites.xlsx", "wb") as f:
    f.write(response.content)

# Read it back with pandas: two header rows (site, then metric)
import pandas as pd
sai = pd.read_excel("sites.xlsx", sheet_name="SAI", header=[0, 1], index_col=0)
production = pd.read_excel("sites.xlsx", sheet_name="Production", header=[0, 1], index_col=0)

# Same call with panel ids, over complete months
panel_ids = [p["id"] for p in client.get("/collections/panel/items", params={"industry": "steel_bf"})["features"]]

response = _session.post(f"{API_URL}/export/panels", json={"panel_ids": panel_ids, "start_date": "2025-10-01T00:00:00Z", "end_date": "2026-09-30T00:00:00Z"}, headers=client.headers)
response.raise_for_status()
with open("panels.xlsx", "wb") as f:
    f.write(response.content)