Skip to content

Latest commit

 

History

1 Commit

Folders and files

NameName
Last commit message
Last commit date
 
 
 
 
 
 

Repository files navigation

Levelmaster Python SDK

A Python client library for the Levelmaster tank monitoring API. Provides clean, Pythonic access to all API resources with automatic authentication handling and pandas DataFrames for tabular data.


Table of Contents


Installation

Install dependencies directly (no package index release yet):

pip install requests pandas

Then clone or copy the levelmaster/ directory into your project, or install from the source directory:

pip install .

Requirements: Python 3.10+, requests >= 2.28, pandas >= 1.5


Quick Start

from levelmaster import LevelmasterClient

# Authenticates immediately on construction
client = LevelmasterClient("you@example.com", "yourpassword")

# Get a live snapshot of every tank
df = client.get_latest_levels()
print(df[["name", "content", "volume", "capacity", "fill_pct"]])
              name content   volume  capacity  fill_pct
0  Baruk Petroleum T1  ULP 95    891.0   79179.0      1.13

Authentication

The Levelmaster API uses the devise_token_auth pattern. On sign-in the server returns three tokens in the response headers (not body):

Header Description
access-token Short-lived bearer token, rotated on every request
client Persistent session identifier
uid Account email address
expiry Unix timestamp when the session expires

The SDK handles all of this transparently:

  • Automatic sign-in on LevelmasterClient() construction.
  • Rolling token refresh — every successful API response carries a new access-token; the SDK updates its session headers automatically.
  • Proactive re-authentication — if the session is within 60 seconds of its expiry time, the SDK re-authenticates before the next request.

You never need to manage tokens manually.


Client Reference

Constructor

LevelmasterClient(
    email: str,
    password: str,
    *,
    base_url: str = "http://www.levelmaster.co.za/api",
    timeout: int = 30,
)

Authenticates immediately. Raises AuthenticationError if credentials are wrong or the server is unreachable.

Parameter Type Default Description
email str — Account email address
password str — Account password
base_url str http://www.levelmaster.co.za/api Override the API base URL (useful for testing)
timeout int 30 HTTP request timeout in seconds

The client also supports use as a context manager, which closes the underlying HTTP session cleanly:

with LevelmasterClient("you@example.com", "password") as client:
    df = client.get_containers()
# session closed automatically here

get_clients()

client.get_clients() -> pd.DataFrame

Returns all clients (top-level account groupings) visible to the authenticated user.

Returns — DataFrame columns:

Column Type Description
type str Always "client"
client_uuid str Unique identifier for the client
name str Client display name

Example:

df = client.get_clients()
print(df)
#      type                             client_uuid    name
# 0  client  afd1e04b-83e1-4901-ba0e-f179a980a827  BB Soft

get_sites()

client.get_sites() -> pd.DataFrame

Returns all sites visible to the authenticated user across all clients.

Returns — DataFrame columns:

Column Type Description
type str Always "site"
site_uuid str Unique identifier for the site
client_uuid str UUID of the parent client
name str Site display name

Example:

df = client.get_sites()
print(df[["site_uuid", "name"]])
#                               site_uuid             name
# 0  3c74d4ef-0797-48a4-a437-7202b874f210  Baruk Petroleum

get_containers()

client.get_containers() -> pd.DataFrame

Returns all containers (tanks/vessels) visible to the authenticated user, with their latest sensor reading included. This is the primary endpoint for a live status dashboard.

Returns — DataFrame columns:

Column Type Description
type str Always "container"
container_uuid str Unique identifier for the container
site_uuid str UUID of the parent site
client_uuid str UUID of the parent client
name str Container display name
content str Product stored (e.g. "ULP 95", "Diesel")
capacity float Maximum capacity in litres
volume float Current volume in litres
distance float Probe distance reading in mm
captured_at datetime64[ns, UTC] Timestamp of the latest reading
water_bot bool Water detected at the bottom of the tank
water_top bool Water detected at the top of the tank
probe_attached bool Whether a probe is physically connected
calibration_fault bool Whether a calibration fault is flagged

captured_at is automatically parsed to a timezone-aware UTC datetime.

Example:

df = client.get_containers()
print(df[["name", "content", "volume", "capacity", "calibration_fault"]])
#                  name content   volume  capacity  calibration_fault
# 0  Baruk Petroleum T1  ULP 95    891.0   79179.0               True

get_client_details()

client.get_client_details(client_uuid: str) -> dict

Returns full details for a single client, including a nested list of all its sites.

Parameter Type Description
client_uuid str UUID of the client to fetch

Returns — dict keys:

Key Type Description
type str "client"
client_uuid str UUID of the client
name str Client display name
sites list[dict] List of site dicts — each has type, site_uuid, name

Example:

details = client.get_client_details("afd1e04b-83e1-4901-ba0e-f179a980a827")
print(details["name"])        # BB Soft
print(len(details["sites"]))  # 6

get_client_sites()

client.get_client_sites(client_uuid: str) -> pd.DataFrame

Convenience wrapper around get_client_details() that returns only the nested sites list as a DataFrame.

Parameter Type Description
client_uuid str UUID of the client

Returns — DataFrame columns: type, site_uuid, name

Example:

df = client.get_client_sites("afd1e04b-83e1-4901-ba0e-f179a980a827")
print(df[["site_uuid", "name"]])
#                               site_uuid                   name
# 0  2a1dabf8-37c0-4c31-9ae0-5603c36f914c          Golden Citrus
# 1  5afe7cb6-fcf4-4304-988c-e0de1198fdc2  Mass Petroleum Sasolburg
# ...

get_site_details()

client.get_site_details(site_uuid: str) -> dict

Returns full details for a single site, including a nested list of its containers and their latest readings.

Parameter Type Description
site_uuid str UUID of the site to fetch

Returns — dict keys:

Key Type Description
type str "site"
site_uuid str UUID of the site
client_uuid str UUID of the parent client
site_name str Site display name
containers list[dict] List of container dicts — each has type, container_uuid, name, content, capacity, volume, captured_at

Example:

details = client.get_site_details("3c74d4ef-0797-48a4-a437-7202b874f210")
print(details["site_name"])         # Baruk Petroleum
print(details["containers"][0])     # {'type': 'container', ...}

get_site_containers()

client.get_site_containers(site_uuid: str) -> pd.DataFrame

Convenience wrapper around get_site_details() that returns only the nested containers list as a DataFrame. captured_at is parsed to UTC datetime.

Parameter Type Description
site_uuid str UUID of the site

Returns — DataFrame columns: type, container_uuid, name, content, capacity, volume, captured_at

Example:

df = client.get_site_containers("3c74d4ef-0797-48a4-a437-7202b874f210")
print(df[["name", "volume", "capacity"]])

get_container_details()

client.get_container_details(container_uuid: str) -> dict

Returns the raw detail response for a single container, including the full historical readings array.

Parameter Type Description
container_uuid str UUID of the container

Returns — dict keys:

Key Type Description
container_uuid str UUID of the container
readings list[dict] List of readings — each has captured_at (ISO 8601 string) and volume (float)

Use get_readings() instead if you want a ready-to-use DataFrame.


get_readings()

client.get_readings(container_uuid: str) -> pd.DataFrame

Returns the full time-series of volume readings for a single container as a sorted DataFrame.

Parameter Type Description
container_uuid str UUID of the container

Returns — DataFrame columns:

Column Type Description
captured_at datetime64[ns, UTC] Timestamp of the reading (UTC, sorted ascending)
volume float Volume in litres at that timestamp

Example:

df = client.get_readings("127e1743-bf04-4256-97ec-fd9f58526def")
print(df.head())
#                   captured_at   volume
# 0 2026-03-24 06:42:46+00:00    904.0
# 1 2026-03-24 06:47:43+00:00    904.0
# 2 2026-03-24 06:52:39+00:00    904.0

# Standard pandas time-series operations work directly
daily_avg = df.set_index("captured_at").resample("D")["volume"].mean()

get_latest_levels()

client.get_latest_levels() -> pd.DataFrame

Returns the most recent level snapshot for every container visible to the account, with a computed fill_pct column. Uses only the /containers list endpoint — no per-container round-trips.

Returns — DataFrame columns:

Column Type Description
container_uuid str Unique identifier
site_uuid str Parent site UUID
client_uuid str Parent client UUID
name str Container name
content str Product stored
capacity float Maximum capacity (litres)
volume float Current volume (litres)
fill_pct float volume / capacity × 100, rounded to 2 decimal places
distance float Probe distance reading (mm)
captured_at datetime64[ns, UTC] Timestamp of reading
water_bot bool Water at bottom
water_top bool Water at top
probe_attached bool Probe connected
calibration_fault bool Calibration fault present

Example:

df = client.get_latest_levels()

# Find tanks below 20% fill
low = df[df["fill_pct"] < 20]
print(low[["name", "content", "fill_pct"]])

get_all_readings()

client.get_all_readings(container_uuids: list[str] | None = None) -> pd.DataFrame

Fetches and concatenates historical readings for multiple containers into one DataFrame. If container_uuids is omitted, all containers visible to the account are queried.

Note: This makes one HTTP request per container. For accounts with many containers, consider passing a specific subset of UUIDs.

Parameter Type Description
container_uuids list[str], optional UUIDs to fetch. Defaults to all containers on the account.

Returns — DataFrame columns:

Column Type Description
container_uuid str Which container the reading belongs to
captured_at datetime64[ns, UTC] Timestamp (UTC)
volume float Volume in litres

Sorted by container_uuid then captured_at.

Example:

# Fetch all readings for two specific tanks
uuids = [
    "127e1743-bf04-4256-97ec-fd9f58526def",
    "some-other-container-uuid",
]
df = client.get_all_readings(uuids)

# Pivot to compare tanks side by side
pivot = df.pivot(index="captured_at", columns="container_uuid", values="volume")

Data Schemas

Clients

client_uuid  (str)   — "afd1e04b-83e1-4901-ba0e-f179a980a827"
name         (str)   — "BB Soft"

Sites

site_uuid    (str)   — "3c74d4ef-0797-48a4-a437-7202b874f210"
client_uuid  (str)   — parent client UUID
name         (str)   — "Baruk Petroleum"

Containers (tanks)

container_uuid     (str)            — "127e1743-bf04-4256-97ec-fd9f58526def"
site_uuid          (str)            — parent site UUID
client_uuid        (str)            — parent client UUID
name               (str)            — "Baruk Petroleum T1"
content            (str)            — "ULP 95"
capacity           (float)          — 79179.0  (litres)
volume             (float)          — 891.0    (litres)
distance           (float)          — 101.0    (mm, probe distance)
captured_at        (datetime, UTC)  — last reading timestamp
water_bot          (bool)           — water detected at bottom
water_top          (bool)           — water detected at top
probe_attached     (bool)           — whether probe is connected
calibration_fault  (bool)           — calibration fault flag

Readings

captured_at  (datetime, UTC)  — timestamp of the reading
volume       (float)          — volume in litres at that time

Error Handling

All SDK exceptions inherit from LevelmasterError:

LevelmasterError
├── AuthenticationError   — wrong credentials, expired session, network error during login
├── NotFoundError         — the requested UUID does not exist (HTTP 404)
└── APIError              — any other unexpected HTTP error
    ├── .status_code      — int HTTP status code
    └── .response         — the raw requests.Response object

Example:

from levelmaster import LevelmasterClient, AuthenticationError, NotFoundError, APIError

try:
    client = LevelmasterClient("you@example.com", "wrongpassword")
except AuthenticationError as e:
    print(f"Login failed: {e}")

try:
    df = client.get_readings("nonexistent-uuid")
except NotFoundError:
    print("That container UUID does not exist.")
except APIError as e:
    print(f"API error {e.status_code}: {e}")

Recipes & Examples

Live dashboard snapshot

from levelmaster import LevelmasterClient

client = LevelmasterClient("you@example.com", "password")

df = client.get_latest_levels()
print(df[["name", "content", "volume", "capacity", "fill_pct", "captured_at"]])

Find tanks with alerts

df = client.get_latest_levels()

alerts = df[
    (df["fill_pct"] < 15) |        # critically low
    df["water_bot"] |               # water contamination
    df["calibration_fault"]         # sensor issue
]
print(alerts[["name", "fill_pct", "water_bot", "calibration_fault"]])

Plot a tank's level history

import matplotlib.pyplot as plt

df = client.get_readings("127e1743-bf04-4256-97ec-fd9f58526def")
df = df.set_index("captured_at")

df["volume"].plot(title="Baruk Petroleum T1 — Volume over Time", ylabel="Litres")
plt.tight_layout()
plt.show()

Daily min/max volume per container

df = client.get_all_readings()
df = df.set_index("captured_at")

daily = (
    df.groupby("container_uuid")["volume"]
    .resample("D")
    .agg(["min", "max", "mean"])
    .reset_index()
)
print(daily)

Detect deliveries (large volume increases)

df = client.get_readings("127e1743-bf04-4256-97ec-fd9f58526def")
df["delta"] = df["volume"].diff()

deliveries = df[df["delta"] > 5000]   # adjust threshold as needed
print(deliveries[["captured_at", "volume", "delta"]])

Export all container data to Excel

import pandas as pd

df_containers = client.get_containers()
df_readings = client.get_all_readings()

with pd.ExcelWriter("levelmaster_export.xlsx") as writer:
    df_containers.to_excel(writer, sheet_name="Containers", index=False)
    df_readings.to_excel(writer, sheet_name="Readings", index=False)

Iterate all sites under a client

clients_df = client.get_clients()

for _, row in clients_df.iterrows():
    sites_df = client.get_client_sites(row["client_uuid"])
    print(f"\n{row['name']} ({len(sites_df)} sites):")
    for _, site in sites_df.iterrows():
        containers_df = client.get_site_containers(site["site_uuid"])
        print(f"  {site['name']}: {len(containers_df)} containers")

API Overview

The Levelmaster API is a JSON REST API at http://www.levelmaster.co.za/api. All authenticated endpoints require three headers obtained from the sign-in response:

access-token: <token>
client:       <client-id>
uid:          <email>
Method Endpoint Description
POST /auth/sign_in Authenticate. Returns tokens in response headers. Body: email, password (form-encoded).
GET /clients List all clients
GET /client/{client_uuid} Client detail + nested sites
GET /sites List all sites
GET /site/{site_uuid} Site detail + nested containers
GET /containers List all containers with latest readings
GET /container/{container_uuid} Container detail with full reading history

All timestamps in API responses use ISO 8601 format with timezone offset (e.g. 2026-03-27T08:03:48.000+02:00). The SDK converts these to UTC datetime64 automatically.

About

No description, website, or topics provided.

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages