Metadata-Version: 2.5
Name: igwn-connect
Version: 0.8.0
Summary: A service built on SGN connecting data between LLDD, Arrakis and frame files
Project-URL: Homepage, https://git.ligo.org/ngdd/igwn-connect
Project-URL: Issue Tracker, https://git.ligo.org/ngdd/igwn-connect/issues
Project-URL: Source Code, https://git.ligo.org/ngdd/igwn-connect.git
Author-email: Olivia Godwin <olivia.godwin@ligo.org>
Maintainer-email: Olivia Godwin <olivia.godwin@ligo.org>
License-Expression: GPL-3.0-or-later
License-File: LICENSE
Classifier: Development Status :: 2 - Pre-Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)
Classifier: Natural Language :: English
Classifier: Operating System :: POSIX
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Scientific/Engineering
Classifier: Topic :: Scientific/Engineering :: Astronomy
Classifier: Topic :: Scientific/Engineering :: Physics
Requires-Python: >=3.11
Requires-Dist: arrakis>=0.17
Requires-Dist: gpstime
Requires-Dist: igwn-lldd-common>=0.4.1
Requires-Dist: sgn-arrakis>=0.10
Requires-Dist: sgn-gwframe>=0.9
Requires-Dist: sgn-ts>=0.14
Requires-Dist: sgn>=0.12.2
Requires-Dist: sgnmon>=0.2.1
Provides-Extra: dev
Requires-Dist: markdown-callouts>=0.2; extra == 'dev'
Requires-Dist: markdown-exec>=0.5; extra == 'dev'
Requires-Dist: mkdocs-coverage>=0.2; extra == 'dev'
Requires-Dist: mkdocs-gen-files>=0.3; extra == 'dev'
Requires-Dist: mkdocs-literate-nav>=0.4; extra == 'dev'
Requires-Dist: mkdocs-material-igwn; extra == 'dev'
Requires-Dist: mkdocs-section-index>=0.3; extra == 'dev'
Requires-Dist: mkdocs>=1.3; extra == 'dev'
Requires-Dist: mkdocstrings[python]; extra == 'dev'
Requires-Dist: mypy; extra == 'dev'
Requires-Dist: mypy-extensions; extra == 'dev'
Requires-Dist: pip; extra == 'dev'
Requires-Dist: pytest; extra == 'dev'
Requires-Dist: pytest-arrakis>=0.4.1; extra == 'dev'
Requires-Dist: pytest-cov; extra == 'dev'
Requires-Dist: ruff; extra == 'dev'
Requires-Dist: toml>=0.10; extra == 'dev'
Provides-Extra: docs
Requires-Dist: markdown-callouts>=0.2; extra == 'docs'
Requires-Dist: markdown-exec>=0.5; extra == 'docs'
Requires-Dist: mkdocs-coverage>=0.2; extra == 'docs'
Requires-Dist: mkdocs-gen-files>=0.3; extra == 'docs'
Requires-Dist: mkdocs-literate-nav>=0.4; extra == 'docs'
Requires-Dist: mkdocs-material-igwn; extra == 'docs'
Requires-Dist: mkdocs-section-index>=0.3; extra == 'docs'
Requires-Dist: mkdocs>=1.3; extra == 'docs'
Requires-Dist: mkdocstrings[python]; extra == 'docs'
Requires-Dist: toml>=0.10; extra == 'docs'
Provides-Extra: lint
Requires-Dist: mypy; extra == 'lint'
Requires-Dist: mypy-extensions; extra == 'lint'
Requires-Dist: pip; extra == 'lint'
Requires-Dist: ruff; extra == 'lint'
Provides-Extra: test
Requires-Dist: pytest; extra == 'test'
Requires-Dist: pytest-arrakis>=0.4.1; extra == 'test'
Requires-Dist: pytest-cov; extra == 'test'
Description-Content-Type: text/markdown

<h1 align="center">igwn-connect</h1>

<p align="center">A service built on <a href="https://git.ligo.org/greg/sgn">SGN</a>, connecting data between LLDD, Arrakis, and frame files.</p>

<p align="center">
  <a href="https://git.ligo.org/ngdd/igwn-connect/-/pipelines/latest">
    <img alt="ci" src="https://git.ligo.org/ngdd/igwn-connect/badges/main/pipeline.svg" />
  </a>
  <a href="https://git.ligo.org/ngdd/igwn-connect/-/pipelines/latest">
    <img alt="coverage" src="https://git.ligo.org/ngdd/igwn-connect/badges/main/coverage.svg" />
  </a>
  <a href="https://pypi.org/project/igwn-connect/">
    <img alt="pypi version" src="https://img.shields.io/pypi/v/igwn-connect.svg" />
  </a>
</p>

---

Stream gravitational-wave detector timeseries between the low-latency data
distribution system (LLDD), the [Arrakis](https://docs.ligo.org/ngdd/arrakis-python)
timeseries service, and GWF frame files. Pick a `--source` and a `--sink` and
the connector builds and runs the [SGN](https://git.ligo.org/greg/sgn)
pipeline between them.

## Resources

* [Source Code](https://git.ligo.org/ngdd/igwn-connect)
* [Issue Tracker](https://git.ligo.org/ngdd/igwn-connect/-/issues)

## Installation

```
pip install igwn-connect
```

Or from source:

```
git clone git@git.ligo.org:ngdd/igwn-connect.git
cd igwn-connect
pip install .
```

## Features

* Sources: LLDD (Kafka frame streams), Arrakis, GWF frame file directories,
  replayed frame sets, and synthetic test signals
* Sinks: Arrakis, LLDD, GWF frame files, and a null sink for testing
* Any source can be paired with any sink
* Live streams are resilient to interruptions: sources reconnect with capped
  exponential backoff and bridge outages with gap buffers, and sinks drop a
  file or block that fails to write rather than stopping
  (`--max-write-failures`, `--max-publish-failures`)
* Bounded (`--start`/`--end`) or continuous live operation
* Arrakis replay namespace support for both streaming and publishing
  (`--replay-id`)
* Replay a fixed set of frame files as live data (`--source replay`):
  retimestamped to the present, paced to the wall clock, and looped
* Frame writing options: file retention (`--max-files`, `--retention-time`),
  and skipping files that contain no real data (`--skip-all-gap`)
* Live monitoring of the running pipeline (`--monitor`): an HTML dashboard,
  Prometheus metrics, and a health endpoint

## Quickstart

The CLI shows contextual options: pass `--source` and/or `--sink` with
`--help` to see the options relevant to that combination, e.g.

``` bash
igwn-connect --source frames --sink arrakis --help
```

### Publish live DAQ data from LLDD to Arrakis

``` bash
igwn-connect --source lldd --sink arrakis \
    --ifo L1 --topic Live_LLO_Data \
    --bootstrap-servers kafka1:9092,kafka2:9092 \
    --arrakis-url grpc://arrakis-online1:31206 \
    --publisher-id L1-lldd
```

The channels to publish are discovered from the publisher's registration on
the Arrakis server.

### Write an Arrakis stream to frame files

The Arrakis source reads the server URL from the `ARRAKIS_SERVER`
environment variable:

``` bash
export ARRAKIS_SERVER=grpc://arrakis-online1:31206
igwn-connect --source arrakis --sink frames \
    --channels L1:GDS-CALIB_STRAIN L1:GDS-CALIB_STATE_VECTOR \
    --frame-duration 64 --output-dir /data/frames \
    --skip-all-gap
```

Without `--start` and `--end` this streams live data continuously; press
`Ctrl+C` to stop.

### Publish frame files to LLDD

Watch a directory for new frame files and publish them to Kafka:

``` bash
igwn-connect --source frames --sink lldd \
    --channels H1:GDS-CALIB_STRAIN \
    --watch-dir /data/live/H1 \
    --ifo H1 --topic Live_LHO_Data \
    --bootstrap-servers kafka1:9092
```

### Replay frame files into Arrakis

Replay a frame set as if it were live, publishing into a replay namespace:

``` bash
igwn-connect --source replay --sink arrakis \
    --frames /data/O3ReplayMDC/L1 \
    --arrakis-url grpc://arrakis-replay1:31206 \
    --publisher-id L1-lldd-O3Replay --replay-id O3Replay
```

`--frames` takes directories of `.gwf` files, LAL `.cache` files, or
individual `.gwf` files (T050017 names), in any mix. The replay window
`[replay_start, replay_end)` is mapped onto the present by adding a whole
number of window durations, the same mapping the Arrakis frames backend
uses to serve replays, so every instance replaying the same window stays
coherent. With `--sink arrakis` the window defaults to the server's
registered window for `--replay-id` and the channels come from the
publisher's registration; other sinks need `--replay-start`,
`--replay-end`, and `--channels`. Data is delivered one second at a time
once the wall clock passes it, spans no file covers are delivered as gaps,
and the window loops until `--end` or `Ctrl+C`. `--target-latency N`
holds each second of data back a further N seconds, emulating the
transport latency of a live stream.

### Test a pipeline with synthetic data

``` bash
igwn-connect --source fake --sink null \
    --channels H1:TEST --rate 2048 --signal-type sin \
    --start 0 --end 10
```

## Monitoring

Serve live monitoring for the running pipeline with
[sgnmon](https://greg.docs.ligo.org/sgnmon):

``` bash
igwn-connect --source arrakis --sink frames \
    --channels L1:GDS-CALIB_STRAIN \
    --frame-duration 64 --output-dir /data/frames \
    --monitor
```

This starts a background web server (default port 9090; `--monitor-port`
changes it and implies `--monitor`, with `0` picking a free port) exposing:

* `/` — a live dashboard drawing the pipeline graph with per-channel rates,
  latencies, and gap fractions
* `/metrics` — Prometheus metrics for scraping
* `/health` — a JSON health report (HTTP 503 when unhealthy), usable directly
  by container orchestration or `sgnmon check`
* `/readyz` — readiness probe: 200 while the pipeline's run loop is running
  (between ready and stopping) and every health check passes, 503 while
  starting, stopping, or a check fails
* `/healthz` — liveness probe: 200 while the run loop keeps striding, 503
  once it has gone stale
* `/status` — a JSON snapshot of all probes

Rather than observing every pad, the connector taps each link once at the
receiving element, plus post-alignment consumption (`adp`) to distinguish
"data arrived" from "data processed", and the source's output pads so input
rate and latency are visible per channel at the origin. The data-freshness
health check tolerates three missed output strides (`--frame-duration` for
the frames sink, `--delta-t` for the LLDD sink) before failing, so slow
output cadences do not false-alarm.

`--monitor-max-latency N` (implies `--monitor`) adds a publishing-latency check:
`/health` and `/readyz` fail when the data the sink is publishing or
writing trails real time by more than N seconds, per channel, and
`/health/max_latency` reports that check alone. It applies after the same
grace period as the freshness check, so a source still catching up at
startup does not trip it, and it is only meaningful for live data.

### Health reporting under systemd

The pipeline reports its lifecycle through
[`sgn.health`](https://docs.ligo.org/greg/sgn/user/report-health/): a
`Type=notify` unit (or podman's default `--sdnotify=container`) sets
`NOTIFY_SOCKET`, and the connector then sends `READY=1` once the graph is
running, `WATCHDOG=1` on every stride, and `STOPPING=1` when the run loop
exits. This needs no flag; `--monitor` adds the monitoring server's probes
alongside it, driven by the same lifecycle.

```ini
[Service]
Type=notify
ExecStart=/usr/bin/igwn-connect --source lldd --sink arrakis ...
WatchdogSec=60s
TimeoutStopSec=120s
Restart=on-failure
```

* `WATCHDOG=1` is only sent as the graph strides, so `WatchdogSec=` must
  exceed the longest stall a source can have: the LLDD source's
  `--poll-timeout` is 1 s and the frames source's `--queue-timeout` is 10 s
  by default, so `60s` leaves ample margin.
* On SIGTERM the connector drains the pipeline to end-of-stream, then sends
  `STOPPING=1`, which disarms the watchdog and starts `TimeoutStopSec=`;
  set it generously enough for the sink to flush.

## Usage with Docker

The container image is hosted on `containers.ligo.org`. Pull the latest version with:

```bash
podman pull docker://containers.ligo.org/ngdd/igwn-connect:latest
```

Run with podman (or docker):

```bash
podman run --rm --net=host \
    docker://containers.ligo.org/ngdd/igwn-connect:latest \
    --source lldd --sink arrakis \
    --ifo L1 \
    --arrakis-url grpc://arrakis-online1:31206 \
    --publisher-id L1-lldd \
    --bootstrap-servers kafka1:9092,kafka2:9092 \
    --topic Live_LLO_Data
```
