Example clients

pip install websockets
export PELYR_KEY=pk_…
python stream.py

import asyncio, json, os, random, sys, websockets

URL = os.environ.get("PELYR_URL", "wss://stream.pelyr.com/v1/stream")
KEY = os.environ.get("PELYR_KEY")
if not KEY:
    sys.exit("PELYR_KEY missing")

ABO = {
    "type": "subscribe",
    "id": "ostsee",
    "bbox": [{"west": 18.0, "south": 59.0, "east": 26.0, "north": 66.0}],
    "fields": "position",
}

async def einmal() -> int:
    # Survives the welcome frame -- "position" below needs it for every
    # single frame, not just once at connect time.
    namensnennung: dict[str, str] = {}

    async with websockets.connect(
        URL,
        additional_headers={"Authorization": f"Bearer {KEY}"},
        ping_interval=None,
    ) as ws:
        async for roh in ws:
            frame = json.loads(roh)
            typ = frame.get("type")

            if typ == "welcome":
                # Each feed has exactly one id from 100 up; several feeds
                # share an id only when they deliberately run under one
                # collective attribution (id 1 = Pelyr). Look up by id, never
                # by licence text: every data frame carries its id in
                # "license", and whoever wants the attribution of the data
                # actually received remembers this mapping and looks it up
                # per frame (see "position" below).
                namensnennung = {q["id"]: q["attribution"] for q in frame["sources"]}
                for text in namensnennung.values():
                    print(text)
                await ws.send(json.dumps(ABO))

            elif typ == "position":
                d = frame["data"]
                # This frame's own attribution, looked up by license -- not
                # the list of every possible source from above.
                print(d["mmsi"], d["lat"], d["lon"], d.get("sog"),
                      namensnennung.get(frame["license"]))

            elif typ == "heartbeat" and frame["dropped"]:
                print("Datenverlust:", frame["dropped"], file=sys.stderr)

    return 1000

async def main() -> None:
    while True:
        try:
            code = await einmal()
        except websockets.ConnectionClosed as ende:
            code = ende.code

        # Endgueltig -- Wiederverbinden aendert nichts und erzeugt nur Last.
        if code in (4002, 4010, 4011):
            sys.exit(f"beendet, Code {code}")

        # 4012 heisst Redeploy: mit Jitter wiederkommen, nicht sofort.
        await asyncio.sleep(random.uniform(5, 30) if code == 4012 else 2.0)

asyncio.run(main())

go get github.com/coder/websocket
export PELYR_KEY=pk_…
go run stream.go

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "maps"
    "math/rand"
    "net/http"
    "os"
    "slices"
    "time"

    "github.com/coder/websocket"
)

type rahmen struct {
    Typ     string `json:"type"`
    Sources []struct {
        ID          string `json:"id"`
        License     string `json:"license"`
        Attribution string `json:"attribution"`
    } `json:"sources"`
    License string `json:"license"`
    Data    struct {
        MMSI int64    `json:"mmsi"`
        Lat  *float64 `json:"lat"`
        Lon  *float64 `json:"lon"`
    } `json:"data"`
    Dropped uint64 `json:"dropped"`
}

const abo = `{"type":"subscribe","id":"ostsee",` +
    `"bbox":[{"west":18,"south":59,"east":26,"north":66}],"fields":"position"}`

func einmal(ctx context.Context, key string) int {
    c, _, err := websocket.Dial(ctx, "wss://stream.pelyr.com/v1/stream",
        &websocket.DialOptions{
            HTTPHeader: http.Header{"Authorization": {"Bearer " + key}},
        })
    if err != nil {
        return 0
    }
    defer c.CloseNow()
    c.SetReadLimit(1 << 20)

    // Survives the welcome frame -- "position" below needs it for every
    // single frame, not just once at connect time.
    namensnennung := map[string]string{}

    for {
        _, roh, err := c.Read(ctx)
        if err != nil {
            return int(websocket.CloseStatus(err))
        }
        var r rahmen
        if json.Unmarshal(roh, &r) != nil {
            continue
        }

        switch r.Typ {
        case "welcome":
            // Each feed has exactly one id from 100 up; several feeds
            // share an id only when they deliberately run under one
            // collective attribution (id 1 = Pelyr). Look up by id, never
            // by licence text: every data frame carries its id in
            // "license", and whoever wants the attribution of the data
            // actually received remembers this mapping and looks it up
            // per frame (see "position" below).
            namensnennung = map[string]string{}
            for _, q := range r.Sources {
                namensnennung[q.ID] = q.Attribution
            }
            // Sorted by id -- map iteration order is randomized in Go.
            for _, id := range slices.Sorted(maps.Keys(namensnennung)) {
                fmt.Println(namensnennung[id])
            }
            if c.Write(ctx, websocket.MessageText, []byte(abo)) != nil {
                return 0
            }
        case "position":
            var lat, lon float64
            if r.Data.Lat != nil {
                lat = *r.Data.Lat
            }
            if r.Data.Lon != nil {
                lon = *r.Data.Lon
            }
            // This frame's own attribution, looked up by license -- not
            // the list of every possible source from above.
            fmt.Println(r.Data.MMSI, lat, lon, namensnennung[r.License])
        case "heartbeat":
            if r.Dropped > 0 {
                fmt.Fprintln(os.Stderr, "Datenverlust:", r.Dropped)
            }
        }
    }
}

func main() {
    key := os.Getenv("PELYR_KEY")
    ctx := context.Background()
    for {
        switch code := einmal(ctx, key); code {
        case 4002, 4010, 4011:
            fmt.Fprintf(os.Stderr, "beendet, Code %d\n", code)
            os.Exit(1)
        case 4012:
            time.Sleep(time.Duration(5+rand.Intn(25)) * time.Second)
        default:
            time.Sleep(2 * time.Second)
        }
    }
}