diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index 327158b..7476f9a 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -1,7 +1,7 @@ { "plugin_library": { "version": "1.0.0", - "last_updated": "2026-06-24T00:00:00+00:00", + "last_updated": "2026-07-14T00:00:00+00:00", "plugins": [ { "name": "Downsampler", @@ -252,6 +252,17 @@ "last_update": "2026-07-09", "trigger_types_supported": ["scheduler"] }, + { + "name": "Signal Filter", + "path": "influxdata/signal_filter/signal_filter.py", + "description": "Applies streaming digital IIR filters to numeric fields", + "author": "InfluxData", + "docs_file_link": "https://github.com/influxdata/influxdb3_plugins/blob/main/influxdata/signal_filter/README.md", + "required_plugins": [], + "required_libraries": ["numpy", "scipy", "influxdata-plugin-utils>=0.2.0"], + "last_update": "2026-07-14", + "trigger_types_supported": ["data_writes"] + }, { "name": "River Auto Profiler", "path": "influxdata/river_auto_profiler/river_auto_profiler.py", diff --git a/influxdata/signal_filter/README.md b/influxdata/signal_filter/README.md new file mode 100644 index 0000000..a4e8e7b --- /dev/null +++ b/influxdata/signal_filter/README.md @@ -0,0 +1,408 @@ +# Signal Filter Plugin + +⚑ data-write 🏷️ signal-processing, filtering, transformation πŸ”§ InfluxDB 3 Core, InfluxDB 3 Enterprise + +## Description + +The Signal Filter Plugin applies streaming digital IIR filters (Butterworth, +Chebyshev I, Bessel β€” lowpass/highpass/bandpass/bandstop β€” or manually supplied +second-order sections) to numeric time-series fields as they are written to +InfluxDB 3. Filtering is causal and per series: each tag combination gets its own +filter whose delay-line state persists in the trigger cache across WAL commits, so +output over a stream of small writes is identical to filtering the whole signal at +once. It pairs naturally with the +[`signal_generator`](https://github.com/influxdata/influxdb3_plugins/tree/main/influxdata/signal_generator) +plugin but works with any measurement carrying numeric fields. + +- **Streaming, stateful**: per-series delay-line state persists in the trigger + cache, so chunked writes filter identically to a single batch +- **Preset or manual design**: SciPy-designed Butterworth, Chebyshev I, and Bessel + prototypes, or raw second-order sections you supply +- **Automatic sample-rate inference**: infers `fs` per series from the data when + not configured explicitly +- **Multi-field, multi-series**: filters each configured field and each tag + combination independently +- **Flexible output routing**: rename, prefix/suffix, and redirect the filtered + field to another measurement or database + +## Configuration + +Plugin parameters may be specified as key-value pairs in the `--trigger-arguments` +flag (`influxdb3 create trigger`) or in the `trigger_arguments` field of the API. +Values are strings; the plugin coerces them. Alternatively, supply every parameter +from a TOML file via `config_file_path` β€” see [TOML configuration](#toml-configuration). + +> **CLI limitation:** the `sos` argument is a JSON array containing commas, and +> `influxdb3 create trigger --trigger-arguments` splits on every comma, so the value +> is fragmented before it reaches the plugin. Configure `sos` (i.e. `design_type=manual`) +> through the InfluxDB 3 Explorer UI, the `/api/v3/configure/processing_engine_trigger` +> API, or a TOML file β€” not the CLI. Space-separated arguments such as `input_fields` +> and `tag_keys` are unaffected. + +### Plugin metadata + +This plugin includes a JSON metadata schema in its docstring that defines supported +trigger types and configuration parameters. This metadata enables the +[InfluxDB 3 Explorer](https://docs.influxdata.com/influxdb3/explorer/) UI to display +and configure the plugin. + +### Input parameters + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `input_measurement` | string | *(all tables)* | Table to filter. If omitted, every table the trigger fires for is filtered. | +| `input_fields` | string | `value` | Space-separated numeric fields to filter; each is filtered independently. | +| `tag_keys` | string | *(auto)* | Space-separated tag columns that define a series. Defaults to all string-valued columns except `time` and the input fields. Set explicitly if a string *field* would otherwise be mistaken for a tag. | + +### Design parameters + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `design_type` | string | `preset` | `preset` (SciPy-designed IIR) or `manual` (SOS coefficients supplied via `sos`). | +| `prototype` | string | `butter` | Preset prototype: `butter`, `cheby1`, or `bessel`. | +| `order` | integer | `4` | Filter order, 1–12. Band filters yield effective order 2N. | +| `ripple` | float | β€” | Passband ripple in dB, 0.01–80 (0.1–3 typical). Required for `cheby1`; invalid otherwise. | +| `filter_type` | string | `lowpass` | `lowpass`, `highpass`, `bandpass`, or `bandstop`. | +| `fc` | float | β€” | Convenience alias for the single cutoff (Hz) of lowpass/highpass. Invalid for band types or together with the parameter it maps to. | +| `fc1` | float | β€” | Lower cutoff (Hz). Required for highpass and band filters. | +| `fc2` | float | β€” | Upper cutoff (Hz). Required for lowpass and band filters. Band filters require `fc1 < fc2`. | +| `bessel_norm` | string | `phase` | Bessel normalization: `phase`, `delay`, or `mag`. Bessel only. | +| `sos` | JSON string | β€” | Manual second-order sections as JSON `[[b0,b1,b2,a0,a1,a2], ...]`. Required for `design_type` `manual`; invalid otherwise. Rows are normalized by `a0`; unstable filters (any pole magnitude β‰₯ 1) are rejected. | +| `sample_rate` | float | *(inferred)* | Sample rate in Hz for preset design. If omitted, inferred per series from the median inter-sample interval and frozen once enough samples are seen (see [Sample-rate inference](#sample-rate-inference)). | +| `init_from_first_sample` | boolean | `true` | Initialize filter state from the first sample to suppress the startup transient. | + +### Output parameters + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `output_target_database` | string | *(trigger db)* | Database to write filtered output to. | +| `output_measurement` | string | *(source table)* | Measurement to write filtered output to. | +| `output_field` | string | *(source field)* | Base name override for the output field. Only valid when a single input field is configured. | +| `field_prefix` | string | *(empty)* | Prefix for the output field name. | +| `field_suffix` | string | `_filtered` | Suffix for the output field name. | +| `config_file_path` | string | β€” | Path to a TOML file supplying all parameters; mutually exclusive with inline arguments. Relative paths resolve against `PLUGIN_DIR`. | + +The final output field name is `{field_prefix}{output_field or source_field}{field_suffix}` +β€” by default, `value` becomes `value_filtered`. The raw input field is never copied +to the output. + +### TOML configuration + +To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and +reference the file with the `config_file_path` trigger argument (relative paths +resolve against `PLUGIN_DIR`, then `INFLUXDB3_PLUGIN_DIR`, then the parent of +`VIRTUAL_ENV`). The TOML file then supplies **all** parameters β€” it is mutually +exclusive with inline trigger arguments, so passing both is rejected. See +[`signal_filter_config_data_writes.toml`](signal_filter_config_data_writes.toml) +for an annotated template. + +## Data requirements + +- Input fields must be numeric (float or int). Null, boolean, and string values + contribute no samples; NaN/Β±Inf samples are dropped (they would permanently + poison IIR filter state). +- Duplicate timestamps within a commit keep the last occurrence, matching the + database's last-write-wins semantics. +- **Samples must arrive in time order across commits.** The plugin records the + last filtered timestamp per series and drops late/backfilled samples with a + warning β€” replaying old samples through a stateful causal filter would corrupt + the output. To re-filter history, delete the trigger's cache state (or recreate + the trigger) and replay the data in order. + +## Software Requirements + +- **InfluxDB 3 Core/Enterprise**: with the Processing Engine enabled + (`--plugin-dir` configured). +- **Python packages**: `numpy`, `scipy`, and `influxdata-plugin-utils`. + +### Installation steps + +1. Start InfluxDB 3 with the Processing Engine enabled (`--plugin-dir /path/to/plugins`): + + ```bash + influxdb3 serve \ + --node-id node0 \ + --object-store file \ + --data-dir ~/.influxdb3 \ + --plugin-dir ~/.plugins + ``` + +2. Install the Python dependencies into the plugin environment: + + ```bash + influxdb3 install package numpy scipy influxdata-plugin-utils + ``` + + or via the HTTP API: + + ```bash + curl -X POST "http://localhost:8181/api/v3/configure/plugin_environment/install_packages" \ + -H "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"packages": ["numpy", "scipy", "influxdata-plugin-utils"]}' + ``` + +3. Copy `signal_filter.py` into your plugin directory (or upload it with + `POST /api/v3/plugins/files`). + +## Trigger setup + +Create the trigger with `run_async: false` (the plugin's per-series state assumes +one invocation in flight at a time) and `error_behavior: log` or `retry` (state is +saved only after all writes are buffered, so retries re-filter and re-emit the same +points idempotently). Note the lowercase values β€” the live API rejects `Log`. + +### Data write trigger + +```bash +influxdb3 create trigger \ + --database mydb \ + --plugin-filename signal_filter.py \ + --trigger-spec "table:signal" \ + --trigger-arguments 'input_fields=value,filter_type=lowpass,fc=5.0,order=4' \ + signal_lowpass +``` + +or via the HTTP API: + +```bash +curl -X POST "http://localhost:8181/api/v3/configure/processing_engine_trigger" \ + -H "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{ + "db": "mydb", + "plugin_filename": "signal_filter.py", + "trigger_name": "signal_lowpass", + "trigger_specification": "table:signal", + "trigger_settings": {"run_async": false, "error_behavior": "log"}, + "trigger_arguments": { + "input_fields": "value", + "filter_type": "lowpass", + "fc": "5.0", + "order": "4" + }, + "disabled": false + }' +``` + +### Enable the trigger + +```bash +influxdb3 enable trigger --database mydb signal_lowpass +``` + +### Testing without a trigger + +Use the WAL plugin test endpoint. Passing a fixed `cache_name` across calls +exercises cross-commit state continuity: + +```bash +curl -X POST "http://localhost:8181/api/v3/plugin_test/wal" \ + -H "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{ + "filename": "signal_filter.py", + "database": "mydb", + "input_lp": "signal,host=a value=1.0 1717000000000000000\nsignal,host=a value=2.0 1717000000100000000", + "cache_name": "sigfilt_test", + "input_arguments": {"filter_type": "lowpass", "fc": "5.0", "sample_rate": "10.0"} + }' +``` + +Note: test caches expire after 30 minutes by default; production trigger caches +persist indefinitely. + +## Example usage + +### Example 1: Lowpass smoothing (defaults) + +Attenuate high-frequency noise on the `value` field of the `signal` table with a +4th-order Butterworth lowpass at 5 Hz: + +```bash +# Create and enable the trigger +influxdb3 create trigger \ + --database mydb \ + --plugin-filename signal_filter.py \ + --trigger-spec "table:signal" \ + --trigger-arguments 'input_fields=value,filter_type=lowpass,fc=5.0,order=4,sample_rate=100.0' \ + signal_lowpass +influxdb3 enable trigger --database mydb signal_lowpass + +# Write some data, then query the filtered field +influxdb3 query \ + --database mydb \ + "SELECT time, value, value_filtered FROM signal ORDER BY time DESC LIMIT 10" +``` + +**Expected output**: for every written `value`, a `value_filtered` field appears at +the same timestamp, with high-frequency content removed. The raw `value` is left +untouched. + +### Example 2: Bandpass with an explicit sample rate + +Isolate a 1–5 Hz band on a `vibration` field sampled at 50 Hz: + +```bash +influxdb3 create trigger \ + --database mydb \ + --plugin-filename signal_filter.py \ + --trigger-spec "table:sensors" \ + --trigger-arguments 'input_measurement=sensors,input_fields=vibration,filter_type=bandpass,fc1=1.0,fc2=5.0,order=4,sample_rate=50.0' \ + vibration_bandpass +influxdb3 enable trigger --database mydb vibration_bandpass +``` + +**Expected output**: a `vibration_filtered` field carrying only the 1–5 Hz band, +per tag combination present in `sensors`. + +### Example 3: Manual second-order sections + +Apply coefficients you designed elsewhere, routed to a separate measurement: + +```bash +curl -X POST "http://localhost:8181/api/v3/configure/processing_engine_trigger" \ + -H "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{ + "db": "mydb", + "plugin_filename": "signal_filter.py", + "trigger_name": "manual_sos", + "trigger_specification": "table:signal", + "trigger_settings": {"run_async": false, "error_behavior": "log"}, + "trigger_arguments": { + "design_type": "manual", + "sos": "[[0.2, 0.4, 0.2, 1.0, -0.4, 0.2]]", + "output_measurement": "signal_filtered" + }, + "disabled": false + }' +``` + +**Expected output**: filtered points written to `signal_filtered` (not the source +`signal` table). Manual mode needs no sample rate β€” the coefficients are already +digital. + +## Sample-rate inference + +Preset design needs the sample rate. Resolution order per series: + +1. Explicit `sample_rate` argument β€” always wins, so config corrections take + effect immediately (a changed rate re-designs the filter and resets state). +2. Frozen per-series value from a previous successful inference. +3. Inference: `fs = 1e9 / median(inter-sample interval)` over accumulated + timestamps. An estimate is frozen only after β‰₯ 8 inter-sample intervals have + been observed; smaller commits accumulate timestamps across invocations and + are skipped (with an info log, samples not emitted) until the threshold is + met. At typical `signal_generator` rates a single commit clears it + immediately. + +Manual mode needs no sample rate β€” the coefficients are already digital. + +## Write-loop behavior + +Writing the output into the source measurement re-fires this trigger. This is +safe by default: re-fired rows carry only the output field, the input field is +null on them, and null values produce no samples. **However**, if your overrides +resolve the output field to the *same name* as the input field in the same +measurement and database (for example `field_suffix=""` with no `output_field`), +the output feeds the filter again and grows without bound. The plugin logs a +prominent warning in that configuration β€” change `field_suffix`, `output_field`, +`output_measurement`, or `output_target_database` to break the cycle. + +## Code overview + +### Files + +- `signal_filter.py`: The main plugin code β€” config parsing/validation, filter + design (preset + manual), the streaming runtime, per-series state helpers, and + the `process_writes` WAL entry point. +- `signal_filter_config_data_writes.toml`: Annotated example TOML configuration. +- `test_signal_filter.py`: Unit and integration tests (mock-based; no running + engine required). + +### Logging + +Every log line is prefixed with a per-invocation task id: `[] signal_filter: ...`. +Per invocation the plugin logs an info summary β€” tables and series processed, +samples in, points written, and counts of dropped non-finite, out-of-order, and +warm-up-skipped samples. Warnings cover out-of-order drops and the write-loop +hazard; errors cover missing dependencies, invalid configuration, and filter +design failures (for example a cutoff at or above the Nyquist frequency). Logs are +available in the `system.processing_engine_logs` table. + +### Main functions + +#### `process_writes(influxdb3_local, table_batches, args)` + +Entry point for data write triggers. Validates configuration, then for each table +batch: extracts per-(field, series) samples, resolves the sample rate, designs (or +reuses a memoized) filter, applies it with the cached delay-line state, writes the +filtered points, and finally saves the advanced per-series state. + +Key operations: + +1. Guards that `numpy`, `scipy`, and `influxdata-plugin-utils` are installed; logs an install command otherwise +2. Parses and validates trigger arguments (inline, or entirely from a TOML file) +3. Warns on any write-loop hazard configuration +4. Groups rows into per-(field, series) samples, dropping null/non-numeric/non-finite values +5. Resolves the sample rate (explicit β†’ frozen β†’ inferred with warm-up) +6. Designs the IIR filter (memoized by parameters + `fs`) and checks stability +7. Applies the filter using cached `zi` state, writing each point with `write_sync(..., no_sync=True)` +8. Saves the advanced state (delay line, `fs`, coeff hash, last timestamp) to the cache + +## Troubleshooting + +### Common issues + +#### Issue: "required packages are not installed" + +**Cause**: The plugin environment lacks `numpy`, `scipy`, or `influxdata-plugin-utils`. + +**Solution**: Run `influxdb3 install package numpy scipy influxdata-plugin-utils` and re-enable the trigger. + +#### Issue: No output points + +**Cause**: Warm-up (sample-rate inference not yet frozen), a single-sample sparse +series, or a non-numeric/null input field. + +**Solution**: + +- Check the info logs for warm-up skips: with no `sample_rate` argument the + plugin waits for β‰₯ 8 inter-sample intervals per series before filtering. +- A single-sample series with no `sample_rate` cannot infer a rate; set the + argument explicitly for very sparse data. +- Verify the input field is numeric and non-null in the written rows. + +#### Issue: "filter design failed: cutoff ... must be within (0, fs/2)" + +**Cause**: The (possibly inferred) sample rate puts your cutoff at or beyond Nyquist. + +**Solution**: Set `sample_rate` explicitly or lower the cutoff. + +#### Issue: Output has a transient after a server restart + +**Cause**: The cache is cleared on restart, so filters re-initialize and inferred +sample rates re-freeze (the cutoff may shift very slightly if the new estimate +differs). + +**Solution**: This is expected; set `sample_rate` explicitly to pin the design. + +#### Issue: Series explosion / unexpected series + +**Cause**: The automatic tag heuristic treats every string-valued column as a tag, +including string *fields*. + +**Solution**: Set `tag_keys` explicitly to bound the series set. + +### Viewing logs + +```bash +influxdb3 query \ + --database YOUR_DATABASE \ + "SELECT * FROM system.processing_engine_logs WHERE trigger_name = 'signal_lowpass' ORDER BY event_time DESC LIMIT 20" +``` + +## Questions/Comments + +For additional support, see the [Support section](../README.md#support). diff --git a/influxdata/signal_filter/manifest.toml b/influxdata/signal_filter/manifest.toml new file mode 100644 index 0000000..ad91d9a --- /dev/null +++ b/influxdata/signal_filter/manifest.toml @@ -0,0 +1,19 @@ +manifest_schema_version = "1.3" + +[plugin] +name = "signal_filter" +version = "0.1.0" +description = "Applies streaming digital IIR filters to numeric fields" +triggers = ["process_writes"] +homepage = "https://www.influxdata.com/" +repository = "https://github.com/influxdata/influxdb3_plugins/tree/main/influxdata/signal_filter" +documentation = "https://github.com/influxdata/influxdb3_plugins/blob/main/influxdata/signal_filter/README.md" +exclude = [ + ".git/", ".venv/", "__pycache__/", "*.pyc", "tests/**", + ".claude/", ".vscode/", ".idea/", ".DS_Store", + "*.py", "!signal_filter.py", +] + +[dependencies] +database_version = ">=3.8.2" +python = ["numpy", "scipy", "influxdata-plugin-utils>=0.2.0"] diff --git a/influxdata/signal_filter/requirements.txt b/influxdata/signal_filter/requirements.txt new file mode 100644 index 0000000..b17103e --- /dev/null +++ b/influxdata/signal_filter/requirements.txt @@ -0,0 +1,3 @@ +numpy +scipy +influxdata-plugin-utils>=0.2.0 diff --git a/influxdata/signal_filter/signal_filter.py b/influxdata/signal_filter/signal_filter.py new file mode 100644 index 0000000..1297201 --- /dev/null +++ b/influxdata/signal_filter/signal_filter.py @@ -0,0 +1,819 @@ +""" +{ + "plugin_type": ["onwrite"], + "onwrite_args_config": [ + { + "name": "input_measurement", + "example": "signal", + "description": "Table to filter. If omitted, every table the trigger fires for is filtered.", + "required": false + }, + { + "name": "input_fields", + "example": "value", + "description": "Space-separated numeric fields to filter; each is filtered independently. Defaults to 'value'.", + "required": false + }, + { + "name": "tag_keys", + "example": "host region", + "description": "Space-separated tag columns that define a series. Defaults to all string-valued columns except 'time' and the input fields.", + "required": false + }, + { + "name": "design_type", + "example": "preset", + "description": "'preset' (SciPy-designed IIR) or 'manual' (SOS coefficients supplied via 'sos'). Defaults to 'preset'.", + "required": false + }, + { + "name": "prototype", + "example": "butter", + "description": "Preset prototype: 'butter', 'cheby1', or 'bessel'. Defaults to 'butter'.", + "required": false + }, + { + "name": "order", + "example": "4", + "description": "Filter order, 1-12. Band filters yield effective order 2N. Defaults to 4.", + "required": false + }, + { + "name": "ripple", + "example": "1.0", + "description": "Passband ripple in dB, 0.01-80 (0.1-3 typical). Required for 'cheby1'; invalid otherwise.", + "required": false + }, + { + "name": "filter_type", + "example": "lowpass", + "description": "'lowpass', 'highpass', 'bandpass', or 'bandstop'. Defaults to 'lowpass'.", + "required": false + }, + { + "name": "fc", + "example": "5.0", + "description": "Convenience alias for the single cutoff (Hz) of lowpass/highpass. Invalid for band types or together with the parameter it maps to.", + "required": false + }, + { + "name": "fc1", + "example": "1.0", + "description": "Lower cutoff (Hz). Required for highpass and band filters.", + "required": false + }, + { + "name": "fc2", + "example": "5.0", + "description": "Upper cutoff (Hz). Required for lowpass and band filters.", + "required": false + }, + { + "name": "bessel_norm", + "example": "phase", + "description": "Bessel normalization: 'phase', 'delay', or 'mag'. Defaults to 'phase'. Bessel only.", + "required": false + }, + { + "name": "sos", + "example": "[[0.1, 0.2, 0.1, 1.0, -0.5, 0.2]]", + "description": "Manual second-order sections as JSON [[b0,b1,b2,a0,a1,a2], ...]. Required for design_type 'manual'; invalid otherwise.", + "required": false + }, + { + "name": "sample_rate", + "example": "10.0", + "description": "Sample rate in Hz for preset design. If omitted, inferred per series from median inter-sample interval and frozen once enough samples are seen.", + "required": false + }, + { + "name": "init_from_first_sample", + "example": "true", + "description": "Initialize filter state from the first sample to suppress the startup transient. Defaults to true.", + "required": false + }, + { + "name": "output_target_database", + "example": "filtered_db", + "description": "Database to write filtered output to. Defaults to the trigger's database.", + "required": false + }, + { + "name": "output_measurement", + "example": "signal_filtered", + "description": "Measurement to write filtered output to. Defaults to the source measurement.", + "required": false + }, + { + "name": "output_field", + "example": "smoothed", + "description": "Base name override for the output field. Only valid when a single input field is configured.", + "required": false + }, + { + "name": "field_prefix", + "example": "flt_", + "description": "Prefix for the output field name. Defaults to ''.", + "required": false + }, + { + "name": "field_suffix", + "example": "_filtered", + "description": "Suffix for the output field name. Defaults to '_filtered'.", + "required": false + }, + { + "name": "config_file_path", + "example": "config.toml", + "description": "Path to a TOML file supplying all parameters; mutually exclusive with inline arguments. Relative paths resolve against PLUGIN_DIR.", + "required": false + } + ] +} +""" + +import hashlib +import json +import math +import tomllib +import uuid +from dataclasses import dataclass + +try: + import numpy as np + from scipy import signal as sp_signal + + _IMPORT_ERROR = None +except ImportError as exc: # engine without numpy/scipy installed + np = None + sp_signal = None + _IMPORT_ERROR = str(exc) + +try: + # Shared helpers published by InfluxData; see influxdata-plugin-utils on PyPI. + from influxdata_plugin_utils.config import load_plugin_config + from influxdata_plugin_utils.parsing import ( + parse_bool, + parse_delimited_list, + parse_int, + ) + from influxdata_plugin_utils.write import build_line, write_data + + _UTILS_IMPORT_ERROR = None +except ImportError as exc: # engine without influxdata-plugin-utils installed + load_plugin_config = None + parse_bool = parse_delimited_list = parse_int = None + build_line = write_data = None + _UTILS_IMPORT_ERROR = str(exc) + +try: + from influxdb3_pe import LineBuilder # type: ignore +except ImportError: # the processing engine injects LineBuilder at runtime + pass + + +# --------------------------------------------------------------------------- +# Registries and defaults +# --------------------------------------------------------------------------- + +FILTER_TYPES = ("lowpass", "highpass", "bandpass", "bandstop") +BESSEL_NORMS = ("phase", "delay", "mag") +DESIGN_TYPES = ("preset", "manual") + +WARMUP_MIN_INTERVALS = 8 +WARMUP_MAX_TIMES = 64 +CACHE_KEY_FMT = "signal_filter:{table}:{field}:{series_hash}" + +DEFAULT_INPUT_FIELDS = ("value",) +DEFAULT_FIELD_SUFFIX = "_filtered" + +_DESIGN_CACHE = {} # (design key) -> (sos ndarray, coeff_hash) + + +class ConfigError(Exception): + """Raised when trigger arguments fail validation.""" + + +# --------------------------------------------------------------------------- +# Config parsing + validation +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Config: + input_measurement: str | None + input_fields: tuple + tag_keys: tuple | None + design_type: str + prototype: str + order: int + ripple: float | None + filter_type: str + fc1: float | None + fc2: float | None + bessel_norm: str + sos: tuple | None + sample_rate: float | None + init_from_first_sample: bool + output_target_database: str | None + output_measurement: str | None + output_field: str | None + field_prefix: str + field_suffix: str + + +def _to_float(name, value): + try: + result = float(value) + except (TypeError, ValueError): + raise ConfigError(f"'{name}' must be a number, got {value!r}") from None + if not math.isfinite(result): + raise ConfigError(f"'{name}' must be finite, got {value!r}") + return result + + +def _to_int(name, value): + try: + return parse_int(value) + except ValueError: + raise ConfigError(f"'{name}' must be an integer, got {value!r}") from None + + +def _to_bool(name, value): + try: + return parse_bool(value) + except ValueError: + raise ConfigError(f"'{name}' must be a boolean, got {value!r}") from None + + +def _parse_manual_sos(raw): + if isinstance(raw, str): + try: + raw = json.loads(raw) + except json.JSONDecodeError as exc: + raise ConfigError(f"'sos' is not valid JSON: {exc}") from None + if not isinstance(raw, list) or not raw: + raise ConfigError("'sos' must be a non-empty JSON array of sections") + normalized = [] + for i, section in enumerate(raw): + if not isinstance(section, list) or len(section) != 6: + raise ConfigError(f"'sos' section {i} must have 6 coefficients [b0,b1,b2,a0,a1,a2]") + try: + coeffs = [float(c) for c in section] + except (TypeError, ValueError): + raise ConfigError(f"'sos' section {i} contains a non-numeric coefficient") from None + if any(not math.isfinite(c) for c in coeffs): + raise ConfigError(f"'sos' section {i} contains a non-finite coefficient") + a0 = coeffs[3] + if a0 == 0.0: + raise ConfigError(f"'sos' section {i} has a0 == 0") + normalized.append(tuple(c / a0 for c in coeffs)) + return tuple(normalized) + + +def _resolve_cutoffs(merged): + """Apply the fc alias and per-band presence rules; returns (fc1, fc2).""" + filter_type = merged.get("filter_type", "lowpass") + fc = merged.get("fc") + fc1 = merged.get("fc1") + fc2 = merged.get("fc2") + + if fc is not None: + if filter_type in ("bandpass", "bandstop"): + raise ConfigError("'fc' is not valid for band filters; use 'fc1' and 'fc2'") + if filter_type == "lowpass": + if fc2 is not None: + raise ConfigError("'fc' and 'fc2' are both set; use one") + fc2 = fc + else: # highpass + if fc1 is not None: + raise ConfigError("'fc' and 'fc1' are both set; use one") + fc1 = fc + + fc1 = _to_float("fc1", fc1) if fc1 is not None else None + fc2 = _to_float("fc2", fc2) if fc2 is not None else None + for name, value in (("fc1", fc1), ("fc2", fc2)): + if value is not None and value <= 0: + raise ConfigError(f"'{name}' must be > 0, got {value}") + + if filter_type == "lowpass": + if fc2 is None: + raise ConfigError("lowpass requires a cutoff: set 'fc2' (or 'fc')") + if fc1 is not None: + raise ConfigError("'fc1' is not used for lowpass") + elif filter_type == "highpass": + if fc1 is None: + raise ConfigError("highpass requires a cutoff: set 'fc1' (or 'fc')") + if fc2 is not None: + raise ConfigError("'fc2' is not used for highpass") + else: # bandpass / bandstop + if fc1 is None or fc2 is None: + raise ConfigError(f"{filter_type} requires both 'fc1' and 'fc2'") + if not fc1 < fc2: + raise ConfigError(f"'fc1' must be < 'fc2', got {fc1} >= {fc2}") + return fc1, fc2 + + +def parse_config(args): + """Validate trigger args, sourced either inline or entirely from a TOML file. + + Following the other influxdata plugins, ``config_file_path`` is mutually + exclusive with inline arguments: when it is set the TOML file supplies every + parameter, which keeps precedence explicit rather than silently merging. + """ + args = dict(args or {}) + has_toml = bool(args.get("config_file_path")) + if has_toml: + extra = sorted(k for k in args if k != "config_file_path" and args[k] is not None) + if extra: + raise ConfigError( + "'config_file_path' is mutually exclusive with inline trigger " + f"arguments; also set: {', '.join(extra)}" + ) + # load_plugin_config resolves the TOML path (PLUGIN_DIR -> INFLUXDB3_PLUGIN_DIR + # -> parent of VIRTUAL_ENV) and layers the sources; "toml" vs "args" keeps the + # two mutually exclusive rather than silently merging. + try: + settings = load_plugin_config(args, source="toml" if has_toml else "args") + except FileNotFoundError: + raise ConfigError(f"config file not found: {args['config_file_path']}") from None + except tomllib.TOMLDecodeError as exc: + raise ConfigError(f"config file is not valid TOML: {exc}") from None + except ValueError as exc: # plugin directory could not be resolved + raise ConfigError(str(exc)) from None + merged = {k.lower(): v for k, v in settings.as_dict().items()} + + design_type = str(merged.get("design_type", "preset")) + if design_type not in DESIGN_TYPES: + raise ConfigError(f"'design_type' must be one of {DESIGN_TYPES}, got {design_type!r}") + + input_fields = tuple(parse_delimited_list(merged.get("input_fields", "value"))) or DEFAULT_INPUT_FIELDS + tag_keys_raw = merged.get("tag_keys") + tag_keys = tuple(parse_delimited_list(tag_keys_raw)) if tag_keys_raw else None + + output_field = merged.get("output_field") + if output_field is not None and len(input_fields) != 1: + raise ConfigError("'output_field' is only valid with a single input field") + + filter_type = str(merged.get("filter_type", "lowpass")) + if filter_type not in FILTER_TYPES: + raise ConfigError(f"'filter_type' must be one of {FILTER_TYPES}, got {filter_type!r}") + + prototype = str(merged.get("prototype", "butter")) + order = 4 + ripple = None + bessel_norm = str(merged.get("bessel_norm", "phase")) + fc1 = fc2 = None + sos = None + sample_rate = None + + if design_type == "manual": + if merged.get("sos") is None: + raise ConfigError("design_type 'manual' requires 'sos'") + sos = _parse_manual_sos(merged["sos"]) + else: + if merged.get("sos") is not None: + raise ConfigError("'sos' is only valid with design_type 'manual'") + if prototype not in PRESET_PROTOTYPES: + raise ConfigError( + f"'prototype' must be one of {tuple(PRESET_PROTOTYPES)}, got {prototype!r}" + ) + order = _to_int("order", merged.get("order", 4)) + if not 1 <= order <= 12: + raise ConfigError(f"'order' must be 1-12, got {order}") + if bessel_norm not in BESSEL_NORMS: + raise ConfigError(f"'bessel_norm' must be one of {BESSEL_NORMS}, got {bessel_norm!r}") + if prototype == "cheby1": + if merged.get("ripple") is None: + raise ConfigError("'ripple' is required for prototype 'cheby1'") + ripple = _to_float("ripple", merged["ripple"]) + if not 0.01 <= ripple <= 80: + raise ConfigError(f"'ripple' must be 0.01-80 dB, got {ripple}") + elif merged.get("ripple") is not None: + raise ConfigError("'ripple' is only valid for prototype 'cheby1'") + fc1, fc2 = _resolve_cutoffs(merged) + if merged.get("sample_rate") is not None: + sample_rate = _to_float("sample_rate", merged["sample_rate"]) + if sample_rate <= 0: + raise ConfigError(f"'sample_rate' must be > 0, got {sample_rate}") + + return Config( + input_measurement=str(merged["input_measurement"]) if merged.get("input_measurement") else None, + input_fields=input_fields, + tag_keys=tag_keys, + design_type=design_type, + prototype=prototype, + order=order, + ripple=ripple, + filter_type=filter_type, + fc1=fc1, + fc2=fc2, + bessel_norm=bessel_norm, + sos=sos, + sample_rate=sample_rate, + init_from_first_sample=_to_bool( + "init_from_first_sample", merged.get("init_from_first_sample", True) + ), + output_target_database=str(merged["output_target_database"]) + if merged.get("output_target_database") + else None, + output_measurement=str(merged["output_measurement"]) + if merged.get("output_measurement") + else None, + output_field=str(output_field) if output_field else None, + field_prefix=str(merged.get("field_prefix", "")), + field_suffix=str(merged.get("field_suffix", DEFAULT_FIELD_SUFFIX)), + ) + + +def resolve_output_field(cfg, input_field): + return f"{cfg.field_prefix}{cfg.output_field or input_field}{cfg.field_suffix}" + + +def loop_hazard_fields(cfg): + """Input fields whose resolved output name loops back onto themselves. + + The natural null-skip protection only works while the output field name + differs from the input field name; a same-name write into the same + measurement and database re-fires the trigger with a non-null input field. + """ + if cfg.output_target_database is not None: + return [] + if ( + cfg.output_measurement is not None + and cfg.input_measurement is not None + and cfg.output_measurement != cfg.input_measurement + ): + return [] + return [f for f in cfg.input_fields if resolve_output_field(cfg, f) == f] + + +# --------------------------------------------------------------------------- +# Sample-rate inference +# --------------------------------------------------------------------------- + + +def infer_sample_rate(times_ns): + """fs = 1e9 / median(diff(unique sorted times)); None when < 2 distinct.""" + distinct = sorted(set(times_ns)) + if len(distinct) < 2: + return None + diffs = np.diff(np.asarray(distinct, dtype=np.float64)) + return 1e9 / float(np.median(diffs)) + + +def merge_warmup_times(state, times_ns): + """Accumulate distinct timestamps across commits, keeping the most recent.""" + previous = state.get("warmup_times", []) if state else [] + merged = sorted(set(previous) | set(times_ns)) + return merged[-WARMUP_MAX_TIMES:] + + +# --------------------------------------------------------------------------- +# Filter design +# --------------------------------------------------------------------------- + + +def _design_butter(cfg, wn, fs): + return sp_signal.butter(cfg.order, wn, btype=cfg.filter_type, output="sos", fs=fs) + + +def _design_cheby1(cfg, wn, fs): + return sp_signal.cheby1( + cfg.order, cfg.ripple, wn, btype=cfg.filter_type, output="sos", fs=fs + ) + + +def _design_bessel(cfg, wn, fs): + return sp_signal.bessel( + cfg.order, wn, btype=cfg.filter_type, norm=cfg.bessel_norm, output="sos", fs=fs + ) + + +PRESET_PROTOTYPES = { + "butter": _design_butter, + "cheby1": _design_cheby1, + "bessel": _design_bessel, +} + + +def check_stability(sos): + _, poles, _ = sp_signal.sos2zpk(np.asarray(sos, dtype=np.float64)) + if poles.size and float(np.max(np.abs(poles))) >= 1.0: + raise ValueError( + f"unstable filter: pole magnitude {float(np.max(np.abs(poles))):.6f} >= 1" + ) + + +def _design_key(cfg, fs): + if cfg.design_type == "manual": + return ("manual", json.dumps(cfg.sos)) + params = { + "prototype": cfg.prototype, + "order": cfg.order, + "ripple": cfg.ripple, + "filter_type": cfg.filter_type, + "fc1": cfg.fc1, + "fc2": cfg.fc2, + "bessel_norm": cfg.bessel_norm, + } + return ("preset", json.dumps(params, sort_keys=True), repr(float(fs))) + + +def design_iir(cfg, fs): + if cfg.design_type == "manual": + sos = np.asarray(cfg.sos, dtype=np.float64) + check_stability(sos) + return sos + nyquist = fs / 2.0 + for name, cutoff in (("fc1", cfg.fc1), ("fc2", cfg.fc2)): + if cutoff is not None and not 0 < cutoff < nyquist: + raise ValueError( + f"cutoff {name}={cutoff} Hz must be within (0, fs/2) = (0, {nyquist}) Hz" + ) + if cfg.filter_type == "lowpass": + wn = cfg.fc2 + elif cfg.filter_type == "highpass": + wn = cfg.fc1 + else: + wn = [cfg.fc1, cfg.fc2] + sos = PRESET_PROTOTYPES[cfg.prototype](cfg, wn, float(fs)) + check_stability(sos) + return sos + + +FILTER_FAMILIES = {"iir": design_iir} # Seam 2: add "fir" in a later PR + + +def design_filter(cfg, fs): + """Design (or fetch memoized) coefficients; returns (sos, coeff_hash).""" + key = _design_key(cfg, fs) + cached = _DESIGN_CACHE.get(key) + if cached is not None: + return cached + sos = FILTER_FAMILIES["iir"](cfg, fs) + coeff_hash = hashlib.sha256(repr(key).encode()).hexdigest() + _DESIGN_CACHE[key] = (sos, coeff_hash) + return sos, coeff_hash + + +# --------------------------------------------------------------------------- +# Streaming runtime +# --------------------------------------------------------------------------- + + +def init_zi(sos, first_value, init_from_first_sample): + zi_unit = sp_signal.sosfilt_zi(sos) + if init_from_first_sample: + return zi_unit * float(first_value) + return np.zeros_like(zi_unit) + + +def apply_filter(sos, values, zi): + x = np.asarray(values, dtype=np.float64) + filtered, zf = sp_signal.sosfilt(sos, x, zi=zi) + return filtered, zf + + +# --------------------------------------------------------------------------- +# Per-series state +# --------------------------------------------------------------------------- + + +def series_hash(tag_items): + canonical = ",".join(f"{k}={v}" for k, v in tag_items) + return hashlib.sha256(canonical.encode()).hexdigest() + + +def state_key(table, field, tag_items): + return CACHE_KEY_FMT.format(table=table, field=field, series_hash=series_hash(tag_items)) + + +# --------------------------------------------------------------------------- +# Row/series extraction +# --------------------------------------------------------------------------- + + +def extract_series(rows, input_fields, tag_keys): + """Group rows into per-(field, series) samples. + + Applies the input hygiene rules: null and non-numeric values contribute no + sample, non-finite values are dropped (they would poison IIR state), and + duplicate timestamps keep the last occurrence in row order to match the + database's last-write-wins semantics. + """ + groups = {} + dropped_nonfinite = 0 + input_field_set = set(input_fields) + for row in rows: + time_ns = row.get("time") + if time_ns is None: + continue + if tag_keys is not None: + tag_items = tuple( + sorted((k, str(row[k])) for k in tag_keys if row.get(k) is not None) + ) + else: + tag_items = tuple( + sorted( + (k, v) + for k, v in row.items() + if k != "time" and k not in input_field_set and isinstance(v, str) + ) + ) + for field in input_fields: + value = row.get(field) + if value is None or isinstance(value, bool) or not isinstance(value, (int, float)): + continue + if not math.isfinite(value): + dropped_nonfinite += 1 + continue + groups.setdefault((field, tag_items), []).append((time_ns, float(value))) + + deduped = {} + for key, samples in groups.items(): + by_time = {} + for time_ns, value in samples: # later rows overwrite earlier ones + by_time[time_ns] = value + deduped[key] = sorted(by_time.items()) + return deduped, dropped_nonfinite + + +# --------------------------------------------------------------------------- +# WAL entry point (thin adapter over the entry-agnostic core above; Seam 1) +# --------------------------------------------------------------------------- + + +def _batch_parts(batch): + """Return (table_name, rows) for a WAL batch. + + The live engine (verified on 3.10.2) passes plain dicts; the influxdb3_pe + reference documents a TableBatch object. Accept both. + """ + if isinstance(batch, dict): + return batch["table_name"], batch["rows"] + return batch.table_name, batch.rows + + +def process_writes(influxdb3_local, table_batches: list, args: dict | None = None): + task_id = str(uuid.uuid4()) + if np is None or sp_signal is None or load_plugin_config is None: + influxdb3_local.error( + f"[{task_id}] signal_filter: required packages are not installed in the " + f"plugin environment (import error: {_IMPORT_ERROR or _UTILS_IMPORT_ERROR}). " + "Run: influxdb3 install package numpy scipy influxdata-plugin-utils" + ) + return + + try: + cfg = parse_config(dict(args or {})) + except ConfigError as exc: + influxdb3_local.error(f"[{task_id}] signal_filter: invalid configuration: {exc}") + return + + for field in loop_hazard_fields(cfg): + influxdb3_local.warn( + f"[{task_id}] signal_filter: output field for '{field}' resolves to the same " + "name in the same measurement and database; this feeds the filter its own " + "output (unbounded write loop). Set field_suffix/output_field/" + "output_measurement to break the cycle." + ) + + stats = { + "tables": 0, + "series": 0, + "samples_in": 0, + "points_written": 0, + "dropped_nonfinite": 0, + "dropped_stale": 0, + "warmup_skipped": 0, + } + + for batch in table_batches: + table, rows = _batch_parts(batch) + if cfg.input_measurement is not None and table != cfg.input_measurement: + continue + groups, dropped_nonfinite = extract_series(rows, cfg.input_fields, cfg.tag_keys) + stats["dropped_nonfinite"] += dropped_nonfinite + if not groups: + continue + stats["tables"] += 1 + for (field, tag_items), samples in sorted(groups.items()): + stats["series"] += 1 + stats["samples_in"] += len(samples) + _process_series( + influxdb3_local, cfg, table, field, tag_items, samples, stats, task_id + ) + + influxdb3_local.info( + f"[{task_id}] " + + "signal_filter: {tables} table(s), {series} series: {samples_in} samples in, " + "{points_written} points written, dropped {dropped_nonfinite} non-finite, " + "{dropped_stale} out-of-order, {warmup_skipped} in warm-up".format(**stats) + ) + + +def _process_series(influxdb3_local, cfg, table, field, tag_items, samples, stats, task_id): + cache = influxdb3_local.cache + key = state_key(table, field, tag_items) + state = cache.get(key) + + last_time_ns = state.get("last_time_ns") if state else None + if last_time_ns is not None: + fresh = [(t, v) for t, v in samples if t > last_time_ns] + stale = len(samples) - len(fresh) + if stale: + stats["dropped_stale"] += stale + influxdb3_local.warn( + f"[{task_id}] signal_filter: {table}.{field}: dropped {stale} out-of-order " + "sample(s) at or before the last processed timestamp (backfill through a " + "stateful causal filter would corrupt output)" + ) + samples = fresh + if not samples: + return + + if cfg.design_type == "manual": + fs = None + else: + fs = cfg.sample_rate + if fs is None and state: + fs = state.get("fs") + if fs is None: + merged_times = merge_warmup_times(state, [t for t, _ in samples]) + if len(merged_times) - 1 < WARMUP_MIN_INTERVALS: + new_state = dict(state or {}) + new_state["warmup_times"] = merged_times + cache.put(key, new_state) + stats["warmup_skipped"] += len(samples) + influxdb3_local.info( + f"[{task_id}] signal_filter: {table}.{field}: inferring sample rate " + f"({len(merged_times) - 1}/{WARMUP_MIN_INTERVALS} intervals seen); " + "batch skipped" + ) + return + fs = infer_sample_rate(merged_times) + + try: + sos, coeff_hash = design_filter(cfg, fs) + except ValueError as exc: + influxdb3_local.error( + f"[{task_id}] signal_filter: {table}.{field}: filter design failed: {exc}" + ) + return + + zi = None + if state and state.get("coeff_hash") == coeff_hash and state.get("zi") is not None: + zi = np.asarray(state["zi"], dtype=np.float64) + if zi.shape != (sos.shape[0], 2): # corrupt/stale cache entry + influxdb3_local.warn( + f"[{task_id}] signal_filter: {table}.{field}: cached filter state has shape " + f"{zi.shape}, expected {(sos.shape[0], 2)}; re-initializing" + ) + zi = None + if zi is None: + zi = init_zi(sos, samples[0][1], cfg.init_from_first_sample) + + filtered, zf = apply_filter(sos, [v for _, v in samples], zi) + + out_field = resolve_output_field(cfg, field) + out_measurement = cfg.output_measurement or table + tags = dict(tag_items) + builders = [ + build_line( + LineBuilder, + out_measurement, + tags=tags, + fields={out_field: float(value)}, + time_ns=time_ns, + ) + for (time_ns, _), value in zip(samples, filtered) + ] + + # Write the whole series in one batch before saving state: a failed write + # leaves state unadvanced, so an engine retry re-filters the same samples + # and re-emits identical points (same timestamps overwrite). retries=0 keeps + # the engine's error_behavior as the single retry authority. no_sync=True + # avoids deadlocking the ingest pipeline from inside a synchronous WAL + # trigger (the flush cannot complete until this trigger returns). database + # routes to the trigger's own database when output_target_database is None. + write_data( + influxdb3_local, + builders, + batch=True, + retries=0, + no_sync=True, + database=cfg.output_target_database, + ) + stats["points_written"] += len(samples) + + cache.put( + key, + { + "fs": fs, + "coeff_hash": coeff_hash, + "zi": zf.tolist(), + "last_time_ns": samples[-1][0], + "warmup_times": [], + }, + ) diff --git a/influxdata/signal_filter/signal_filter_config_data_writes.toml b/influxdata/signal_filter/signal_filter_config_data_writes.toml new file mode 100644 index 0000000..e2e42cd --- /dev/null +++ b/influxdata/signal_filter/signal_filter_config_data_writes.toml @@ -0,0 +1,74 @@ +# Signal Filter Plugin Data Write Configuration Template +# Copy this file to your PLUGIN_DIR (e.g., /plugins) and reference it with +# --trigger-arguments config_file_path= +# Example: If the file is at /plugins/configs/signal_filter_config_data_writes.toml, use +# config_file_path=configs/signal_filter_config_data_writes.toml +# This file supplies all parameters; it is mutually exclusive with inline +# trigger arguments (passing both is rejected). + +########## Required Parameters ########## +# Filter type: 'lowpass', 'highpass', 'bandpass', or 'bandstop'. +filter_type = "lowpass" # e.g., "lowpass", "bandpass" + +# Cutoff frequency in Hz. lowpass -> fc2 (or fc); highpass -> fc1 (or fc); +# band filters -> fc1 and fc2 with fc1 < fc2. Required for preset design. +fc2 = 5.0 # e.g., 5.0 +#fc1 = 1.0 # e.g., 1.0 (lower cutoff; required for highpass and band filters) + +########## Optional Parameters ########## +# Table to filter. Omit to filter every table the trigger fires for. +#input_measurement = "signal" # e.g., "signal", "sensors" + +# Space-separated numeric fields to filter; each is filtered independently. +input_fields = "value" # e.g., "value", "temp humidity" + +# Space-separated tag columns defining a series. Omit to auto-detect +# (all string-valued columns except 'time' and the input fields). +#tag_keys = "host region" # e.g., "host region" + +# 'preset' (SciPy-designed IIR) or 'manual' (supply 'sos' below). +design_type = "preset" # e.g., "preset", "manual" + +# Preset prototype: 'butter', 'cheby1', or 'bessel'. +prototype = "butter" # e.g., "butter", "cheby1", "bessel" + +# Filter order, 1-12. Band filters yield effective order 2N. +order = 4 # e.g., 4 + +# Passband ripple in dB (0.01-80). Required for 'cheby1' only. +#ripple = 1.0 # e.g., 1.0 + +# Bessel normalization: 'phase', 'delay', or 'mag'. Bessel only. +#bessel_norm = "phase" # e.g., "phase", "delay", "mag" + +# Manual second-order sections, JSON-style nested arrays [[b0,b1,b2,a0,a1,a2], ...]. +# Required when design_type = "manual". +#sos = "[[0.2, 0.4, 0.2, 1.0, -0.4, 0.2]]" # e.g., "[[0.2,0.4,0.2,1.0,-0.4,0.2]]" + +# Sample rate in Hz. Omit to infer per series from the data (frozen after +# 8 observed inter-sample intervals). +sample_rate = 10.0 # e.g., 10.0, 100.0 + +# Initialize filter state from the first sample to suppress the startup transient. +init_from_first_sample = true # e.g., true, false + +# Database for the filtered output. Omit for the trigger's database. +#output_target_database = "filtered_db" # e.g., "filtered_db" + +# Measurement for the filtered output. Omit for the source measurement. +#output_measurement = "signal_filtered" # e.g., "signal_filtered" + +# Base name override for the output field (single input field only). +#output_field = "smoothed" # e.g., "smoothed" + +# Output field name = field_prefix + (output_field or source field) + field_suffix. +#field_prefix = "" # e.g., "flt_" +field_suffix = "_filtered" # e.g., "_filtered" + +###### Example: Create Trigger Using This Config ###### +# influxdb3 create trigger \ +# --database your_database_name \ +# --plugin-filename signal_filter.py \ +# --trigger-spec "table:signal" \ +# --trigger-arguments config_file_path=signal_filter_config_data_writes.toml \ +# your_trigger_name diff --git a/influxdata/signal_filter/test_signal_filter.py b/influxdata/signal_filter/test_signal_filter.py new file mode 100644 index 0000000..cc792fd --- /dev/null +++ b/influxdata/signal_filter/test_signal_filter.py @@ -0,0 +1,788 @@ +"""Unit and integration tests for the signal_filter plugin. + +Mirrors the mock-based approach used across this repo: fake influxdb3_local, +cache, LineBuilder, and TableBatch; no running engine required. +""" + +import json +import math +import os +import sys +from collections import namedtuple + +import numpy as np +import pytest +from scipy import signal as sp_signal + +sys.path.insert(0, os.path.dirname(__file__)) +import signal_filter as sf + +AttrTableBatch = namedtuple("AttrTableBatch", ["table_name", "rows"]) + + +def TableBatch(table_name, rows): + """Live-engine WAL batch shape (verified on 3.10.2): a plain dict.""" + return {"table_name": table_name, "rows": rows} + + +# --------------------------------------------------------------------------- +# Fakes +# --------------------------------------------------------------------------- + + +class FakeCache: + def __init__(self): + self.store = {} + + def get(self, key, default=None, use_global=None): + return self.store.get(key, default) + + def put(self, key, value, ttl=None, use_global=None): + self.store[key] = value + + def delete(self, key, use_global=None): + return self.store.pop(key, None) is not None + + +def _fmt_float(value): + # Mirror the real LineBuilder.float64_field rendering. + return f"{int(value)}.0" if value % 1 == 0 else str(value) + + +class FakeLineBuilder: + def __init__(self, measurement): + self.measurement = measurement + self.tags = [] + self.fields = {} + self.timestamp = None + + def tag(self, key, value): + self.tags.append((key, value)) + return self + + def float64_field(self, key, value): + self.fields[key] = value + return self + + def time_ns(self, timestamp_ns): + self.timestamp = timestamp_ns + return self + + def build(self): + line = self.measurement + if self.tags: + line += "," + ",".join(f"{k}={v}" for k, v in self.tags) + line += " " + ",".join(f"{k}={_fmt_float(v)}" for k, v in self.fields.items()) + if self.timestamp is not None: + line += f" {self.timestamp}" + return line + + +Record = namedtuple("Record", ["measurement", "tags", "fields", "timestamp"]) + + +def _parse_lp(line): + """Parse one line-protocol record (sufficient for this plugin's output).""" + head, fields_str, ts = line.rsplit(" ", 2) + parts = head.split(",") + tags = dict(kv.split("=", 1) for kv in parts[1:]) + fields = {k: float(v) for k, v in (kv.split("=", 1) for kv in fields_str.split(","))} + return Record(parts[0], tags, fields, int(ts)) + + +class FakeLocal: + def __init__(self, cache=None): + self.cache = cache if cache is not None else FakeCache() + self.infos = [] + self.warns = [] + self.errors = [] + self.writes = [] # (db_name | None, Record) per emitted point + self.fail_writes = False + + def _record_batch(self, db_name, batch): + # The plugin hands a _BatchLines; the engine calls build(). Exercise + # that path, then expand back to one Record per line for assertions. + for lp in batch.build().split("\n"): + self.writes.append((db_name, _parse_lp(lp))) + + def info(self, *args): + self.infos.append(" ".join(str(a) for a in args)) + + def warn(self, *args): + self.warns.append(" ".join(str(a) for a in args)) + + def error(self, *args): + self.errors.append(" ".join(str(a) for a in args)) + + def write_sync(self, batch, no_sync=False): + if self.fail_writes: + raise RuntimeError("simulated write failure") + self._record_batch(None, batch) + + def write_sync_to_db(self, db_name, batch, no_sync=False): + if self.fail_writes: + raise RuntimeError("simulated write failure") + self._record_batch(db_name, batch) + + +@pytest.fixture(autouse=True) +def _plugin_env(monkeypatch): + monkeypatch.setattr(sf, "LineBuilder", FakeLineBuilder, raising=False) + sf._DESIGN_CACHE.clear() + yield + + +def make_rows(times, values, tags=None, field="value", extra=None): + rows = [] + for t, v in zip(times, values): + row = {"time": t, field: v} + row.update(tags or {}) + row.update(extra or {}) + rows.append(row) + return rows + + +def written_values(local, field="value_filtered"): + return [line.fields[field] for _, line in local.writes] + + +def written_times(local): + return [line.timestamp for _, line in local.writes] + + +BASE_ARGS = {"fc2": "5.0", "sample_rate": "100.0"} + + +def run(local, rows, args=None, table="signal"): + sf.process_writes(local, [TableBatch(table, rows)], dict(BASE_ARGS if args is None else args)) + + +# --------------------------------------------------------------------------- +# M1 β€” skeleton +# --------------------------------------------------------------------------- + + +def test_docstring_header_is_valid_json_with_expected_args(): + header = json.loads(sf.__doc__) + assert header["plugin_type"] == ["onwrite"] + names = {arg["name"] for arg in header["onwrite_args_config"]} + assert names == { + "input_measurement", "input_fields", "tag_keys", "design_type", "prototype", + "order", "ripple", "filter_type", "fc", "fc1", "fc2", "bessel_norm", "sos", + "sample_rate", "init_from_first_sample", "output_target_database", + "output_measurement", "output_field", "field_prefix", "field_suffix", + "config_file_path", + } + + +def test_missing_scipy_reports_install_command(monkeypatch): + monkeypatch.setattr(sf, "np", None) + monkeypatch.setattr(sf, "sp_signal", None) + local = FakeLocal() + sf.process_writes(local, [], {}) + assert len(local.errors) == 1 + assert "influxdb3 install package numpy scipy" in local.errors[0] + + +# --------------------------------------------------------------------------- +# M2 β€” config parsing + validation +# --------------------------------------------------------------------------- + + +def test_defaults(): + cfg = sf.parse_config({"fc2": "5.0"}) + assert cfg.input_fields == ("value",) + assert cfg.design_type == "preset" + assert cfg.prototype == "butter" + assert cfg.order == 4 + assert cfg.filter_type == "lowpass" + assert cfg.fc2 == 5.0 + assert cfg.init_from_first_sample is True + assert cfg.field_prefix == "" + assert cfg.field_suffix == "_filtered" + assert cfg.sample_rate is None + + +@pytest.mark.parametrize( + "args, fragment", + [ + ({"design_type": "zap"}, "design_type"), + ({"prototype": "ellip", "fc2": "5"}, "prototype"), + ({"order": "0", "fc2": "5"}, "order"), + ({"order": "13", "fc2": "5"}, "order"), + ({"order": "four", "fc2": "5"}, "order"), + ({"prototype": "cheby1", "fc2": "5"}, "ripple"), + ({"prototype": "cheby1", "ripple": "0.001", "fc2": "5"}, "ripple"), + ({"prototype": "cheby1", "ripple": "90", "fc2": "5"}, "ripple"), + ({"ripple": "1", "fc2": "5"}, "ripple"), + ({"filter_type": "notch"}, "filter_type"), + ({"filter_type": "lowpass"}, "cutoff"), + ({"filter_type": "highpass"}, "cutoff"), + ({"filter_type": "bandpass", "fc1": "1"}, "fc2"), + ({"filter_type": "bandpass", "fc": "3"}, "band"), + ({"fc": "3", "fc2": "5"}, "both"), + ({"filter_type": "highpass", "fc": "3", "fc1": "5"}, "both"), + ({"fc1": "1", "fc2": "5"}, "fc1"), + ({"filter_type": "highpass", "fc1": "1", "fc2": "5"}, "fc2"), + ({"filter_type": "bandpass", "fc1": "5", "fc2": "5"}, "fc1"), + ({"fc2": "-5"}, "fc2"), + ({"fc2": "5", "sample_rate": "0"}, "sample_rate"), + ({"fc2": "5", "bessel_norm": "weird"}, "bessel_norm"), + ({"design_type": "manual"}, "sos"), + ({"design_type": "manual", "sos": "not json"}, "JSON"), + ({"design_type": "manual", "sos": "[[1, 2, 3]]"}, "6 coefficients"), + ({"design_type": "manual", "sos": "[[1, 0, 0, 0, 0, 0]]"}, "a0"), + ({"sos": "[[1,0,0,1,0,0]]", "fc2": "5"}, "manual"), + ({"fc2": "5", "output_field": "x", "input_fields": "a b"}, "output_field"), + ({"fc2": "5", "init_from_first_sample": "maybe"}, "init_from_first_sample"), + ], +) +def test_rejections(args, fragment): + with pytest.raises(sf.ConfigError) as excinfo: + sf.parse_config(args) + assert fragment.lower() in str(excinfo.value).lower() + + +def test_fc_alias_lowpass_and_highpass(): + assert sf.parse_config({"fc": "5"}).fc2 == 5.0 + cfg = sf.parse_config({"filter_type": "highpass", "fc": "2"}) + assert cfg.fc1 == 2.0 + assert cfg.fc2 is None + + +def test_manual_sos_a0_normalization(): + cfg = sf.parse_config({"design_type": "manual", "sos": "[[2, 0, 0, 2, 1, 0.5]]"}) + assert cfg.sos == ((1.0, 0.0, 0.0, 1.0, 0.5, 0.25),) + + +def test_toml_config_supplies_all_params(tmp_path): + toml = tmp_path / "cfg.toml" + toml.write_text('fc2 = 9.0\norder = 2\n') + cfg = sf.parse_config({"config_file_path": str(toml)}) + assert cfg.fc2 == 9.0 + assert cfg.order == 2 + + +def test_config_file_path_exclusive_with_inline_args(tmp_path): + toml = tmp_path / "cfg.toml" + toml.write_text("fc2 = 9.0\n") + with pytest.raises(sf.ConfigError, match="mutually exclusive"): + sf.parse_config({"config_file_path": str(toml), "fc2": "5"}) + + +def test_toml_relative_path_requires_plugin_dir(tmp_path, monkeypatch): + monkeypatch.delenv("PLUGIN_DIR", raising=False) + monkeypatch.delenv("INFLUXDB3_PLUGIN_DIR", raising=False) + monkeypatch.delenv("VIRTUAL_ENV", raising=False) + with pytest.raises(sf.ConfigError, match="PLUGIN_DIR"): + sf.parse_config({"config_file_path": "cfg.toml"}) + (tmp_path / "cfg.toml").write_text("fc2 = 7.0\n") + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + assert sf.parse_config({"config_file_path": "cfg.toml"}).fc2 == 7.0 + + +def test_toml_relative_path_virtual_env_fallback(tmp_path, monkeypatch): + monkeypatch.delenv("PLUGIN_DIR", raising=False) + monkeypatch.delenv("INFLUXDB3_PLUGIN_DIR", raising=False) + venv = tmp_path / "venv" + venv.mkdir() + (tmp_path / "cfg.toml").write_text("fc2 = 3.0\n") # parent of the venv + monkeypatch.setenv("VIRTUAL_ENV", str(venv)) + assert sf.parse_config({"config_file_path": "cfg.toml"}).fc2 == 3.0 + + +def test_loop_hazard_detection(): + hazard = sf.parse_config({"fc2": "5", "field_suffix": ""}) + assert sf.loop_hazard_fields(hazard) == ["value"] + assert sf.loop_hazard_fields(sf.parse_config({"fc2": "5"})) == [] + other_db = sf.parse_config( + {"fc2": "5", "field_suffix": "", "output_target_database": "elsewhere"} + ) + assert sf.loop_hazard_fields(other_db) == [] + other_meas = sf.parse_config( + {"fc2": "5", "field_suffix": "", "input_measurement": "signal", + "output_measurement": "signal_out"} + ) + assert sf.loop_hazard_fields(other_meas) == [] + # output_measurement set but input unrestricted: rows in the output + # measurement still loop back, so the hazard stands. + all_tables = sf.parse_config( + {"fc2": "5", "field_suffix": "", "output_measurement": "signal_out"} + ) + assert sf.loop_hazard_fields(all_tables) == ["value"] + + +# --------------------------------------------------------------------------- +# M3 β€” sample rate +# --------------------------------------------------------------------------- + + +def test_infer_uniform(): + times = [i * 100_000_000 for i in range(10)] # 10 Hz + assert sf.infer_sample_rate(times) == pytest.approx(10.0) + + +def test_infer_jittered_median_robust(): + base = [i * 100_000_000 for i in range(20)] + base[5] += 30_000_000 # jitter one point + base[13] -= 20_000_000 + assert sf.infer_sample_rate(base) == pytest.approx(10.0, rel=0.01) + + +def test_infer_insufficient(): + assert sf.infer_sample_rate([]) is None + assert sf.infer_sample_rate([123]) is None + assert sf.infer_sample_rate([123, 123]) is None + + +def test_merge_warmup_accumulates_and_caps(): + state = {"warmup_times": [1, 2, 3]} + merged = sf.merge_warmup_times(state, [3, 4, 5]) + assert merged == [1, 2, 3, 4, 5] + big = sf.merge_warmup_times({"warmup_times": list(range(100))}, [200]) + assert len(big) == sf.WARMUP_MAX_TIMES + assert big[-1] == 200 + + +# --------------------------------------------------------------------------- +# M4 β€” filter design +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("prototype", ["butter", "cheby1", "bessel"]) +@pytest.mark.parametrize("filter_type", ["lowpass", "highpass", "bandpass", "bandstop"]) +def test_preset_matrix_matches_scipy(prototype, filter_type): + args = {"prototype": prototype, "filter_type": filter_type, "order": "4"} + if prototype == "cheby1": + args["ripple"] = "1.0" + if filter_type == "lowpass": + args["fc2"] = "5.0" + wn = 5.0 + elif filter_type == "highpass": + args["fc1"] = "5.0" + wn = 5.0 + else: + args["fc1"], args["fc2"] = "2.0", "8.0" + wn = [2.0, 8.0] + cfg = sf.parse_config(args) + sos, coeff_hash = sf.design_filter(cfg, 100.0) + if prototype == "butter": + expected = sp_signal.butter(4, wn, btype=filter_type, output="sos", fs=100.0) + elif prototype == "cheby1": + expected = sp_signal.cheby1(4, 1.0, wn, btype=filter_type, output="sos", fs=100.0) + else: + expected = sp_signal.bessel(4, wn, btype=filter_type, norm="phase", output="sos", fs=100.0) + np.testing.assert_array_equal(sos, expected) + assert isinstance(coeff_hash, str) and len(coeff_hash) == 64 + + +def test_design_memoized(): + cfg = sf.parse_config(BASE_ARGS) + sos_a, hash_a = sf.design_filter(cfg, 100.0) + sos_b, hash_b = sf.design_filter(cfg, 100.0) + assert sos_a is sos_b + assert hash_a == hash_b + + +def test_coeff_hash_changes_with_fs_and_params(): + cfg = sf.parse_config(BASE_ARGS) + _, hash_100 = sf.design_filter(cfg, 100.0) + _, hash_50 = sf.design_filter(cfg, 50.0) + assert hash_100 != hash_50 + cfg2 = sf.parse_config({**BASE_ARGS, "order": "2"}) + _, hash_o2 = sf.design_filter(cfg2, 100.0) + assert hash_o2 != hash_100 + + +def test_nyquist_rejected(): + cfg = sf.parse_config({"fc2": "6.0"}) + with pytest.raises(ValueError, match="fs/2"): + sf.design_iir(cfg, 10.0) + + +def test_unstable_manual_sos_rejected(): + # z^2 - 2.5z + 1 has poles at 2.0 and 0.5 -> unstable + cfg = sf.parse_config({"design_type": "manual", "sos": "[[1, 0, 0, 1, -2.5, 1]]"}) + with pytest.raises(ValueError, match="unstable"): + sf.design_iir(cfg, None) + + +def test_stable_manual_sos_accepted(): + cfg = sf.parse_config({"design_type": "manual", "sos": "[[0.2, 0.4, 0.2, 1, -0.4, 0.2]]"}) + sos = sf.design_iir(cfg, None) + assert sos.shape == (1, 6) + + +# --------------------------------------------------------------------------- +# M5 β€” streaming runtime +# --------------------------------------------------------------------------- + + +def test_chunked_apply_equals_whole_signal(): + rng = np.random.default_rng(42) + x = rng.normal(size=200) + sos = sp_signal.butter(4, 5.0, output="sos", fs=100.0) + zi = sf.init_zi(sos, x[0], True) + whole, _ = sf.apply_filter(sos, x, zi.copy()) + parts = [] + zi_run = zi.copy() + for chunk in np.array_split(x, 7): + y, zi_run = sf.apply_filter(sos, chunk, zi_run) + parts.append(y) + np.testing.assert_array_equal(np.concatenate(parts), whole) + + +def test_init_from_first_sample_suppresses_transient(): + sos = sp_signal.butter(4, 5.0, output="sos", fs=100.0) + dc = np.full(50, 7.5) + primed, _ = sf.apply_filter(sos, dc, sf.init_zi(sos, 7.5, True)) + np.testing.assert_allclose(primed, 7.5, rtol=1e-9) + cold, _ = sf.apply_filter(sos, dc, sf.init_zi(sos, 7.5, False)) + assert abs(cold[0] - 7.5) > 1.0 # visible startup transient + + +# --------------------------------------------------------------------------- +# M6 β€” extraction +# --------------------------------------------------------------------------- + + +def test_tag_heuristic_strings_minus_time_and_inputs(): + rows = [{"time": 1, "value": 1.0, "host": "a", "count": 3}] + groups, dropped = sf.extract_series(rows, ("value",), None) + assert list(groups) == [("value", (("host", "a"),))] + assert dropped == 0 + + +def test_tag_keys_override(): + rows = [{"time": 1, "value": 1.0, "host": "a", "note": "x"}] + groups, _ = sf.extract_series(rows, ("value",), ("host",)) + assert list(groups) == [("value", (("host", "a"),))] + + +def test_nonfinite_dropped_and_counted(): + rows = make_rows([1, 2, 3, 4], [1.0, float("nan"), float("inf"), 2.0]) + groups, dropped = sf.extract_series(rows, ("value",), None) + assert dropped == 2 + assert groups[("value", ())] == [(1, 1.0), (4, 2.0)] + + +def test_duplicate_timestamps_last_wins(): + rows = make_rows([1, 2, 2, 3], [1.0, 5.0, 9.0, 3.0]) + groups, _ = sf.extract_series(rows, ("value",), None) + assert groups[("value", ())] == [(1, 1.0), (2, 9.0), (3, 3.0)] + + +def test_unsorted_times_sorted(): + rows = make_rows([3, 1, 2], [3.0, 1.0, 2.0]) + groups, _ = sf.extract_series(rows, ("value",), None) + assert groups[("value", ())] == [(1, 1.0), (2, 2.0), (3, 3.0)] + + +def test_null_bool_and_string_values_skipped(): + rows = [ + {"time": 1, "value": None}, + {"time": 2, "value": True}, + {"time": 3, "value": "text"}, + ] + groups, dropped = sf.extract_series(rows, ("value",), None) + assert groups == {} + assert dropped == 0 + + +# --------------------------------------------------------------------------- +# M7 β€” per-series state +# --------------------------------------------------------------------------- + + +def uniform_times(n, start=0, fs=100.0): + step = int(1e9 / fs) + return [start + i * step for i in range(n)] + + +def test_state_saved_after_run(): + local = FakeLocal() + times = uniform_times(20) + run(local, make_rows(times, np.sin(np.arange(20)).tolist(), tags={"host": "a"})) + assert len(local.writes) == 20 + (key, state), = local.cache.store.items() + assert key.startswith("signal_filter:signal:value:") + assert state["fs"] == 100.0 + assert state["last_time_ns"] == times[-1] + assert state["warmup_times"] == [] + assert isinstance(state["zi"], list) + np.asarray(state["zi"], dtype=np.float64) # round-trips + + +def test_out_of_order_samples_dropped_with_warning(): + local = FakeLocal() + times = uniform_times(20) + values = np.sin(np.arange(20)).tolist() + run(local, make_rows(times[:10], values[:10])) + first_writes = len(local.writes) + # second commit replays old timestamps plus new ones + run(local, make_rows(times[5:], values[5:])) + assert any("out-of-order" in w for w in local.warns) + assert len(local.writes) == first_writes + 10 # only times[10:] filtered + assert written_times(local) == times + + +def test_param_change_resets_state(): + local = FakeLocal() + times = uniform_times(20) + values = np.sin(np.arange(20)).tolist() + run(local, make_rows(times[:10], values[:10])) + hash_before = list(local.cache.store.values())[0]["coeff_hash"] + run(local, make_rows(times[10:], values[10:]), args={**BASE_ARGS, "fc2": "8.0"}) + hash_after = list(local.cache.store.values())[0]["coeff_hash"] + assert hash_before != hash_after + + +def test_corrupt_cached_zi_shape_reinitializes(): + local = FakeLocal() + times = uniform_times(20) + values = np.sin(np.arange(20)).tolist() + run(local, make_rows(times[:10], values[:10])) + key, state = next(iter(local.cache.store.items())) + state["zi"] = [[0.0, 0.0]] # wrong section count for a 4th-order filter + run(local, make_rows(times[10:], values[10:])) + assert any("re-initializing" in w for w in local.warns) + assert len(local.writes) == 20 # second batch still filtered + healed = local.cache.store[key]["zi"] + assert np.asarray(healed).shape != (1, 2) + + +def test_explicit_sample_rate_beats_frozen_state(): + local = FakeLocal() + times = uniform_times(20) + values = list(range(20)) + run(local, make_rows(times[:10], [float(v) for v in values[:10]]), args={"fc2": "5.0"}) + state = list(local.cache.store.values())[0] + assert state["fs"] == pytest.approx(100.0) # inferred and frozen + run( + local, + make_rows(times[10:], [float(v) for v in values[10:]]), + args={"fc2": "5.0", "sample_rate": "50.0"}, + ) + state = list(local.cache.store.values())[0] + assert state["fs"] == 50.0 + + +# --------------------------------------------------------------------------- +# M8 β€” adapter integration +# --------------------------------------------------------------------------- + + +def test_split_commits_equal_single_commit(): + n = 100 + times = uniform_times(n) + rng = np.random.default_rng(7) + values = (np.sin(2 * np.pi * 1.0 * np.arange(n) / 100.0) + 0.1 * rng.normal(size=n)).tolist() + rows = make_rows(times, values, tags={"host": "a"}) + + single = FakeLocal() + run(single, rows) + + split = FakeLocal() + for i in range(0, n, 20): + run(split, rows[i : i + 20]) + + assert written_times(single) == written_times(split) + assert written_values(single) == written_values(split) # float-exact + + +def test_lowpass_attenuates_high_tone_passes_low_tone(): + fs, n = 100.0, 400 + t = np.arange(n) / fs + low = np.sin(2 * np.pi * 1.0 * t) + high = np.sin(2 * np.pi * 30.0 * t) + rows = make_rows(uniform_times(n), (low + high).tolist()) + local = FakeLocal() + run(local, rows) + out = np.asarray(written_values(local))[100:] # skip settling + t_s = t[100:] + + def amplitude(sig, freq): + c = np.cos(2 * np.pi * freq * t_s) + s = np.sin(2 * np.pi * freq * t_s) + return 2 * math.hypot(np.dot(sig, c) / len(sig), np.dot(sig, s) / len(sig)) + + assert amplitude(out, 30.0) < 0.1 # >= 20 dB down + assert amplitude(out, 1.0) > 0.7 + + +def test_write_failure_leaves_state_unsaved_and_retry_is_clean(): + n = 40 + times = uniform_times(n) + values = np.sin(np.arange(n) * 0.3).tolist() + rows = make_rows(times, values) + + reference = FakeLocal() + run(reference, rows[:20]) + run(reference, rows[20:]) + + local = FakeLocal() + run(local, rows[:20]) + state_before = json.dumps(local.cache.store, sort_keys=True, default=str) + local.fail_writes = True + with pytest.raises(RuntimeError): + run(local, rows[20:]) + assert json.dumps(local.cache.store, sort_keys=True, default=str) == state_before + local.fail_writes = False + run(local, rows[20:]) # retry + assert written_values(local) == written_values(reference) + assert written_times(local) == written_times(reference) + + +def test_refired_rows_with_only_output_field_are_skipped(): + local = FakeLocal() + rows = [ + {"time": t, "value": None, "value_filtered": 1.0, "host": "a"} + for t in uniform_times(10) + ] + run(local, rows) + assert local.writes == [] + assert local.errors == [] + + +def test_warmup_accumulates_across_commits_then_filters(): + local = FakeLocal() + times = uniform_times(9) + values = [float(i) for i in range(9)] + args = {"fc2": "5.0"} # no sample_rate -> inference with warm-up + run(local, make_rows(times[:3], values[:3]), args=args) + assert local.writes == [] + run(local, make_rows(times[3:6], values[3:6]), args=args) + assert local.writes == [] + assert any("warm" in i or "inferring" in i for i in local.infos) + run(local, make_rows(times[6:], values[6:]), args=args) + assert len(local.writes) == 3 # only the freezing batch is emitted + state = list(local.cache.store.values())[0] + assert state["fs"] == pytest.approx(100.0) + assert state["warmup_times"] == [] + + +def test_single_sample_commits_never_freeze_bad_fs(): + local = FakeLocal() + args = {"fc2": "5.0"} + for i in range(5): + run(local, make_rows([uniform_times(9)[i]], [float(i)]), args=args) + assert local.writes == [] + state = list(local.cache.store.values())[0] + assert "fs" not in state or state.get("fs") is None + + +def test_multiple_input_fields_filtered_independently(): + local = FakeLocal() + times = uniform_times(20) + rows = [] + for i, t in enumerate(times): + rows.append({"time": t, "a": float(i), "b": float(-i), "host": "x"}) + run(local, rows, args={**BASE_ARGS, "input_fields": "a b"}) + fields = {name for _, line in local.writes for name in line.fields} + assert fields == {"a_filtered", "b_filtered"} + assert len(local.writes) == 40 + + +def test_series_isolation_by_tags(): + local = FakeLocal() + times = uniform_times(20) + rows_a = make_rows(times, [1.0] * 20, tags={"host": "a"}) + rows_b = make_rows(times, [100.0] * 20, tags={"host": "b"}) + run(local, rows_a + rows_b) + by_host = {} + for _, line in local.writes: + by_host.setdefault(dict(line.tags)["host"], []).append(line.fields["value_filtered"]) + np.testing.assert_allclose(by_host["a"], 1.0, rtol=1e-9) + np.testing.assert_allclose(by_host["b"], 100.0, rtol=1e-9) + assert len(local.cache.store) == 2 + + +def test_output_naming_and_routing(): + local = FakeLocal() + times = uniform_times(20) + run( + local, + make_rows(times, [1.0] * 20), + args={ + **BASE_ARGS, + "output_field": "smooth", + "field_prefix": "f_", + "field_suffix": "_out", + "output_measurement": "clean", + "output_target_database": "otherdb", + }, + ) + db, line = local.writes[0] + assert db == "otherdb" + assert line.measurement == "clean" + assert "f_smooth_out" in line.fields + + +def test_input_measurement_filters_tables(): + local = FakeLocal() + rows = make_rows(uniform_times(20), [1.0] * 20) + sf.process_writes( + local, + [TableBatch("noise", rows), TableBatch("signal", rows)], + {**BASE_ARGS, "input_measurement": "signal"}, + ) + assert len(local.writes) == 20 + assert all(line.measurement == "signal" for _, line in local.writes) + + +def test_config_error_logged_not_raised(): + local = FakeLocal() + sf.process_writes(local, [], {"design_type": "bogus"}) + assert len(local.errors) == 1 + assert "invalid configuration" in local.errors[0] + + +def test_loop_hazard_warning_emitted(): + local = FakeLocal() + run(local, make_rows(uniform_times(20), [1.0] * 20), args={**BASE_ARGS, "field_suffix": ""}) + assert any("write loop" in w for w in local.warns) + + +def test_summary_logged(): + local = FakeLocal() + run(local, make_rows(uniform_times(20), [1.0] * 20)) + assert any("points written" in i for i in local.infos) + + +def test_attribute_style_table_batch_also_accepted(): + local = FakeLocal() + rows = make_rows(uniform_times(20), [1.0] * 20) + sf.process_writes(local, [AttrTableBatch("signal", rows)], dict(BASE_ARGS)) + assert len(local.writes) == 20 + + +def test_manual_mode_needs_no_sample_rate(): + local = FakeLocal() + times = uniform_times(3) # far below warm-up threshold + run( + local, + make_rows(times, [1.0, 2.0, 3.0]), + args={"design_type": "manual", "sos": "[[0.5, 0, 0, 1, 0, 0]]"}, + ) + assert written_values(local) == [0.5, 1.0, 1.5] + + +# --------------------------------------------------------------------------- +# M9 β€” batched writes +# --------------------------------------------------------------------------- + + +def test_series_written_as_single_batch(): + # One write_sync call per series (batched), not one per point. + calls = [] + local = FakeLocal() + orig = local._record_batch + + def counting(db_name, batch): + calls.append(db_name) + orig(db_name, batch) + + local._record_batch = counting + run(local, make_rows(uniform_times(20), [1.0] * 20, tags={"host": "a"})) + assert calls == [None] # single batched write for the one series + assert len(local.writes) == 20 # expanded back to 20 points