# Examples

> Complete scripts for the common scenarios, in Node.js (TypeScript) and Python. Each one
> is type-checked or compiled on every docs build, and the code on the guide pages comes
> from the same files.

## Setup

- An API token; see [Authentication](/api/authentication). The scripts read it from
  `PARKLIO_API_TOKEN`, and ids from `PRODUCT_ID` and `LOT_ID`.
- **Node.js 24+**, which runs TypeScript files directly, and `socket.io-client` 4.
- **Python 3.10+**, with `httpx` and `python-socketio` 5.

```bash title="Install and configure"
npm install socket.io-client          # Node.js
pip install httpx "python-socketio[client]"   # Python

export PARKLIO_API_TOKEN="<your-api-token>"
export PRODUCT_ID=123
export LOT_ID=7
```

Run a script with `node open.ts` or `python3 open.py`.

## Open a product and wait for the result

Connect to realtime, send `open`, and wait for the `lot_change` or `gateway_error` with
your `event_id`. Explained in the [Quickstart](/get-started/quickstart).

**Node.js**

```ts title="open.ts"
import { io } from 'socket.io-client';

const token = process.env.PARKLIO_API_TOKEN;
const productId = process.env.PRODUCT_ID;

interface OperationEvent {
  event_id?: string;
  type?: number | 'Timeout';
  message?: string;
  data: Record<string, any>;
}
type Result = { ok: boolean; event: OperationEvent };

// 1. Connect first. `connect` fires once the server has accepted the token.
const socket = io('https://api.parklio.com/v2/realtime', {
  path: '/socketio',
  query: { token },
});
await new Promise((resolve, reject) => {
  socket.once('connect', () => resolve(undefined));
  socket.once('connect_error', reject);
});

// 2. Collect results by event_id. One can arrive before the HTTP response does.
const results = new Map<string, Result>();
const waiters = new Map<string, (result: Result) => void>();
const settle = (ok: boolean) => (raw: string) => {
  const event: OperationEvent = JSON.parse(raw); // payloads are JSON strings
  if (!event.event_id) return; // a status update nobody requested
  const waiter = waiters.get(event.event_id);
  if (waiter) waiter({ ok, event });
  else results.set(event.event_id, { ok, event });
};
socket.on('lot_change', settle(true));
socket.on('gateway_error', settle(false));

// 3. Send the command.
const res = await fetch(`https://api.parklio.com/v2/products/${productId}/state`, {
  method: 'POST',
  headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
  body: JSON.stringify({ operation: 'open' }),
});
const body = await res.json();
if (!body.success) throw new Error(JSON.stringify(body.errors));
const eventId: string = body.data.event_id;
console.log('Sent, event_id', eventId);

// 4. Wait for the matching event.
const { ok, event } = await new Promise<Result>((resolve) => {
  const early = results.get(eventId);
  if (early) resolve(early);
  else waiters.set(eventId, resolve);
});
console.log(ok ? `Done: ${JSON.stringify(event.data.barrier ?? event.data)}` : `Failed: ${event.message ?? event.type}`);
socket.close();
```

**Python**

```python title="open.py"
import json
import os
import threading

import httpx
import socketio

TOKEN = os.environ["PARKLIO_API_TOKEN"]
PRODUCT_ID = os.environ["PRODUCT_ID"]
NS = "/v2/realtime"

sio = socketio.Client()
results = {}
event_id = None
done = threading.Event()
lock = threading.Lock()


def settle(ok):
    def handler(raw):
        global event_id
        event = json.loads(raw)  # payloads are JSON strings
        if not event.get("event_id"):
            return  # a status update nobody requested
        with lock:
            results[event["event_id"]] = (ok, event)
            if event["event_id"] == event_id:
                done.set()
    return handler


sio.on("lot_change", settle(True), namespace=NS)
sio.on("gateway_error", settle(False), namespace=NS)

# 1. Connect first. connect() returns once the server has accepted the token.
sio.connect(
    f"https://api.parklio.com?token={TOKEN}",
    namespaces=[NS],
    socketio_path="socketio",
    wait_timeout=10,
)

# 2. Send the command.
res = httpx.post(
    f"https://api.parklio.com/v2/products/{PRODUCT_ID}/state",
    headers={"Authorization": f"Bearer {TOKEN}"},
    json={"operation": "open"},
)
body = res.json()
if not body["success"]:
    raise SystemExit(body["errors"])
with lock:
    event_id = body["data"]["event_id"]
    if event_id in results:  # it arrived before the HTTP response
        done.set()
print("Sent, event_id", event_id)

# 3. Wait for the matching event.
done.wait(timeout=180)
ok, event = results.get(event_id, (False, {"type": "no result"}))
print("Done" if ok else f"Failed: {event.get('message', event.get('type'))}")
sio.disconnect()
```

## Monitor devices

Alert on offline products, error flags and low batteries from status updates, and watch
Gateways. Explained in [Monitor devices](/guides/monitor-devices).

**Node.js**

```ts title="monitor-devices.ts"
// The monitor-devices guide: status updates, offline products, Gateways.
import { io } from 'socket.io-client';

const socket = io('https://api.parklio.com/v2/realtime', {
  path: '/socketio',
  query: { token: process.env.PARKLIO_API_TOKEN },
});

// alert() and resolve() stand for your alerting.
const last = new Map<number, { online: boolean; errorCode: string | null; low: boolean }>();

socket.on('lot_change', (raw: string) => {
  const { event_id, data } = JSON.parse(raw);
  if (event_id) return; // command results are handled elsewhere

  const online = data.status === 'online';
  const errorCode: string | null = data.product_error?.error_code ?? null;
  const low = isLowBattery(data.battery_status);
  const prev = last.get(data.id);
  last.set(data.id, { online, errorCode, low });

  if (!online && prev?.online) alert(data, 'offline');
  if (online && prev?.online === false) resolve(data, 'back online');
  if (errorCode && errorCode !== prev?.errorCode) alert(data, `error flags ${errorCode}`, data.product_error);
  if (!errorCode && prev?.errorCode) resolve(data, 'error cleared');
  if (low && !prev?.low) alert(data, `battery at ${data.battery_status}%`);
});

function isLowBattery(value: unknown) {
  // Only barriers report a battery level. 0–100 is percent; above 100 means grid power.
  return typeof value === 'number' && value <= 20;
}

socket.on('gateway_status', (raw: string) => {
  const { uuid, status } = JSON.parse(raw);
  if (status === 'offline') alert({ uuid }, 'Gateway offline');
});

// Your alerting.
function alert(subject: Record<string, any>, what: string, details?: unknown) {
  console.warn('ALERT', subject.id ?? subject.uuid, what, details ?? '');
}
function resolve(subject: Record<string, any>, what: string) {
  console.info('RESOLVED', subject.id ?? subject.uuid, what);
}
```

**Python**

```python title="monitor_devices.py"
"""The monitor-devices guide: status updates, offline products, Gateways."""
import json
import os

import socketio

NS = "/v2/realtime"
TOKEN = os.environ["PARKLIO_API_TOKEN"]
sio = socketio.Client()


def alert(subject, what, details=None):
    print("ALERT", subject.get("id", subject.get("uuid")), what, details or "")


def resolve(subject, what):
    print("RESOLVED", subject.get("id", subject.get("uuid")), what)


# alert() and resolve() stand for your alerting.
last = {}  # product id -> (online, error_code, low)


def is_low_battery(value):
    # Only barriers report a battery level. 0–100 is percent; above 100 means grid power.
    return isinstance(value, (int, float)) and value <= 20


@sio.on("lot_change", namespace=NS)
def on_lot_change(raw):
    event = json.loads(raw)
    if event.get("event_id"):
        return  # command results are handled elsewhere
    data = event["data"]
    online = data["status"] == "online"
    error_code = (data.get("product_error") or {}).get("error_code")
    low = is_low_battery(data.get("battery_status"))
    prev_online, prev_code, prev_low = last.get(data["id"], (None, None, False))
    last[data["id"]] = (online, error_code, low)

    if not online and prev_online:
        alert(data, "offline")
    if online and prev_online is False:
        resolve(data, "back online")
    if error_code and error_code != prev_code:
        alert(data, f"error flags {error_code}", data["product_error"])
    if not error_code and prev_code:
        resolve(data, "error cleared")
    if low and not prev_low:
        alert(data, f"battery at {data['battery_status']}%")


@sio.on("gateway_status", namespace=NS)
def on_gateway_status(raw):
    event = json.loads(raw)
    if event["status"] == "offline":
        alert({"uuid": event["uuid"]}, "Gateway offline")


if __name__ == "__main__":
    sio.connect(f"https://api.parklio.com?token={TOKEN}", namespaces=[NS], socketio_path="socketio")
    sio.wait()
```

## Create an operation token for your app

Mint a short-lived token for one product, to hand to your app. Explained in
[Operation tokens for your app](/guides/mobile-app-operation-tokens).

**Node.js**

```ts title="operation-token.ts"
// The operation-tokens guide: mint a short-lived token for your app.

export async function createOperationToken(productId: number, minutes: number) {
  const start = new Date();
  const end = new Date(start.getTime() + minutes * 60_000);
  const res = await fetch('https://api.parklio.com/v2/operation-tokens', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${process.env.PARKLIO_API_TOKEN}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      product_id: productId,
      start_time: start.toISOString(),
      end_time: end.toISOString(),
    }),
  });
  const body = await res.json();
  if (!body.success) throw new Error(`${body.errors[0].code}: ${body.errors[0].message}`);
  return body.data;
}

if (import.meta.main) {
  const token = await createOperationToken(Number(process.env.PRODUCT_ID), 15);
  // Hand these to your app, unchanged.
  const { token_open, token_close, token_status, end_time, product } = token;
  console.log({ device_id: product.device_id, end_time, token_open, token_close, token_status });
}
```

**Python**

```python title="create_token.py"
import os
from datetime import datetime, timedelta, timezone

import httpx


def create_operation_token(product_id, minutes):
    start = datetime.now(timezone.utc)
    end = start + timedelta(minutes=minutes)
    body = httpx.post(
        "https://api.parklio.com/v2/operation-tokens",
        headers={"Authorization": f"Bearer {os.environ['PARKLIO_API_TOKEN']}"},
        json={
            "product_id": product_id,
            "start_time": start.isoformat(),
            "end_time": end.isoformat(),
        },
    ).json()
    if not body["success"]:
        error = body["errors"][0]
        raise RuntimeError(f"{error['code']}: {error['message']}")
    return body["data"]
```

## Whitelist plates and bind them to a product

Add plates to a lot's whitelist, bind them to a product, and check the binding. Explained
in [Plate-based access](/guides/plate-based-access).

**Node.js**

```ts title="plate-access.ts"
// The plate-based-access guide: whitelist plates on a lot, bind them to a product.
const API = 'https://api.parklio.com/v2';
const headers = {
  Authorization: `Bearer ${process.env.PARKLIO_API_TOKEN}`,
  'Content-Type': 'application/json',
};

async function call(method: string, path: string, body?: unknown) {
  const init = body === undefined ? { method, headers } : { method, headers, body: JSON.stringify(body) };
  const res = await fetch(`${API}${path}`, init);
  const json = await res.json();
  if (!json.success) throw new Error(`${path}: ${JSON.stringify(json.errors)}`);
  return json;
}

export async function whitelistAndBind(lotId: number, productId: number, plates: string[], until: Date) {
  // 1. Whitelist the plates on the lot. One result per plate.
  const { data: results } = await call('POST', '/lot-whitelists', {
    lot_id: lotId,
    name: 'Tenants',
    license_plates: plates,
    start_time: new Date().toISOString(),
    end_time: until.toISOString(),
  });
  const failed = results.filter((r: { success: boolean }) => !r.success);
  if (failed.length) console.warn('Not whitelisted:', failed);
  const licenseIds = results.filter((r: { success: boolean }) => r.success).map((r: { license_id: number }) => r.license_id);

  // 2. Bind them to the product. Unknown ids are ignored silently...
  await call('POST', `/products/${productId}/bind-licenses`, { license_ids: licenseIds });

  // 3. ...so check what's bound.
  const { data: bound } = await call('GET', `/products/${productId}/bound-licenses?size=300`);
  const boundIds = new Set(bound.map((b: { id: number }) => b.id));
  return licenseIds.filter((id: number) => !boundIds.has(id)); // ids that didn't bind
}

if (import.meta.main) {
  const until = new Date(Date.now() + 30 * 24 * 3600_000);
  const missing = await whitelistAndBind(Number(process.env.LOT_ID), Number(process.env.PRODUCT_ID), ['ZG1234AB'], until);
  console.log(missing.length ? `Not bound: ${missing}` : 'Whitelisted and bound');
}
```

**Python**

```python title="plate_access.py"
"""The plate-based-access guide: whitelist plates on a lot, bind them to a product."""
import os
from datetime import datetime, timedelta, timezone

import httpx

API = "https://api.parklio.com/v2"
HEADERS = {"Authorization": f"Bearer {os.environ['PARKLIO_API_TOKEN']}"}


def call(method, path, body=None):
    res = httpx.request(method, f"{API}{path}", headers=HEADERS, json=body)
    data = res.json()
    if not data["success"]:
        raise RuntimeError(f"{path}: {data['errors']}")
    return data


def whitelist_and_bind(lot_id, product_id, plates, until):
    # 1. Whitelist the plates on the lot. One result per plate.
    results = call("POST", "/lot-whitelists", {
        "lot_id": lot_id,
        "name": "Tenants",
        "license_plates": plates,
        "start_time": datetime.now(timezone.utc).isoformat(),
        "end_time": until.isoformat(),
    })["data"]
    failed = [r for r in results if not r["success"]]
    if failed:
        print("Not whitelisted:", failed)
    license_ids = [r["license_id"] for r in results if r["success"]]

    # 2. Bind them to the product. Unknown ids are ignored silently...
    call("POST", f"/products/{product_id}/bind-licenses", {"license_ids": license_ids})

    # 3. ...so check what's bound.
    bound = call("GET", f"/products/{product_id}/bound-licenses?size=300")["data"]
    bound_ids = {b["id"] for b in bound}
    return [i for i in license_ids if i not in bound_ids]  # ids that didn't bind


if __name__ == "__main__":
    until = datetime.now(timezone.utc) + timedelta(days=30)
    missing = whitelist_and_bind(
        int(os.environ["LOT_ID"]), int(os.environ["PRODUCT_ID"]), ["ZG1234AB"], until
    )
    print(f"Not bound: {missing}" if missing else "Whitelisted and bound")
```

## Live occupancy

Keep free-space counts per zone up to date from `zone_change` events, and re-read the
zones every minute for changes that send no event. Explained in
[Live occupancy](/guides/live-occupancy).

**Node.js**

```ts title="live-occupancy.ts"
// The live-occupancy guide: read zone counts, keep them live, re-read regularly.
import { io } from 'socket.io-client';

const LOT_ID = Number(process.env.LOT_ID);
const headers = { Authorization: `Bearer ${process.env.PARKLIO_API_TOKEN}` };
const socket = io('https://api.parklio.com/v2/realtime', {
  path: '/socketio',
  query: { token: process.env.PARKLIO_API_TOKEN },
});

const zones = new Map<number, { spot_count: number; occupied_count: number }>();

socket.on('zone_change', (raw: string) => {
  for (const zone of JSON.parse(raw).zones) {
    zones.set(zone.id, { spot_count: zone.spot_count, occupied_count: zone.occupied_count });
    render(zone.id); // your UI
  }
});

function render(zoneId: number) {
  const zone = zones.get(zoneId);
  if (zone) console.log(`zone ${zoneId}: ${zone.spot_count - zone.occupied_count} free`);
}

// LPR, payment-machine and recount changes send no event: re-read every minute.
async function readZones(lotId: number) {
  const url = `https://api.parklio.com/v2/zones?filter=lot_id:eq:${lotId}&size=300`;
  const body = await (await fetch(url, { headers })).json();
  for (const zone of body.data) {
    zones.set(zone.id, { spot_count: zone.spot_count, occupied_count: zone.occupied_count });
    render(zone.id);
  }
}
socket.on('connect', () => readZones(LOT_ID).catch(console.error));
setInterval(() => readZones(LOT_ID).catch(console.error), 60_000);
```

**Python**

```python title="live_occupancy.py"
"""The live-occupancy guide: read zone counts, keep them live, re-read regularly."""
import json
import os
import threading

import httpx
import socketio

NS = "/v2/realtime"
TOKEN = os.environ["PARKLIO_API_TOKEN"]
LOT_ID = int(os.environ["LOT_ID"])

zones = {}  # zone id -> {"spot_count", "occupied_count"}
lock = threading.Lock()
sio = socketio.Client()


def render(zone_id):
    zone = zones[zone_id]
    print(f"zone {zone_id}: {zone['spot_count'] - zone['occupied_count']} free")


def remember(zone):
    with lock:
        zones[zone["id"]] = {
            "spot_count": zone["spot_count"],
            "occupied_count": zone["occupied_count"],
        }
        render(zone["id"])


@sio.on("zone_change", namespace=NS)
def on_zone_change(raw):
    for zone in json.loads(raw)["zones"]:
        remember(zone)


def read_zones(lot_id):
    # LPR, payment-machine and recount changes send no event: re-read regularly.
    res = httpx.get(
        f"https://api.parklio.com/v2/zones?filter=lot_id:eq:{lot_id}&size=300",
        headers={"Authorization": f"Bearer {TOKEN}"},
    )
    for zone in res.json()["data"]:
        remember(zone)


@sio.event(namespace=NS)
def connect():
    read_zones(LOT_ID)  # on the first connect and after every reconnect


if __name__ == "__main__":
    sio.connect(f"https://api.parklio.com?token={TOKEN}", namespaces=[NS], socketio_path="socketio")
    while True:
        sio.sleep(60)
        read_zones(LOT_ID)
```

## Next

- [Patterns](/realtime/patterns): the `PendingOperations` helper, deduplication and
  resync.
- [Troubleshooting](/resources/troubleshooting)