lokate

The python client for lokate, the Arkitekt backup of a phone's location timeline: points, visits, trips and places, uploaded idempotently and restorable page by page.

Installation

pip install lokate

Usage

Every lokate operation is a method of the Lokate client, in a blocking and an a-prefixed async flavour. The server keys everything by the token's user and writes by its device (client_device claim).

In an arkitekt app

The client is injected by annotation. Add the service to your app and ask for lokate: Lokate:

from arkitekt import App, run
from lokate import Lokate, lokate_service

app = App("where-was-i", "0.1.0", services=[lokate_service])


@app.action
async def places_i_keep(lokate: Lokate) -> list[str]:
    """The names of my places, from the server copy."""
    names, cursor = [], None
    while True:
        page = await lokate.aget_changes(cursor=cursor)
        names += [place.name for place in page.places]
        cursor = page.next_cursor
        if not page.has_more:
            return names


if __name__ == "__main__":
    run(app)

From a script

from arkitekt import easy
from lokate import lokate_service

with easy("my-script", lokate_service) as lokate:
    state = lokate.get_sync_state()
    print(state.point_count, state.last_point_ts)

Standalone

Without arkitekt, build the client over a rath link of your own:

import datetime
from lokate import Lokate
from lokate.api.schema import PointInput
from lokate.rath import LokateRath

lokate = Lokate(rath=LokateRath(link=...))

with lokate:
    result = lokate.upload_points([
        PointInput(client_id="fix-1", ts=datetime.datetime.now(datetime.UTC), lat=48.2, lon=16.37),
    ])
    print(result.accepted, result.duplicates)

Reading the timeline

Everything is readable back, and only ever your own rows, from all of your devices:

import datetime
from lokate.api.schema import Granularity, PointFilter

day = lokate.get_day(date=datetime.date(2026, 9, 30), timezone="Europe/Vienna")
for visit in day.visits:
    print(visit.start, visit.place.name if visit.place else "?", visit.duration)

since = datetime.datetime(2026, 9, 1, tzinfo=datetime.UTC)
until = datetime.datetime(2026, 10, 1, tzinfo=datetime.UTC)
tracks = lokate.get_route(since=since, until=until, simplify=10)   # GeoJSON per device
stats = lokate.get_stats(since=since, until=until, granularity=Granularity.WEEK)
points = lokate.list_points(filters=PointFilter(since=since, near={"lat": 48.2, "lon": 16.37, "radius": 500}))

list_devices, list_points, list_visits, list_trips, list_places take filters, pagination and ordering; count_* and get_*(id) go with them. get_place_stats gives the time spent per place.

The backup protocol

• Every write can be retried. upload_points counts points it already has as duplicates. replace_segments(from_=...) and merge_places leave unchanged rows alone.
• A point is keyed by (device, client_id, ts). Never send one client_id with two different timestamps.
• Batches hold at most 1000 rows, and so does a get_changes page.
• Leave defaulted arguments out rather than passing None. For set_retention, though, days=None is meaningful: it means keep forever.

Development

The generated API (lokate/api/schema.py) comes from the checked-in schema.graphql and the documents in graphql/. The config comment in graphql.config.yaml has the command that refreshes the schema from a lokate-server checkout. Then regenerate with turms:

uvx --with 'graphql-core<3.3' --with-editable . --from <path to turms> turms gen

uv run pytest -m "not integration"   # no server needed
uv run pytest -m integration          # a real lokate + postgres/PostGIS via dokker

The integration suite runs jhnnsrs/lokate:${LOKATE_SERVICE_TAG:-latest}. To test unreleased server changes, add a gitignored tests/integration/docker-compose.local.yml that builds the image from a local lokate-server checkout.

See RELEASING.md for how versions are cut.