Alert Monitoring Pipeline

Overview

The Telemax alert system delivers driver and vehicle events (speeding, geofence, ignition, etc.) as AlertDto records. Your integration polls GetAlerts on an interval, processes new alerts, and acknowledges them to track which have been handled. Key fields per alert:

Field	Type	Description
`Id`	integer	Unique alert identifier — use for deduplication and acknowledgement
`UtcTime`	string	When the alert occurred (UTC, no `Z` suffix)
`AlertType`	integer	Numeric discriminator identifying the alert category
`Description`	string	Human-readable alert message
`VehicleId`	integer	Vehicle that triggered the alert
`VehicleName`	string	Vehicle display name
`Address`	string	Reverse-geocoded location
`Lat` / `Lng`	double	Location coordinates
`IsRead`	boolean	Whether the alert has been acknowledged

AlertType is an integer discriminator. The mapping of type integers to alert meanings (e.g. overspeed, geofence entry) is not documented in the current API. Contact your Telemax account manager for the type list relevant to your account (GAP-19).

Step 1 — Poll with a sliding window

GetAlerts takes a from datetime and a num limit. Use a sliding window: advance from to the timestamp of the latest alert received after each successful poll.

num is a legacy max-count parameter (recommended maximum: 100). There is no pagination cursor — if more than 100 alerts match your window, advance from and poll again.

import os, time, datetime, requests

API_BASE   = "https://api.telemax.com.au"
API_KEY    = os.environ["TELEMAX_API_KEY"]
COMPANY_ID = 85
POLL_SEC   = 60
MAX_ALERTS = 100

_token, _expires_at = None, 0

def get_token():
    global _token, _expires_at
    if _token and time.time() < _expires_at - 300:
        return _token
    r = requests.post(f"{API_BASE}/api/Authentication/token/api-key",
                      data={"apiKey": API_KEY})
    r.raise_for_status()
    d = r.json()
    _token, _expires_at = d["access_token"], time.time() + d["expires_in"]
    return _token

def fetch_alerts(from_dt: str) -> list:
    r = requests.post(
        f"{API_BASE}/api/GetAlerts",
        params={"compId": COMPANY_ID, "from": from_dt, "num": MAX_ALERTS},
        headers={"Authorization": f"Bearer {get_token()}"},
    )
    r.raise_for_status()
    return r.json()

seen_ids = set()
# Start from 24 hours ago
window_from = (datetime.datetime.utcnow() - datetime.timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%SZ")

while True:
    try:
        alerts = fetch_alerts(window_from)
        new_alerts = [a for a in alerts if a["Id"] not in seen_ids]

        for alert in new_alerts:
            seen_ids.add(alert["Id"])
            process_alert(alert)            # your handler — see Step 3

        if new_alerts:
            latest = max(new_alerts, key=lambda a: a["UtcTime"])
            window_from = latest["UtcTime"]

    except Exception as e:
        print(f"Poll error: {e}")

    time.sleep(POLL_SEC)

const API_BASE = "https://api.telemax.com.au";
const COMPANY_ID = 85;
let token = null, tokenExpiresAt = 0;
const seenIds = new Set();
let windowFrom = new Date(Date.now() - 86_400_000).toISOString();

async function getToken() {
  if (token && Date.now() < tokenExpiresAt - 300_000) return token;
  const res = await fetch(`${API_BASE}/api/Authentication/token/api-key`, {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({ apiKey: "YOUR_API_KEY" }),
  });
  const { access_token, expires_in } = await res.json();
  token = access_token;
  tokenExpiresAt = Date.now() + expires_in * 1000;
  return token;
}

async function pollAlerts() {
  const res = await fetch(
    `${API_BASE}/api/GetAlerts?compId=${COMPANY_ID}&from=${encodeURIComponent(windowFrom)}&num=100`,
    { method: "POST", headers: { Authorization: `Bearer ${await getToken()}` } },
  );
  const alerts = await res.json();
  const newAlerts = alerts.filter(a => !seenIds.has(a.Id));

  for (const alert of newAlerts) {
    seenIds.add(alert.Id);
    handleAlert(alert);  // your handler — see Step 3
  }

  if (newAlerts.length > 0) {
    const latest = newAlerts.reduce((a, b) => a.UtcTime > b.UtcTime ? a : b);
    windowFrom = latest.UtcTime;
  }
}

pollAlerts();
setInterval(pollAlerts, 60_000);

Step 2 — Filter by alert type

Use AlertType to route alerts to the appropriate handler. Until the full type mapping is documented, you can build a partial table from observed values in your own alert stream:

Python

ALERT_HANDLERS = {
    1:  "overspeed",
    5:  "geofence_enter",
    6:  "geofence_exit",
    # extend as you discover type values for your account
}

def process_alert(alert):
    alert_name = ALERT_HANDLERS.get(alert["AlertType"], f"type_{alert['AlertType']}")
    print(f"[{alert_name}] {alert['VehicleName']} — {alert['Description']} @ {alert['Address']}")
    dispatch_notification(alert_name, alert)

Step 3 — Acknowledge alerts

Mark one alert as read

SetAlertAsRead requires both alertId and date (the exact UtcTime string from the alert). The isRead parameter is present in the schema but is ignored by the backend — the call always marks the alert as read regardless of the value passed.

cURL

curl -s -X POST \
  "https://api.telemax.com.au/api/SetAlertAsRead?alertId=101&date=2025-05-01T10:30:00&isRead=true" \
  -H "Authorization: Bearer YOUR_TOKEN"

Python

def mark_read(alert):
    requests.post(
        f"{API_BASE}/api/SetAlertAsRead",
        params={
            "alertId": alert["Id"],
            "date":    alert["UtcTime"],
            "isRead":  "true",      # value ignored; pass true for clarity
        },
        headers={"Authorization": f"Bearer {get_token()}"},
    ).raise_for_status()

Bulk clear all alerts

When you have fully caught up and want to reset the unread state for the entire company:

cURL

curl -s -X POST \
  "https://api.telemax.com.au/api/SetAllAlertsAsRead?compId=85" \
  -H "Authorization: Bearer YOUR_TOKEN"

SetAllAlertsAsRead marks all alerts for the company as read — not just the ones your integration has processed. In multi-tenant or multi-integration setups, only call this when every consumer is caught up.

Building notification rules

A practical pattern for routing alerts to the right channel:

Python

import smtplib, json, requests as rq

SEVERITY = {
    "overspeed":     "high",
    "geofence_exit": "medium",
    "geofence_enter":"low",
}

CHANNELS = {
    "high":   lambda a: send_sms(a),
    "medium": lambda a: send_email(a),
    "low":    lambda a: log_to_dashboard(a),
}

def dispatch_notification(alert_name, alert):
    severity = SEVERITY.get(alert_name, "low")
    CHANNELS[severity](alert)

Gotchas

num is not a page cursor. There is no offset — to retrieve older alerts, move the from window backwards.
isRead is ignored. SetAlertAsRead always marks as read; do not branch on this parameter.
AlertType meanings are undocumented. Build your mapping empirically from your own account’s alert stream and verify with Telemax support (GAP-19).
UtcTime has no Z suffix but is always UTC — treat it as UTC when parsing.

Getting Started

Guides

Authentication

Companies & fleet

Telemetry & replay

Alerts

Insights & diagnostics

Vehicle commands

Operations & health

Reference

Alert Monitoring Pipeline

Overview

Step 1 — Poll with a sliding window

Step 2 — Filter by alert type

Step 3 — Acknowledge alerts

Mark one alert as read

Bulk clear all alerts

Building notification rules

Gotchas

​Overview

​Step 1 — Poll with a sliding window

​Step 2 — Filter by alert type

​Step 3 — Acknowledge alerts

​Mark one alert as read

​Bulk clear all alerts

​Building notification rules

​Gotchas

Overview

Step 1 — Poll with a sliding window

Step 2 — Filter by alert type

Step 3 — Acknowledge alerts

Mark one alert as read

Bulk clear all alerts

Building notification rules

Gotchas