apsia-sdk

Official Python SDK for the Apsia API — the energy layer for satellite operations: battery intelligence from the telemetry you already receive, a per-orbit energy twin and energy-aware activity planning.

pip install apsia-sdk

Quickstart

from datetime import datetime, timedelta, timezone

from apsia_sdk import ApsiaClient

client = ApsiaClient()           # public energy endpoints need no key

# Not sure how a file will be read? Inspect it first — same detection analyze() applies, no
# recomputation. Correct only what auto-detection got wrong (a flipped sign, mA read as A, a
# relative time base with no epoch, local time) and send that back to analyze().
seen = client.energy.inspect(open("telemetry.csv").read())
print(seen["columns"])

# Battery health from a telemetry CSV (any column names; the server maps them)
soh = client.energy.analyze(open("telemetry.csv").read(), pack={"preset": "gomspace_nanopower_bp4"})
print(soh["report"]["headline"])
print(soh["soh_now"], soh["fade_per_year"], soh["rul_days"])       # p50 scalars; p10/p90 under soh["soh"]
print(soh["pack_learned"])                                        # capacity, resistance and SoC learned in flight

# Correcting what inspect() flagged: units, an epoch for a relative time base, a fixed sign
# convention, or a local-time offset — send only what you changed
soh = client.energy.analyze(
    open("telemetry.csv").read(),
    units={"i_batt": "mA"}, sign_convention="discharge_positive", tz_offset_h=2.0,
)
# A binary file (xlsx / Parquet / zip) instead of CSV text: send its raw bytes
soh = client.energy.analyze(file_bytes=open("telemetry.xlsx", "rb").read(), filename="telemetry.xlsx")

# Pass telemetry (what ground stations hear): give the NORAD id and every sample is
# placed on its orbit → end-of-eclipse trend with a verdict and the 7/30-day margin forecast
soh = client.energy.analyze(open("beacons.csv").read(), norad=41460)
print(soh["pass_indicator"]["verdict"], soh["pass_indicator"]["forecast"])

# The same numbers as a PDF, with the file, its hash, the date and the engine version on the page
pdf = client.energy.report_pdf(analysis=soh, filename="telemetry.csv", lang="en")
open("battery-report.pdf", "wb").write(pdf)

# Per-orbit energy twin for an orbit (or "tle": {line1, line2}) + power sheet
twin_body = {
    "orbit": {"alt_km": 550, "inc_deg": 97.6},
    "horizon_h": 24,
    "array": {"p_bol_w": 40},
    "pack": {"preset": "gomspace_nanopower_bp4"},
    "loads": {"class": "6U"},
}
# start from the satellite as observed, not from a full nominal pack: state of charge (propagated to
# t0), capacity and resistance learned from its telemetry; every response says what it used in
# state_used. A forecast must start at or after the observation and no more than 168 h (state's
# max_soc_age_h) after it, so t0 is derived from the analysis itself, not a date picked by hand.
state = client.energy.state_from_analysis(soh)
if state:
    twin_body["state"] = state
    twin_body["t0"] = state["observed_at"]
else:
    twin_body["t0"] = "2026-01-01T00:00:00Z"    # no usable state (e.g. a nominal pack): forecast from full

twin = client.energy.twin(twin_body)
print(twin["forecast"]["totals"]["min_soc"]["p50"], twin["report"]["headline"])
open("energy-forecast.pdf", "wb").write(client.energy.twin_report_pdf(twin, inputs=twin_body))

# Check an activity plan against that twin, then get a battery-aware reorder. Starts are relative to
# t0 so the plan always falls inside the forecast, whatever telemetry produced the state above.
t0 = twin_body["t0"]
t0_dt = datetime.fromtimestamp(t0, tz=timezone.utc) if isinstance(t0, (int, float)) else datetime.fromisoformat(str(t0).replace("Z", "+00:00"))
start_1 = (t0_dt + timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ")
plan_csv = f"activity_id,satellite_id,type,start_utc,duration_s,power_w,priority,value\nIMG-1,SAT-1,IMAGING,{start_1},600,60,1,800\n"
check = client.energy.plan_check({**twin_body, "plan_csv": plan_csv, "risk_policy": "pessimistic"})
print(check["assessment"]["verdict"], check["assessment"]["reasons"])   # feasible | at_risk | infeasible | incomplete
optimized = client.energy.plan_optimize({**twin_body, "plan_csv": plan_csv})
print(optimized["comparison"], optimized["plan_csv"])

# One task in two windows, with the same model as the check; the plan in your own command format
start_2 = (t0_dt + timedelta(hours=1, minutes=40)).strftime("%Y-%m-%dT%H:%M:%SZ")
cmp = client.energy.plan_compare({**twin_body, "plan_csv": plan_csv}, "IMG-1", [start_1, start_2])
print(cmp["recommended"], cmp["verdict"])
cmds = client.energy.plan_commands("{start_doy} EXEC {type} ID={activity_id} DUR={duration_s:d}", plan_rows=optimized["plan_rows"])
print(cmds["text"], cmds["unknown_placeholders"])

analyze, inspect, report_pdf, twin, twin_report_pdf, plan_check, plan_optimize, plan_compare, plan_commands, presets, sample and disposal_check/disposal_templates are public and rate-limited per IP (files up to 8 MB, 24 MB with a workspace key). Every figure comes with a range (p10/p50/p90) or its provenance; sample returns labelled synthetic telemetry for trying the flow.

Workspace

With an API key (X-API-Key, scope energy) the same client keeps a fleet server-side: save_satellite, satellites, satellite, put_satellite, delete_satellite, history, save_run, runs, run, fleet, alerts, ack_alert, rules/add_rule/delete_rule, export, and connectors (test_connector, set_connector, connector, delete_connector, pull, poll) that read a Yamcs, OpenC3 COSMOS or Grafana source on a schedule and raise alerts. grafana/ grafana_rows serve the health / forecast / alert tables to a Grafana JSON datasource.

client = ApsiaClient(api_key="kai_...")      # or APSIA_API_KEY in the environment
client.energy.save_satellite({"id": "SAT-1", "name": "SAT-1", "pack": {"preset": "gomspace_nanopower_bp4"}})
client.energy.set_connector("SAT-1", {
    "kind": "grafana", "base_url": "https://grafana.example",
    "datasource_uid": "influx", "query": {"query": "SELECT ..."},
    "fields": {"v_batt": "battery_voltage", "i_batt": "battery_current", "t_batt": "battery_temp"},
    "schedule_hours": 6,
})
print(client.energy.pull("SAT-1", hours=24)["headline"])
print(client.energy.fleet())

# A read-only link to one saved run (shown once), and a key for a colleague on the same workspace
link = client.energy.share_run(client.energy.runs(limit=1)["runs"][0]["run_id"], days=30)
colleague = client.energy.add_workspace_key("Flight dynamics")        # owner key only; the key is shown once

From your own server

The apsia command sends your latest export to a workspace satellite: it is read with the satellite's saved mapping, analysed, saved, and its alert rules run. One outbound HTTPS call; Apsia never connects to your network. One line per call, ready for a cron log:

export APSIA_API_KEY=kai_...
apsia ingest SAT-1 /data/eps_last24h.csv        # one line: satellite · headline · health and its range · alerts · run
apsia satellites                                # the workspace's satellites and their latest health
apsia alerts                                    # open alerts
apsia analyze telemetry.csv                     # one file, no key needed

client.energy.ingest("SAT-1", path="/data/eps_last24h.csv")    # the same from Python

The on-prem container serves the same endpoints inside your network; it comes with its keys.

Authentication and limits

• Public energy endpoints: no key, per-IP rate limit.
• Everything else: X-API-Key; the SDK reads APSIA_API_KEY automatically.
• Tiers academic · startup · operator · admin, each with a per-minute limit surfaced on every response (X-RateLimit-Limit, -Used, -Remaining, -Reset, X-API-Tier); client.last_rate_limit holds the parsed snapshot.

Request a key through the contact form at https://apsia.space/contact.

Error handling

from apsia_sdk import ApsiaAuthError, ApsiaError, ApsiaRateLimitError, ApsiaServerError

try:
    client.energy.analyze(csv_text)
except ApsiaAuthError:
    ...   # bad / missing / revoked key
except ApsiaRateLimitError as exc:
    # a short per-tier limit: exc.retry_after_seconds is a few seconds and the client already retried
    # up to max_retries before raising. A daily free-quota 429 (no key) is never retried by the client —
    # it raises immediately with retry_after_seconds set to the time until the quota resets
    time.sleep(exc.retry_after_seconds)
except ApsiaServerError:
    ...   # 5xx — Apsia temporarily unavailable
except ApsiaError as exc:
    ...   # everything else

License

Apache 2.0. The SDK is open source; the Apsia engine is proprietary and reachable only through the API.