Metadata-Version: 2.5
Name: lokate
Version: 2.0.0
Summary: The python client for lokate, the arkitekt backup of a phone's location timeline
Author-email: jhnnsrs <jhnnsrs@gmail.com>
Requires-Python: >=3.11
Requires-Dist: aiohttp<4,>=3.9.5
Requires-Dist: koil>=1.0.0
Requires-Dist: rath>=4.1.1
Requires-Dist: websockets>=15.0.1
Description-Content-Type: text/markdown

# lokate

The python client for [lokate](https://github.com/arkitektio/lokate-server), the
[Arkitekt](https://arkitekt.live) backup of a phone's location timeline: points,
visits, trips and places, uploaded idempotently and restorable page by page.

## Installation

```sh
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`:

```python
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

```python
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:

```python
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:

```python
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:

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

```sh
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](RELEASING.md) for how versions are cut.
