Metadata-Version: 2.5
Name: igwn-event-normalizer
Version: 0.1.6
Summary: Normalize GW event formats from GraceDB and SCiMMA
Project-URL: Homepage, https://git.ligo.org/operations/e2e-test-infra/event-normalizer
Requires-Python: >=3.9
Requires-Dist: pydantic>=2.0
Provides-Extra: dev
Requires-Dist: build>=1.2; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: pyyaml>=6.0; extra == 'dev'
Requires-Dist: ruff>=0.12; extra == 'dev'
Requires-Dist: twine>=6.0; extra == 'dev'
Description-Content-Type: text/markdown

# IGWN Event Normalizer

`igwn-event-normalizer` normalizes IGWN gravitational-wave event payloads into
a common typed schema.

It is designed for GraceDB-style event JSON. Different GW pipelines expose
their information through different structures; this package maps
those structures into a stable `NormalizedEvent` model.

Normalized events can be used directly as Pydantic models, converted into a
flat Python dictionary, or serialized as structured or flat JSON.

## Features

- Parse GraceDB-style gravitational-wave event payloads.
- Normalize common event metadata.
- Normalize CBC, Burst, MLy, and aframe structures.
- Extract detector channels for H1, L1, V1, and K1.
- Extract coincidence-level and detector-level CBC fields.
- Extract MultiBurst and MLyBurst fields.
- Normalize known GraceDB labels.
- Preserve selected GraceDB links.
- Convert instrument lists into scalar detector-presence fields.
- Serialize normalized events as dictionaries or JSON.
- Reduce raw event payloads to selected dot-separated paths.
- Validate normalized output with Pydantic.
- Ignore raw fields that are not explicitly mapped by the normalizer.
- Parse initial event upload messages from the GraceDB Kafka ingestion stream
  (`coinc.xml` for CBC pipelines, `.json` for aframe and MLy, and `trigger.txt`
  for cWB).
- Normalize `p_astro.json`, `em_bright.json`, and generic file upload metadata.

---

## Requirements

- Python 3.11 or newer.

---

## Installation

Install from PyPI:

```bash
python -m pip install igwn-event-normalizer
```

For development:

```bash
git clone https://git.ligo.org/operations/e2e-test-infra/event-normalizer.git
cd event-normalizer

python -m pip install -e ".[dev]"
```

---

## Overview

The package separates reduction, normalization, flattening, and JSON
serialization:

```text
Raw GraceDB event
        |
        +--> reduce_event()          -> reduced raw event
        |
        +--> parse_gracedb_event()   -> NormalizedEvent
                                              |
                                              +--> to_flat_dict()
                                              |
                                              +--> JSONSerializer
```


---

# Normalization model

GraceDB event payloads contain common root-level fields and
pipeline-specific `extra_attributes`.

The normalizer maps only supported fields into the output model. Raw fields
that are not explicitly extracted are ignored and are not propagated to
`NormalizedEvent` or to flat output.

Common normalized root-level fields include:

```text
uid
event_type
group
pipeline
search
far
far_is_upper_limit
instruments
H1_channel
L1_channel
V1_channel
K1_channel
reporting_latency
gpstime
submitter
offline
nevents
likelihood
superevent
created
processing_status
content_id
last_updated
```

Pipeline-specific data is stored in nested models:

| Source table | Normalized model attribute | Flat prefix |
|---|---|---|
| `CoincInspiral` | `event.coinc` | `coinc_` |
| `MultiBurst` | `event.burst` | `burst_` |
| `MLyBurst` | `event.mly` | `mly_` |
| H1 `SingleInspiral` | `event.single_H1` | `single_H1_` |
| L1 `SingleInspiral` | `event.single_L1` | `single_L1_` |
| V1 `SingleInspiral` | `event.single_V1` | `single_V1_` |
| K1 `SingleInspiral` | `event.single_K1` | `single_K1_` |
| GraceDB labels | `event.labels` | `label_` |
| GraceDB links | `event.links` | see [Links](#links) |
| `p_astro.json` | `event.p_astro` | `p_astro_` |
| `em_bright.json` | `event.em_bright` | `em_bright_` |
| Generic file metadata | `event.file` | `file_` |

---

# Pipeline-specific normalization

## Standard CBC pipelines

Standard CBC events provide technical data through:

```text
extra_attributes.CoincInspiral
extra_attributes.SingleInspiral
```

### CoincInspiral

`CoincInspiral` contains coincidence-level properties, such as total mass,
chirp mass, network SNR, participating interferometers, and timing fields.

These become fields on:

```python
event.coinc
```

For example:

```python
event.coinc.coinc_mass
event.coinc.coinc_mchirp
event.coinc.coinc_snr
event.coinc.coinc_ifos
event.coinc.coinc_end_time
event.coinc.coinc_combined_far
```

### SingleInspiral

`SingleInspiral` contains detector-specific entries. The normalizer maps each
supported detector to a separate field:

```python
event.single_H1
event.single_L1
event.single_V1
event.single_K1
```

For example:

```python
event.single_H1.single_snr
event.single_H1.single_mass1
event.single_H1.single_mass2
event.single_H1.single_mchirp
event.single_H1.single_spin1z
```

The detector identity is encoded by the parent field name. For example,
`event.single_H1` already identifies the H1 entry, so an additional `ifo`
field is not repeated in the normalized per-detector model.

Detector channel names are handled separately:

```python
event.H1_channel
event.L1_channel
event.V1_channel
event.K1_channel
```

---

## Burst pipelines

Burst events use:

```text
extra_attributes.MultiBurst
```

These are normalized into:

```python
event.burst
```

Examples:

```python
event.burst.burst_duration
event.burst.burst_mchirp
event.burst.burst_snr
event.burst.burst_ifos
event.burst.burst_start_time
event.burst.burst_peak_time
event.burst.burst_central_freq
event.burst.burst_bandwidth
event.burst.burst_amplitude
event.burst.burst_false_alarm_rate
```

For Burst events, detector channels are extracted from:

```text
MultiBurst.hoft
```

The calibration marker `C00`, when present, is ignored.

---

## MLy pipelines

MLy events use:

```text
extra_attributes.MLyBurst
```

These are normalized into:

```python
event.mly
```

Examples:

```python
event.mly.mly_central_freq
event.mly.mly_bandwidth
event.mly.mly_central_time
event.mly.mly_detection_statistic
event.mly.mly_mchirp
event.mly.mly_SNR
event.mly.mly_scores_coherency
event.mly.mly_scores_coincidence
event.mly.mly_scores_combined
```

MLy detector channels are extracted from:

```text
MLyBurst.channels
```

---

## aframe events

aframe events are classified as CBC:

```python
event.group == "CBC"
event.pipeline == "aframe"
```

However, aframe attributes use the `MLyBurst` structure.

Therefore aframe events:

- populate `event.mly`;
- extract channels from `MLyBurst.channels`;
- do not populate `event.coinc`;
- do not populate `event.single_H1`, `event.single_L1`, and related
  `SingleInspiral` fields.

---

# Channels, instruments, labels, and links

## Channels

The normalized model contains one root-level channel field for each supported
detector:

```python
event.H1_channel
event.L1_channel
event.V1_channel
event.K1_channel
```

When no channel is available, the value is the string:

```python
"None"
```

This is intentional: it allows downstream monitoring systems to query and
count missing channel values without relying on null handling.

---

## Instruments

GraceDB often represents instruments as a comma-separated string:

```json
{
  "instruments": "H1,L1,V1"
}
```

The normalized model stores the value as a list:

```python
event.instruments == ["H1", "L1", "V1"]
```

The flat representation converts this list into numeric scalar fields:

```python
{
    "instruments_H1": 1,
    "instruments_L1": 1,
    "instruments_V1": 1,
    "instruments_K1": 0,
}
```

This avoids lists in monitoring-oriented output.

---

## Labels

Known GraceDB labels are represented as optional booleans in:

```python
event.labels
```

For example:

```python
event.labels.label_MOCK
event.labels.label_SKYMAP_READY
event.labels.label_PASTRO_READY
event.labels.label_EMBRIGHT_READY
event.labels.label_EARLY_WARNING
event.labels.label_cWB_r
```

Only labels declared in `EventLabels` are retained.

Unknown labels are skipped and logged as warnings.

---

## Links

GraceDB links are stored in:

```python
event.links
```

Supported link fields include:

```python
event.links.neighbors
event.links.log
event.links.emobservations
event.links.files
event.links.labels
event.links.self
event.links.tags
```

In structured output, links remain nested:

```python
structured = event.model_dump(mode="json")

print(structured["links"]["self"])
print(structured["links"]["log"])
```

In the flat representation, link fields are promoted to root-level fields:

```python
flat = to_flat_dict(event)

print(flat["self"])
print(flat["log"])
print(flat["files"])
```

---

# File artifacts: p_astro, em_bright, and generic files

Some event information is provided as separate file artifacts uploaded to
GraceDB rather than as fields in the main GraceDB event payload.

The package provides parsers and attachment helpers for:

```text
p_astro.json     (EventPastro)
em_bright.json   (EventEMBright)
generic files    (EventFile)
```

Each artifact model also supports optional file upload metadata (`created_delay`,
`file_version`, and `file_size`), allowing downstream monitoring systems to
track upload latency, file versions, empty payloads (`size == 0`), and download
failures (`size == -1`).

The caller is responsible for associating the correct artifact file with the
correct event.

---

## p_astro.json

A `p_astro.json` artifact contains source-class probabilities.

Example:

```json
{
  "Terrestrial": 0.999996409,
  "BNS": 0.000003591,
  "BBH": 0.0,
  "NSBH": 0.0
}
```

Parse it with optional file metadata:

```python
from event_normalizer import parse_pastro

p_astro = parse_pastro(
    raw_pastro,
    created_delay=6,
    file_version=0,
    file_size=89,
)
```

The normalized model contains:

```python
p_astro.p_astro_terrestrial
p_astro.p_astro_bns
p_astro.p_astro_bbh
p_astro.p_astro_nsbh
p_astro.p_astro_created_delay
p_astro.p_astro_version
p_astro.p_astro_size
```

### Validation rules

When a valid payload is provided, all four probability fields are required:

```text
Terrestrial
BNS
BBH
NSBH
```

Each value must be in the interval:

```text
0 <= probability <= 1
```

The four probabilities must sum to approximately `1.0`. A small floating-point
tolerance is accepted.

For monitoring failed or empty file downloads:
- If `file_size` is `-1` (download failed) or `0` (empty file), `parse_pastro()`
  returns an `EventPastro` instance with `None` for all four probabilities and
  preserves the metadata fields.
- If `file_size > 0` is provided but required probability fields are missing
  (e.g. an empty `{}` payload), `p_astro_size` is set to `0` and the
  probabilities are set to `None`.

---

### Attaching p_astro to an event

```python
from event_normalizer import (
    attach_pastro,
    parse_gracedb_event,
    parse_pastro,
)

event = parse_gracedb_event(raw_event)
p_astro = parse_pastro(raw_pastro, created_delay=6, file_version=0, file_size=89)

event = attach_pastro(event, p_astro)
```

The attachment operation returns a new validated `NormalizedEvent`; it does not
modify the original event instance.

A raw dictionary and metadata can also be attached directly:

```python
event = attach_pastro(
    event,
    raw_pastro,
    created_delay=6,
    file_version=0,
    file_size=89,
)
```

In flat output, the fields are promoted to root-level keys:

```python
flat_event = to_flat_dict(event)

print(flat_event["p_astro_terrestrial"])
print(flat_event["p_astro_bns"])
print(flat_event["p_astro_created_delay"])
print(flat_event["p_astro_version"])
print(flat_event["p_astro_size"])
```

---

## em_bright.json

An `em_bright.json` artifact contains EM-bright-related probabilities.

Example:

```json
{
  "HasNS": 1.0,
  "HasRemnant": 1.0,
  "HasMassGap": 0.0678271766372453,
  "HasSSM": 0.25
}
```

Parse the artifact with optional file metadata:

```python
from event_normalizer import parse_embright

em_bright = parse_embright(
    raw_em_bright,
    created_delay=4,
    file_version=0,
    file_size=52,
)
```

The normalized model contains:

```python
em_bright.em_bright_has_ns
em_bright.em_bright_has_remnant
em_bright.em_bright_has_mass_gap
em_bright.em_bright_has_ssm
em_bright.em_bright_created_delay
em_bright.em_bright_version
em_bright.em_bright_size
```

### Validation rules

When a valid payload is provided, `HasNS` is required and `HasRemnant`,
`HasMassGap`, and `HasSSM` are optional. Every provided value must be in the
interval `0 <= probability <= 1`. Unlike `p_astro`, the values are independent
probabilities and are not required to sum to `1.0`.

As with `p_astro`:
- If `file_size` is `-1` (download failed) or `0` (empty file), or if
  `file_size > 0` is supplied with missing `HasNS`, probabilities are set to
  `None` and `em_bright_size` records `-1` or `0`.

---

### Attaching em_bright to an event

```python
from event_normalizer import (
    attach_embright,
    parse_gracedb_event,
)

event = parse_gracedb_event(raw_event)

event = attach_embright(
    event,
    raw_em_bright,
    created_delay=4,
    file_version=0,
    file_size=52,
)
```

In flat output, available fields become root-level keys:

```python
flat_event = to_flat_dict(event)

print(flat_event["em_bright_has_ns"])
print(flat_event["em_bright_created_delay"])
print(flat_event["em_bright_version"])
print(flat_event["em_bright_size"])
```

---

## Generic file metadata (`EventFile`)

For other GraceDB file artifacts (such as `bayestar.multiorder.fits`,
`coinc.xml`, or sky map images) where only upload metadata is needed, use
`parse_file()` and `attach_file()`:

```python
from event_normalizer.parsers import attach_file, parse_gracedb_event

event = parse_gracedb_event(raw_event)

event = attach_file(
    event,
    file_name="bayestar.multiorder.fits",
    created_delay=15,
    file_version=0,
    file_size=14520,
)
```

The normalized `EventFile` model is stored at `event.file` and exposes:

```python
event.file.file_name
event.file.file_created_delay
event.file.file_version
event.file.file_size
```

In flat output, these are promoted to root-level keys (`file_name`,
`file_created_delay`, `file_version`, `file_size`).

--- 

# Initial Kafka upload parsing (`parse_initial_upload`)

Before GraceDB emits downstream event alerts, analysis pipelines upload initial
trigger files through the GraceDB Kafka ingestion broker.

`parse_initial_upload()` accepts a GraceDB Kafka upload message dictionary,
decodes the embedded base64 `eventFile`, parses the pipeline-specific trigger
file, computes `reporting_latency` and `file_created_delay` relative to the
physical `gpstime`, and returns a validated `NormalizedEvent` with `event.file`
populated.

### Supported initial upload formats

| Pipeline | Group | Initial file format | Source content | Populated `NormalizedEvent` fields |
|---|---|---|---|---|
| `gstlal`, `pycbc`, `mbta`, `spiir`, `sgnl` | `CBC` | LIGO_LW XML (`coinc.xml`) | `coinc_inspiral`, `sngl_inspiral`, and `coinc_event` tables | Root fields (`gpstime`, `far`, `instruments`, `nevents`, `likelihood`, channels), `event.coinc`, `event.single_*`, `event.labels`, `event.file` |
| `aframe` | `CBC` | Flat JSON (`*.json`) | `gpstime`, `far`, `ifos`, `channels`, `detection_statistic`, `end_time`, `end_time_ns`, `mchirp`, `mass1`, `mass2`, `mtotal`, `spin1z`, `spin2z`, `template_duration` | Root fields (`gpstime`, `far`, `instruments`, channels), `event.mly`, `event.labels`, `event.file` |
| `mly` | `Burst` | Flat JSON (`*.json`) | `central_time` / `gpstime`, `far`, `ifos`, `channels`, `SNR`, `central_freq`, `bandwidth`, `duration`, `scores`, and source probabilities | Root fields (`gpstime`, `far`, `instruments`, channels), `event.mly`, `event.labels`, `event.file` |
| `cwb` | `Burst` | ASCII (`trigger.txt` / `event.log`) | Key-value trigger summary (`event_time`, `start`, `far`, `ifo`, `event_snr`, `mchirp`, `likelihood`, `nevent`, `duration`, `frequency`, `bandwidth`, `strain`, `code version`, `hoft version`, `CIT`) | Root fields (`gpstime`, `far`, `instruments`, `search`, `nevents`, `likelihood`), `event.burst`, `event.labels`, `event.file` |

### Example

```python
import base64
from pathlib import Path
from event_normalizer.parsers import parse_initial_upload

raw_xml = Path("coinc.xml").read_bytes()

kafka_msg = {
    "type": "upload",
    "gracedb_server": "https://gracedb-playground.ligo.org/api/",
    "graceid": "G4757964",
    "submitter": "sgnl-online-cit-scitoken",
    "pipeline": "SGNL",
    "gracedb_upload": True,
    "labels": ["EARLY_WARNING"],
    "eventFile": ["coinc.xml", base64.b64encode(raw_xml).decode("ascii")],
    "content_id": "0b610b4f-92f9-643b-0e07-0d212502fa56",
    "_producer_timestamp": 1790778302.0,
}

event = parse_initial_upload(kafka_msg)

print(event.uid)
print(event.group)
print(event.pipeline)
print(event.gpstime)
print(event.far)
print(event.file.file_name)
print(event.file.file_size)
print(event.file.file_created_delay)
```

---

# Usage

## Parse a raw event

```python
import json

from event_normalizer import parse_gracedb_event


with open("event.json") as file_handle:
    raw_event = json.load(file_handle)

event = parse_gracedb_event(raw_event)

print(event.uid)
print(event.group)
print(event.pipeline)
print(event.far)
print(event.H1_channel)
```

The parser accepts a GraceDB-like dictionary:

```python
event = parse_gracedb_event(raw_event)
```

A missing or empty `graceid` raises:

```python
ValueError
```

After source fields have been mapped, Pydantic validates the normalized
model, including mandatory fields, field types, and identifier format.

---

## Use the normalized model

```python
event = parse_gracedb_event(raw_event)

if event.coinc is not None:
    print(event.coinc.coinc_snr)

if event.single_H1 is not None:
    print(event.single_H1.single_snr)

if event.labels and event.labels.label_MOCK:
    print("Mock event")
```

The complete field schema is defined in:

```text
event_normalizer/models.py
```

---

## Produce a flat dictionary

Use `to_flat_dict()` when a downstream consumer needs scalar values in a
single-level dictionary.

```python
from event_normalizer.serializers import to_flat_dict

flat_event = to_flat_dict(event)

print(flat_event["uid"])
print(flat_event["far"])
print(flat_event["instruments_H1"])
print(flat_event["H1_channel"])
```

Representative flat keys include:

```text
uid
event_type
far
gpstime
instruments_H1
H1_channel
coinc_snr
burst_duration
mly_central_freq
single_H1_single_snr
label_MOCK
```

---

## Produce JSON

Use `JSONSerializer` when the output must be JSON text.

### Structured JSON

```python
from event_normalizer.serializers import JSONSerializer

structured_json = JSONSerializer.serialize(
    event,
    mode="structured",
)
```

The result preserves nested models:

```json
{
  "uid": "G65309",
  "pipeline": "gstlal",
  "coinc": {
    "coinc_mass": 72.5894737243652,
    "coinc_snr": 9.25419884862828
  }
}
```

### Flat JSON

```python
flat_json = JSONSerializer.serialize(
    event,
    mode="flat",
)
```

The flat mode uses the same representation as `to_flat_dict()`.

### Dictionary output through the serializer

```python
structured_dict = JSONSerializer.to_dict(
    event,
    mode="structured",
)

flat_dict = JSONSerializer.to_dict(
    event,
    mode="flat",
)
```

---

## Reduce a raw event

The reducer is independent from the parser.

It extracts selected paths from a raw event dictionary while preserving the
original nested structure.

```python
from event_normalizer.reducers import reduce_event

reduced_event = reduce_event(
    raw_event,
    paths=[
        "graceid",
        "group",
        "pipeline",
        "far",
        "gpstime",
        "extra_attributes.CoincInspiral.mass",
        "extra_attributes.SingleInspiral.channel",
    ],
)
```

Use `required_fields` to force fields into the result:

```python
reduced_event = reduce_event(
    raw_event,
    paths=[
        "graceid",
        "extra_attributes.CoincInspiral.mass",
    ],
    required_fields=(
        "far",
        "gpstime",
    ),
)
```

For list entries such as `SingleInspiral`, the reducer preserves `ifo` by
default so that detector-specific values remain identifiable.

```python
reduced_event = reduce_event(
    raw_event,
    paths=[
        "extra_attributes.SingleInspiral.channel",
        "extra_attributes.SingleInspiral.snr",
    ],
)
```

---

# Logging

The package uses Python's standard logging system.

Enable colored terminal logging:

```python
import logging

from event_normalizer.logging_utils import configure_logging

configure_logging(logging.DEBUG)
```

The package logger namespace is:

```text
event_normalizer
```

Warnings are emitted for malformed or unsupported structures, including:

- invalid event table types;
- invalid channel entries;
- invalid instruments representation;
- invalid links structure;
- unknown labels.

---

# Development

Run tests:

```bash
python -m pytest
```

Run linting:

```bash
python -m ruff check .
```

Check formatting:

```bash
python -m ruff format --check .
```

Apply available Ruff fixes and formatting:

```bash
python -m ruff check . --fix
python -m ruff format .
```

---

# CI/CD and releases

GitLab CI runs:

```text
test
secret-detection
build
publish
```

The pipeline runs tests, coverage, Ruff checks, GitLab SAST, secret detection,
package build, and package metadata validation.

Versions are derived from Git tags using `hatch-vcs`.

| Tag | Result |
|---|---|
| `v0.1.1` | Stable PyPI release `0.1.1` |
| `v0.2.0rc1` | TestPyPI release candidate `0.2.0rc1` |

Release tags trigger publishing:

```bash
git tag -a v0.1.2 -m "Release v0.1.2"
git push origin v0.1.2
```

PyPI and TestPyPI credentials are stored only as protected and masked GitLab
CI/CD variables. 

---

# License

Apache License 2.0.