From debf4c11bfe5c0b866baefd04c3528fec5a304ca Mon Sep 17 00:00:00 2001 From: Nguyen Trong Thanh Date: Sun, 31 May 2026 23:58:19 +0700 Subject: [PATCH] add Go aiscat binding --- go/aiscat/README.md | 78 +++++++ go/aiscat/aiscat_sources.cpp | 14 ++ go/aiscat/bridge.cpp | 329 ++++++++++++++++++++++++++++ go/aiscat/bridge.h | 37 ++++ go/aiscat/decode.go | 123 +++++++++++ go/aiscat/decoder_test.go | 101 +++++++++ go/aiscat/examples/decode/main.go | 19 ++ go/aiscat/examples/from_tcp/main.go | 36 +++ go/aiscat/format.go | 80 +++++++ go/aiscat/go.mod | 3 + go/aiscat/native.go | 195 +++++++++++++++++ go/aiscat/nocgo.go | 57 +++++ go/aiscat/stream.go | 166 ++++++++++++++ 13 files changed, 1238 insertions(+) create mode 100644 go/aiscat/README.md create mode 100644 go/aiscat/aiscat_sources.cpp create mode 100644 go/aiscat/bridge.cpp create mode 100644 go/aiscat/bridge.h create mode 100644 go/aiscat/decode.go create mode 100644 go/aiscat/decoder_test.go create mode 100644 go/aiscat/examples/decode/main.go create mode 100644 go/aiscat/examples/from_tcp/main.go create mode 100644 go/aiscat/format.go create mode 100644 go/aiscat/go.mod create mode 100644 go/aiscat/native.go create mode 100644 go/aiscat/nocgo.go create mode 100644 go/aiscat/stream.go diff --git a/go/aiscat/README.md b/go/aiscat/README.md new file mode 100644 index 0000000000..3a173c9618 --- /dev/null +++ b/go/aiscat/README.md @@ -0,0 +1,78 @@ +# aiscat-go + +Go bindings for AIS-catcher's NMEA-to-JSON AIS decoder. The package mirrors the +Python `aiscat` binding: a thin native bridge owns the AIS-catcher decoder, while +Go provides ergonomic one-shot, reader, TCP, and UDP helpers. + +## Build Requirements + +- Go with cgo enabled. +- A C++11 compiler available to cgo. + - Linux/macOS: `g++` or `clang++`. + - Windows: MSYS2 MinGW-w64 or another cgo-compatible C/C++ toolchain. + - This checkout was verified on Windows with Zig: + `CGO_ENABLED=1 CC="zig cc" CXX="zig c++" go test ./...`. + +This package links AIS-catcher C++ code and is GPL-3.0-or-later. + +## Quickstart + +```go +package main + +import ( + "fmt" + "log" + + "github.com/jvde-github/ais-catcher/go/aiscat" +) + +func main() { + msg, err := aiscat.DecodeMap( + aiscat.Options{Format: aiscat.FormatDictionary}, + []byte("!AIVDM,1,1,,A,15MgK45P3@G?fl0E`JbR0OwT0@MS,0*4E"), + ) + if err != nil { + log.Fatal(err) + } + fmt.Println(msg["type"], msg["mmsi"], msg["lat"], msg["lon"]) +} +``` + +## Streaming + +```go +dec, err := aiscat.NewDecoder(aiscat.Options{Format: aiscat.FormatDictionary}) +if err != nil { + log.Fatal(err) +} +defer dec.Close() + +dec.Feed([]byte("!AIVDM,...\r\n")) +for { + msg, ok, err := dec.NextMap() + if err != nil { + log.Fatal(err) + } + if !ok { + break + } + fmt.Println(msg["mmsi"]) +} +``` + +## Formats + +The same format names as Python `aiscat` are available: + +- `FormatDictionary` +- `FormatAnnotated` +- `FormatJSON` +- `FormatJSONNMEA` +- `FormatNMEA` +- `FormatNMEATag` +- `FormatBinary` + +`Decoder.Next()` returns `map[string]any` for dictionary/annotated formats and +`[]byte` for byte-shaped formats. Use `NextMap()` or `NextBytes()` when you want +an explicit shape. diff --git a/go/aiscat/aiscat_sources.cpp b/go/aiscat/aiscat_sources.cpp new file mode 100644 index 0000000000..b97dcd6393 --- /dev/null +++ b/go/aiscat/aiscat_sources.cpp @@ -0,0 +1,14 @@ +// Compile the same AIS-catcher source subset used by the Python aiscat module. +// SPDX-License-Identifier: GPL-3.0-or-later + +#include "../../Source/Marine/AIS.cpp" +#include "../../Source/Marine/Message.cpp" +#include "../../Source/Marine/NMEA.cpp" +#include "../../Source/JSON/JSON.cpp" +#include "../../Source/JSON/JSONAIS.cpp" +#include "../../Source/JSON/Keys.cpp" +#include "../../Source/JSON/Parser.cpp" +#include "../../Source/Library/Logger.cpp" +#include "../../Source/Utilities/Convert.cpp" +#include "../../Source/Utilities/Helper.cpp" +#include "../../Source/Utilities/Parse.cpp" diff --git a/go/aiscat/bridge.cpp b/go/aiscat/bridge.cpp new file mode 100644 index 0000000000..5f54e1ae0e --- /dev/null +++ b/go/aiscat/bridge.cpp @@ -0,0 +1,329 @@ +// aiscat-go C ABI bridge for AIS-catcher's C++ decoder. +// SPDX-License-Identifier: GPL-3.0-or-later + +#include "bridge.h" + +#include +#include +#include +#include +#include +#include + +#include "Common.h" +#include "Stream.h" +#include "Marine/NMEA.h" +#include "Marine/Message.h" +#include "JSON/JSONAIS.h" +#include "JSON/JSON.h" +#include "JSON/Keys.h" +#include "JSON/Writer.h" + +namespace { + +enum class OutFormat { + DICTIONARY = AISCAT_FORMAT_DICTIONARY, + ANNOTATED = AISCAT_FORMAT_ANNOTATED, + JSON = AISCAT_FORMAT_JSON, + JSON_NMEA = AISCAT_FORMAT_JSON_NMEA, + NMEA = AISCAT_FORMAT_NMEA, + NMEA_TAG = AISCAT_FORMAT_NMEA_TAG, + BINARY = AISCAT_FORMAT_BINARY, +}; + +static constexpr uint64_t kSkipMask = + (1ULL << AIS::KEY_CLASS) | + (1ULL << AIS::KEY_DEVICE) | + (1ULL << AIS::KEY_DRIVER) | + (1ULL << AIS::KEY_HARDWARE) | + (1ULL << AIS::KEY_SCALED) | + (1ULL << AIS::KEY_VERSION) | + (1ULL << AIS::KEY_RXTIME) | + (1ULL << AIS::KEY_NMEA); + +bool valid_format(int format) { + return format >= AISCAT_FORMAT_DICTIONARY && format <= AISCAT_FORMAT_BINARY; +} + +void write_filtered_value(const JSON::Value &v, JSON::Writer &w, bool annotated); + +void write_annotated_scalar(const JSON::Value &v, int key, JSON::Writer &w) { + w.beginObject().key("value"); + write_filtered_value(v, w, true); + + if (key >= 0 && key < (int)AIS::KEY_COUNT) { + const AIS::KeyInfo &info = AIS::KeyInfoMap[key]; + if (info.unit && info.unit[0]) { + w.kv("unit", info.unit); + } + if (info.description && info.description[0]) { + w.kv("description", info.description); + } + if (info.lookup_table && (v.isInt() || v.isFloat())) { + long n = v.isInt() ? v.getInt() : (long)v.getFloat(); + if (n >= 0 && n < (long)info.lookup_table->size()) { + w.kv("text", (*info.lookup_table)[(size_t)n]); + } + } + } + w.endObject(); +} + +void write_filtered_object(const JSON::JSON &obj, JSON::Writer &w, bool annotated) { + w.beginObject(); + for (const JSON::Member &member : obj.getMembers()) { + int key_index = member.Key(); + if ((unsigned)key_index < 64 && (kSkipMask & (1ULL << key_index))) { + continue; + } + if (key_index < 0 || key_index >= (int)AIS::KEY_COUNT) { + continue; + } + const AIS::KeyStr &key = AIS::KeyMap[key_index][JSON_DICT_FULL]; + if (key.empty()) { + continue; + } + + w.key(key); + const JSON::Value &value = member.Get(); + using T = JSON::Value::Type; + T type = value.getType(); + if (annotated && (type == T::BOOL || type == T::INT || type == T::FLOAT || type == T::STRING)) { + write_annotated_scalar(value, key_index, w); + } else { + write_filtered_value(value, w, annotated); + } + } + w.endObject(); +} + +void write_filtered_value(const JSON::Value &v, JSON::Writer &w, bool annotated) { + switch (v.getType()) { + case JSON::Value::Type::BOOL: + w.val(v.getBool()); + break; + case JSON::Value::Type::INT: + w.val((long long)v.getInt()); + break; + case JSON::Value::Type::FLOAT: + w.val(v.getFloat()); + break; + case JSON::Value::Type::STRING: + w.val(v.getStringRef()); + break; + case JSON::Value::Type::OBJECT: + write_filtered_object(v.getObject(), w, annotated); + break; + case JSON::Value::Type::ARRAY_STRING: { + w.beginArray(); + for (const std::string &s : v.getStringArray()) { + w.val(s); + } + w.endArray(); + break; + } + case JSON::Value::Type::ARRAY: { + w.beginArray(); + for (const JSON::Value &item : v.getArray()) { + write_filtered_value(item, w, annotated); + } + w.endArray(); + break; + } + case JSON::Value::Type::EMPTY: + default: + w.val_null(); + break; + } +} + +std::string stringify_filtered(const JSON::JSON &obj, bool annotated) { + std::string out; + JSON::Writer writer(out); + write_filtered_object(obj, writer, annotated); + writer.finish(); + return out; +} + +class GoSink : public StreamIn { +public: + std::deque queue; + OutFormat format = OutFormat::DICTIONARY; + JSON::Serializer serializer; + std::string scratch; + + void Receive(const JSON::JSON *data, int len, TAG &tag) override { + for (int i = 0; i < len; ++i) { + scratch.clear(); + switch (format) { + case OutFormat::DICTIONARY: + queue.push_back(stringify_filtered(data[i], false)); + break; + case OutFormat::ANNOTATED: + queue.push_back(stringify_filtered(data[i], true)); + break; + case OutFormat::JSON: + serializer.stringify(data[i], scratch); + queue.push_back(scratch); + break; + case OutFormat::JSON_NMEA: { + const AIS::Message *msg = static_cast(data[i].binary); + if (!msg) break; + msg->getNMEAJSON(scratch, tag); + queue.push_back(scratch); + break; + } + case OutFormat::NMEA: { + const AIS::Message *msg = static_cast(data[i].binary); + if (!msg) break; + auto sentences = msg->sentences(); + for (size_t j = 0; j < sentences.size(); ++j) { + scratch.append(sentences[j]); + scratch.push_back('\n'); + } + queue.push_back(scratch); + break; + } + case OutFormat::NMEA_TAG: { + const AIS::Message *msg = static_cast(data[i].binary); + if (!msg) break; + msg->getNMEATagBlock(scratch); + queue.push_back(scratch); + break; + } + case OutFormat::BINARY: { + const AIS::Message *msg = static_cast(data[i].binary); + if (!msg) break; + msg->getBinaryNMEA(scratch, tag); + queue.push_back(scratch); + break; + } + } + } + } +}; + +} // namespace + +struct AisCatDecoder { + AIS::NMEA nmea; + AIS::JSONAIS jsonais; + GoSink sink; + TAG tag; + std::string last_error; + + AisCatDecoder(OutFormat fmt, bool country) { + sink.format = fmt; + tag.clear(); + if (country) { + tag.mode |= 4; + } + nmea.out.Connect(&jsonais); + jsonais.out.Connect(&sink); + } + + int set_error(const char *msg) { + last_error = msg ? msg : "unknown error"; + return -1; + } + + int set_error(const std::exception &e) { + last_error = e.what(); + return -1; + } +}; + +extern "C" { + +AisCatDecoder *aiscat_new(int format, int country) { + if (!valid_format(format)) { + return nullptr; + } + + try { + return new AisCatDecoder((OutFormat)format, country != 0); + } catch (...) { + return nullptr; + } +} + +void aiscat_free(AisCatDecoder *decoder) { + delete decoder; +} + +int aiscat_feed(AisCatDecoder *decoder, const char *data, size_t len) { + if (!decoder) { + return -1; + } + if (len > 0 && !data) { + return decoder->set_error("feed data is null"); + } + if (len > (size_t)INT32_MAX) { + return decoder->set_error("feed chunk is too large"); + } + + try { + RAW raw{Format::TXT, (void *)data, (int)len}; + decoder->nmea.Receive(&raw, 1, decoder->tag); + decoder->last_error.clear(); + return (int)decoder->sink.queue.size(); + } catch (const std::exception &e) { + return decoder->set_error(e); + } catch (...) { + return decoder->set_error("unknown C++ exception"); + } +} + +int aiscat_pending(AisCatDecoder *decoder) { + if (!decoder) { + return -1; + } + return (int)decoder->sink.queue.size(); +} + +int aiscat_next(AisCatDecoder *decoder, unsigned char **out, size_t *len) { + if (!decoder || !out || !len) { + return -1; + } + *out = nullptr; + *len = 0; + if (decoder->sink.queue.empty()) { + return 0; + } + + try { + std::string item = std::move(decoder->sink.queue.front()); + decoder->sink.queue.pop_front(); + + unsigned char *buf = nullptr; + if (!item.empty()) { + buf = (unsigned char *)std::malloc(item.size()); + if (!buf) { + return decoder->set_error("malloc failed"); + } + std::memcpy(buf, item.data(), item.size()); + } + + *out = buf; + *len = item.size(); + decoder->last_error.clear(); + return 1; + } catch (const std::exception &e) { + return decoder->set_error(e); + } catch (...) { + return decoder->set_error("unknown C++ exception"); + } +} + +void aiscat_free_bytes(unsigned char *data) { + std::free(data); +} + +const char *aiscat_last_error(AisCatDecoder *decoder) { + if (!decoder) { + return "decoder is null"; + } + return decoder->last_error.c_str(); +} + +} // extern "C" diff --git a/go/aiscat/bridge.h b/go/aiscat/bridge.h new file mode 100644 index 0000000000..9834c452d5 --- /dev/null +++ b/go/aiscat/bridge.h @@ -0,0 +1,37 @@ +// aiscat-go C ABI bridge for AIS-catcher's C++ decoder. +// SPDX-License-Identifier: GPL-3.0-or-later + +#pragma once + +#include + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct AisCatDecoder AisCatDecoder; + +enum AisCatFormat { + AISCAT_FORMAT_DICTIONARY = 0, + AISCAT_FORMAT_ANNOTATED = 1, + AISCAT_FORMAT_JSON = 2, + AISCAT_FORMAT_JSON_NMEA = 3, + AISCAT_FORMAT_NMEA = 4, + AISCAT_FORMAT_NMEA_TAG = 5, + AISCAT_FORMAT_BINARY = 6 +}; + +AisCatDecoder *aiscat_new(int format, int country); +void aiscat_free(AisCatDecoder *decoder); + +int aiscat_feed(AisCatDecoder *decoder, const char *data, size_t len); +int aiscat_pending(AisCatDecoder *decoder); + +int aiscat_next(AisCatDecoder *decoder, unsigned char **out, size_t *len); +void aiscat_free_bytes(unsigned char *data); + +const char *aiscat_last_error(AisCatDecoder *decoder); + +#ifdef __cplusplus +} +#endif diff --git a/go/aiscat/decode.go b/go/aiscat/decode.go new file mode 100644 index 0000000000..bd34712934 --- /dev/null +++ b/go/aiscat/decode.go @@ -0,0 +1,123 @@ +package aiscat + +import ( + "bytes" + "fmt" +) + +// Decode decodes one logical AIS message from one or more NMEA fragments. It +// returns Message for dictionary/annotated formats and []byte for byte-shaped +// formats. +func Decode(options Options, parts ...[]byte) (any, error) { + if len(parts) == 0 { + return nil, fmt.Errorf("decode requires at least one NMEA fragment") + } + decoder, err := NewDecoder(options) + if err != nil { + return nil, err + } + defer decoder.Close() + + for _, part := range parts { + if _, err := decoder.Feed(part); err != nil { + return nil, err + } + if _, err := decoder.Feed([]byte{'\n'}); err != nil { + return nil, err + } + } + + msg, ok, err := decoder.Next() + if err != nil { + return nil, err + } + if !ok { + return nil, fmt.Errorf("incomplete or invalid NMEA fragment(s)") + } + if extra, ok, err := decoder.Next(); err != nil { + return nil, err + } else if ok { + _ = extra + return nil, fmt.Errorf("decode expects one message; use a streaming decoder for multiple messages") + } + return msg, nil +} + +// DecodeString decodes one logical AIS message from string fragments. +func DecodeString(options Options, parts ...string) (any, error) { + byteParts := make([][]byte, len(parts)) + for i, part := range parts { + byteParts[i] = []byte(part) + } + return Decode(options, byteParts...) +} + +// DecodeMap decodes one logical AIS message into a Go map. It is valid for +// dictionary, annotated, json, and json_nmea formats. +func DecodeMap(options Options, parts ...[]byte) (Message, error) { + decoder, err := NewDecoder(options) + if err != nil { + return nil, err + } + defer decoder.Close() + + for _, part := range parts { + if _, err := decoder.Feed(part); err != nil { + return nil, err + } + if !bytes.HasSuffix(part, []byte{'\n'}) && !bytes.HasSuffix(part, []byte{'\r'}) { + if _, err := decoder.Feed([]byte{'\n'}); err != nil { + return nil, err + } + } + } + + msg, ok, err := decoder.NextMap() + if err != nil { + return nil, err + } + if !ok { + return nil, fmt.Errorf("incomplete or invalid NMEA fragment(s)") + } + if extra, ok, err := decoder.NextBytes(); err != nil { + return nil, err + } else if ok { + _ = extra + return nil, fmt.Errorf("decode expects one message; use a streaming decoder for multiple messages") + } + return msg, nil +} + +// DecodeBytes decodes one logical AIS message into bytes. It works for every +// format; dictionary and annotated are returned as JSON bytes. +func DecodeBytes(options Options, parts ...[]byte) ([]byte, error) { + decoder, err := NewDecoder(options) + if err != nil { + return nil, err + } + defer decoder.Close() + + for _, part := range parts { + if _, err := decoder.Feed(part); err != nil { + return nil, err + } + if _, err := decoder.Feed([]byte{'\n'}); err != nil { + return nil, err + } + } + + msg, ok, err := decoder.NextBytes() + if err != nil { + return nil, err + } + if !ok { + return nil, fmt.Errorf("incomplete or invalid NMEA fragment(s)") + } + if extra, ok, err := decoder.NextBytes(); err != nil { + return nil, err + } else if ok { + _ = extra + return nil, fmt.Errorf("decode expects one message; use a streaming decoder for multiple messages") + } + return msg, nil +} diff --git a/go/aiscat/decoder_test.go b/go/aiscat/decoder_test.go new file mode 100644 index 0000000000..88ba8b44a7 --- /dev/null +++ b/go/aiscat/decoder_test.go @@ -0,0 +1,101 @@ +//go:build cgo + +package aiscat + +import ( + "bytes" + "encoding/json" + "testing" +) + +const type1 = "!AIVDM,1,1,,A,15MgK45P3@G?fl0E`JbR0OwT0@MS,0*4E" + +func TestDecodeMapType1(t *testing.T) { + msg, err := DecodeMap(Options{Format: FormatDictionary}, []byte(type1)) + if err != nil { + t.Fatal(err) + } + if got := int(msg["type"].(float64)); got != 1 { + t.Fatalf("type = %d, want 1", got) + } + if got := int(msg["mmsi"].(float64)); got != 366730000 { + t.Fatalf("mmsi = %d, want 366730000", got) + } + if got := msg["lat"].(float64); got < 37.80 || got > 37.81 { + t.Fatalf("lat = %f, want around 37.803802", got) + } + if got := msg["lon"].(float64); got > -122.39 || got < -122.40 { + t.Fatalf("lon = %f, want around -122.392532", got) + } + for _, key := range []string{"class", "device", "version", "scaled", "rxtime", "nmea"} { + if _, ok := msg[key]; ok { + t.Fatalf("dictionary output should strip %q", key) + } + } +} + +func TestDecoderTwoSentences(t *testing.T) { + dec, err := NewDecoder(Options{Format: FormatDictionary}) + if err != nil { + t.Fatal(err) + } + defer dec.Close() + + if _, err := dec.Feed([]byte(type1 + "\r\n!AIVDM,1,1,,B,177KQJ5000G?tO`K>RA1wUbN0TKH,0*5C\r\n")); err != nil { + t.Fatal(err) + } + + var count int + for { + _, ok, err := dec.NextMap() + if err != nil { + t.Fatal(err) + } + if !ok { + break + } + count++ + } + if count != 2 { + t.Fatalf("decoded %d messages, want 2", count) + } +} + +func TestAnnotated(t *testing.T) { + msg, err := DecodeMap(Options{Format: FormatAnnotated}, []byte(type1)) + if err != nil { + t.Fatal(err) + } + lat, ok := msg["lat"].(map[string]any) + if !ok { + t.Fatalf("lat = %T, want annotated map", msg["lat"]) + } + if lat["unit"] != "degrees" { + t.Fatalf("lat unit = %v, want degrees", lat["unit"]) + } + if _, ok := lat["value"].(float64); !ok { + t.Fatalf("lat value = %T, want float64", lat["value"]) + } +} + +func TestByteFormats(t *testing.T) { + jsonBytes, err := DecodeBytes(Options{Format: FormatJSON}, []byte(type1)) + if err != nil { + t.Fatal(err) + } + var obj map[string]any + if err := json.Unmarshal(jsonBytes, &obj); err != nil { + t.Fatalf("json output did not unmarshal: %v\n%s", err, jsonBytes) + } + if obj["class"] != "AIS" { + t.Fatalf("json class = %v, want AIS", obj["class"]) + } + + nmeaBytes, err := DecodeBytes(Options{Format: FormatNMEA}, []byte(type1)) + if err != nil { + t.Fatal(err) + } + if !bytes.Contains(nmeaBytes, []byte(type1)) { + t.Fatalf("nmea output %q does not contain input", string(nmeaBytes)) + } +} diff --git a/go/aiscat/examples/decode/main.go b/go/aiscat/examples/decode/main.go new file mode 100644 index 0000000000..fd8b824e56 --- /dev/null +++ b/go/aiscat/examples/decode/main.go @@ -0,0 +1,19 @@ +package main + +import ( + "fmt" + "log" + + "github.com/jvde-github/ais-catcher/go/aiscat" +) + +func main() { + msg, err := aiscat.DecodeMap( + aiscat.Options{Format: aiscat.FormatDictionary, Country: true}, + []byte("!AIVDM,1,1,,A,15MgK45P3@G?fl0E`JbR0OwT0@MS,0*4E"), + ) + if err != nil { + log.Fatal(err) + } + fmt.Println(msg["type"], msg["mmsi"], msg["lat"], msg["lon"]) +} diff --git a/go/aiscat/examples/from_tcp/main.go b/go/aiscat/examples/from_tcp/main.go new file mode 100644 index 0000000000..169b75d0d5 --- /dev/null +++ b/go/aiscat/examples/from_tcp/main.go @@ -0,0 +1,36 @@ +package main + +import ( + "context" + "flag" + "fmt" + "log" + "time" + + "github.com/jvde-github/ais-catcher/go/aiscat" +) + +func main() { + address := flag.String("addr", "inst4:5012", "NMEA-over-TCP address") + flag.Parse() + + ctx := context.Background() + results, err := aiscat.FromTCP(ctx, *address, aiscat.Options{Format: aiscat.FormatDictionary}) + if err != nil { + log.Fatal(err) + } + + start := time.Now() + total := 0 + for result := range results { + if result.Err != nil { + log.Fatal(result.Err) + } + total++ + if total%1000 == 0 { + msg := result.Message.(aiscat.Message) + rate := float64(total) / time.Since(start).Seconds() + fmt.Printf("MMSI: %.0f | Total: %d | Rate: %.2f msg/s\n", msg["mmsi"], total, rate) + } + } +} diff --git a/go/aiscat/format.go b/go/aiscat/format.go new file mode 100644 index 0000000000..0977c2cd29 --- /dev/null +++ b/go/aiscat/format.go @@ -0,0 +1,80 @@ +package aiscat + +import "fmt" + +// Format selects the shape returned by Decoder.Next and Decode. +type Format int + +const ( + FormatDictionary Format = iota + FormatAnnotated + FormatJSON + FormatJSONNMEA + FormatNMEA + FormatNMEATag + FormatBinary +) + +func (f Format) String() string { + switch f { + case FormatDictionary: + return "dictionary" + case FormatAnnotated: + return "annotated" + case FormatJSON: + return "json" + case FormatJSONNMEA: + return "json_nmea" + case FormatNMEA: + return "nmea" + case FormatNMEATag: + return "nmea_tag" + case FormatBinary: + return "binary" + default: + return fmt.Sprintf("Format(%d)", int(f)) + } +} + +// ParseFormat accepts the same format names as the Python aiscat package. +func ParseFormat(s string) (Format, error) { + switch s { + case "", "dictionary": + return FormatDictionary, nil + case "annotated": + return FormatAnnotated, nil + case "json": + return FormatJSON, nil + case "json_nmea": + return FormatJSONNMEA, nil + case "nmea": + return FormatNMEA, nil + case "nmea_tag": + return FormatNMEATag, nil + case "binary": + return FormatBinary, nil + default: + return FormatDictionary, fmt.Errorf("unknown aiscat format %q", s) + } +} + +func (f Format) mapShaped() bool { + return f == FormatDictionary || f == FormatAnnotated +} + +func (f Format) valid() bool { + return f >= FormatDictionary && f <= FormatBinary +} + +// Options configures a Decoder. +type Options struct { + Format Format + Country bool +} + +func normalizeOptions(options Options) (Options, error) { + if !options.Format.valid() { + return Options{}, fmt.Errorf("unknown aiscat format %d", int(options.Format)) + } + return options, nil +} diff --git a/go/aiscat/go.mod b/go/aiscat/go.mod new file mode 100644 index 0000000000..5d15823ad9 --- /dev/null +++ b/go/aiscat/go.mod @@ -0,0 +1,3 @@ +module github.com/jvde-github/ais-catcher/go/aiscat + +go 1.22 diff --git a/go/aiscat/native.go b/go/aiscat/native.go new file mode 100644 index 0000000000..4b83f64353 --- /dev/null +++ b/go/aiscat/native.go @@ -0,0 +1,195 @@ +//go:build cgo + +package aiscat + +/* +#cgo CXXFLAGS: -std=c++11 -I../../Source -I../../Source/Application -I../../Source/Library -I../../Source/Marine -I../../Source/JSON -I../../Source/Utilities +#cgo !windows CXXFLAGS: -Wno-unused-parameter -Wno-sign-compare -Wno-psabi +#cgo windows LDFLAGS: -lws2_32 +#include +#include "bridge.h" +*/ +import "C" + +import ( + "encoding/json" + "errors" + "fmt" + "runtime" + "sync" + "unsafe" +) + +// Message is a decoded AIS message in map-shaped formats. +type Message = map[string]any + +// Decoder incrementally decodes NMEA, AIS-catcher JSON envelopes, and +// AIS-catcher binary packets using AIS-catcher's native decoder. +type Decoder struct { + mu sync.Mutex + ptr *C.AisCatDecoder + format Format + closed bool + options Options +} + +// NewDecoder creates a streaming decoder. Call Close when the decoder is no +// longer needed; a finalizer is also installed as a backstop. +func NewDecoder(options Options) (*Decoder, error) { + options, err := normalizeOptions(options) + if err != nil { + return nil, err + } + + country := C.int(0) + if options.Country { + country = 1 + } + ptr := C.aiscat_new(C.int(options.Format), country) + if ptr == nil { + return nil, errors.New("failed to create native aiscat decoder") + } + + decoder := &Decoder{ + ptr: ptr, + format: options.Format, + options: options, + } + runtime.SetFinalizer(decoder, (*Decoder).Close) + return decoder, nil +} + +// Close releases the native decoder. +func (d *Decoder) Close() { + d.mu.Lock() + defer d.mu.Unlock() + d.closeLocked() +} + +func (d *Decoder) closeLocked() { + if d.closed { + return + } + if d.ptr != nil { + C.aiscat_free(d.ptr) + d.ptr = nil + } + d.closed = true + runtime.SetFinalizer(d, nil) +} + +// Format returns the decoder output format. +func (d *Decoder) Format() Format { + return d.format +} + +// Feed appends a chunk to the decoder and returns the number of pending +// decoded messages. +func (d *Decoder) Feed(data []byte) (int, error) { + d.mu.Lock() + defer d.mu.Unlock() + if err := d.checkOpenLocked(); err != nil { + return 0, err + } + + var ptr *C.char + if len(data) > 0 { + ptr = (*C.char)(unsafe.Pointer(&data[0])) + } + n := C.aiscat_feed(d.ptr, ptr, C.size_t(len(data))) + if n < 0 { + return 0, d.lastErrorLocked() + } + return int(n), nil +} + +// FeedString appends a string chunk to the decoder. +func (d *Decoder) FeedString(data string) (int, error) { + return d.Feed(unsafe.Slice(unsafe.StringData(data), len(data))) +} + +// Pending returns the number of decoded messages waiting in the queue. +func (d *Decoder) Pending() (int, error) { + d.mu.Lock() + defer d.mu.Unlock() + if err := d.checkOpenLocked(); err != nil { + return 0, err + } + + n := C.aiscat_pending(d.ptr) + if n < 0 { + return 0, d.lastErrorLocked() + } + return int(n), nil +} + +// NextBytes pops the next decoded message as bytes. Map-shaped formats are +// returned as JSON bytes. +func (d *Decoder) NextBytes() ([]byte, bool, error) { + d.mu.Lock() + defer d.mu.Unlock() + if err := d.checkOpenLocked(); err != nil { + return nil, false, err + } + + var out *C.uchar + var outLen C.size_t + rc := C.aiscat_next(d.ptr, &out, &outLen) + if rc < 0 { + return nil, false, d.lastErrorLocked() + } + if rc == 0 { + return nil, false, nil + } + defer C.aiscat_free_bytes(out) + if outLen == 0 { + return []byte{}, true, nil + } + data := C.GoBytes(unsafe.Pointer(out), C.int(outLen)) + return data, true, nil +} + +// NextMap pops the next decoded message and decodes it into a Go map. It is +// valid for dictionary, annotated, json, and json_nmea formats. +func (d *Decoder) NextMap() (Message, bool, error) { + data, ok, err := d.NextBytes() + if !ok || err != nil { + return nil, ok, err + } + + switch d.format { + case FormatDictionary, FormatAnnotated, FormatJSON, FormatJSONNMEA: + default: + return nil, false, fmt.Errorf("format %s cannot be decoded as a map", d.format) + } + + var msg Message + if err := json.Unmarshal(data, &msg); err != nil { + return nil, false, err + } + return msg, true, nil +} + +// Next pops the next decoded message. It returns Message for dictionary and +// annotated formats, and []byte for byte-shaped formats. +func (d *Decoder) Next() (any, bool, error) { + if d.format.mapShaped() { + return d.NextMap() + } + return d.NextBytes() +} + +func (d *Decoder) checkOpenLocked() error { + if d == nil || d.closed || d.ptr == nil { + return errors.New("aiscat decoder is closed") + } + return nil +} + +func (d *Decoder) lastErrorLocked() error { + msg := C.aiscat_last_error(d.ptr) + if msg == nil { + return errors.New("native aiscat error") + } + return errors.New(C.GoString(msg)) +} diff --git a/go/aiscat/nocgo.go b/go/aiscat/nocgo.go new file mode 100644 index 0000000000..b8aaf84675 --- /dev/null +++ b/go/aiscat/nocgo.go @@ -0,0 +1,57 @@ +//go:build !cgo + +package aiscat + +import "errors" + +var errCgoDisabled = errors.New("aiscat requires cgo and a C++11 compiler") + +// Message is a decoded AIS message in map-shaped formats. +type Message = map[string]any + +// Decoder is unavailable when cgo is disabled. +type Decoder struct { + format Format +} + +// NewDecoder returns an error when the package is built without cgo. +func NewDecoder(options Options) (*Decoder, error) { + options, err := normalizeOptions(options) + if err != nil { + return nil, err + } + return &Decoder{format: options.Format}, errCgoDisabled +} + +func (d *Decoder) Close() {} + +func (d *Decoder) Format() Format { + if d == nil { + return FormatDictionary + } + return d.format +} + +func (d *Decoder) Feed([]byte) (int, error) { + return 0, errCgoDisabled +} + +func (d *Decoder) FeedString(string) (int, error) { + return 0, errCgoDisabled +} + +func (d *Decoder) Pending() (int, error) { + return 0, errCgoDisabled +} + +func (d *Decoder) NextBytes() ([]byte, bool, error) { + return nil, false, errCgoDisabled +} + +func (d *Decoder) NextMap() (Message, bool, error) { + return nil, false, errCgoDisabled +} + +func (d *Decoder) Next() (any, bool, error) { + return nil, false, errCgoDisabled +} diff --git a/go/aiscat/stream.go b/go/aiscat/stream.go new file mode 100644 index 0000000000..a5058faa94 --- /dev/null +++ b/go/aiscat/stream.go @@ -0,0 +1,166 @@ +package aiscat + +import ( + "context" + "errors" + "io" + "net" +) + +// Result is one item emitted by stream helpers. +type Result struct { + Message any + Err error +} + +// FromReader decodes AIS messages from r until EOF, context cancellation, or a +// decode/read error. The channel is closed when the stream ends. +func FromReader(ctx context.Context, r io.Reader, options Options) (<-chan Result, error) { + decoder, err := NewDecoder(options) + if err != nil { + return nil, err + } + + out := make(chan Result) + go func() { + defer close(out) + defer decoder.Close() + + buf := make([]byte, 64*1024) + for { + if err := ctx.Err(); err != nil { + sendResult(ctx, out, Result{Err: err}) + return + } + + n, readErr := r.Read(buf) + if n > 0 { + if _, err := decoder.Feed(buf[:n]); err != nil { + sendResult(ctx, out, Result{Err: err}) + return + } + if !drainDecoder(ctx, decoder, out) { + return + } + } + + if readErr != nil { + if !errors.Is(readErr, io.EOF) { + sendResult(ctx, out, Result{Err: readErr}) + } + return + } + } + }() + return out, nil +} + +// FromTCP connects to address and decodes messages until EOF, context +// cancellation, or an error. Reconnect policy is intentionally left to callers. +func FromTCP(ctx context.Context, address string, options Options) (<-chan Result, error) { + var dialer net.Dialer + conn, err := dialer.DialContext(ctx, "tcp", address) + if err != nil { + return nil, err + } + go func() { + <-ctx.Done() + conn.Close() + }() + + results, err := FromReader(ctx, conn, options) + if err != nil { + conn.Close() + return nil, err + } + + out := make(chan Result) + go func() { + defer close(out) + defer conn.Close() + for result := range results { + if !sendResult(ctx, out, result) { + return + } + } + }() + return out, nil +} + +// FromUDP listens on address and decodes inbound AIS datagrams until context +// cancellation or an error. +func FromUDP(ctx context.Context, address string, options Options) (<-chan Result, error) { + addr, err := net.ResolveUDPAddr("udp", address) + if err != nil { + return nil, err + } + conn, err := net.ListenUDP("udp", addr) + if err != nil { + return nil, err + } + + decoder, err := NewDecoder(options) + if err != nil { + conn.Close() + return nil, err + } + + out := make(chan Result) + go func() { + defer close(out) + defer conn.Close() + defer decoder.Close() + + go func() { + <-ctx.Done() + conn.Close() + }() + + buf := make([]byte, 64*1024) + for { + n, _, readErr := conn.ReadFromUDP(buf) + if n > 0 { + if _, err := decoder.Feed(buf[:n]); err != nil { + sendResult(ctx, out, Result{Err: err}) + return + } + if !drainDecoder(ctx, decoder, out) { + return + } + } + if readErr != nil { + if ctx.Err() != nil { + sendResult(ctx, out, Result{Err: ctx.Err()}) + } else { + sendResult(ctx, out, Result{Err: readErr}) + } + return + } + } + }() + return out, nil +} + +func drainDecoder(ctx context.Context, decoder *Decoder, out chan<- Result) bool { + for { + msg, ok, err := decoder.Next() + if err != nil { + return sendResult(ctx, out, Result{Err: err}) + } + if !ok { + return true + } + if !sendResult(ctx, out, Result{Message: msg}) { + return false + } + } +} + +func sendResult(ctx context.Context, out chan<- Result, result Result) bool { + select { + case <-ctx.Done(): + return false + case out <- result: + return true + } +}