diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index 7370d99..fcddf40 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -336,8 +336,8 @@ "author": "Synthefy", "docs_file_link": "https://github.com/influxdata/influxdb3_plugins/blob/main/influxdata/synthefy_forecasting/README.md", "required_plugins": [], - "required_libraries": ["pandas", "requests"], - "last_update": "2026-01-08", + "required_libraries": ["pandas", "requests", "influxdata-plugin-utils>=0.3.0"], + "last_update": "2026-08-10", "trigger_types_supported": ["http"] }, { diff --git a/influxdata/synthefy_forecasting/README.md b/influxdata/synthefy_forecasting/README.md index 5968e11..121070a 100644 --- a/influxdata/synthefy_forecasting/README.md +++ b/influxdata/synthefy_forecasting/README.md @@ -16,7 +16,7 @@ The Synthefy Forecasting Plugin integrates the Synthefy Forecasting API with Inf ## Configuration -Plugin parameters may be specified as key-value pairs in the `--trigger-arguments` flag (CLI) or in the `trigger_arguments` field (API) when creating a trigger, and/or in the JSON body of each HTTP request. Body values override trigger arguments. +Plugin parameters may be specified as key-value pairs in the `--trigger-arguments` flag (CLI) or in the `trigger_arguments` field (API) when creating a trigger, and/or in the JSON body of each HTTP request. Body values override trigger arguments. A `null` in the body counts as unset, so the trigger argument applies; send `{}` for `tags` or `[]` for `metadata_fields` to clear them. ### Plugin metadata @@ -38,11 +38,12 @@ If both are set, the header takes precedence. | `measurement` | string | required | Source measurement (table) containing historical data | | `field` | string | `"value"` | Field name to forecast | | `tags` | string \| dict | `""` | Tag filters. Trigger args: dot-separated string `key:val1@val2.key2:val3`. Request body: JSON object mapping tag name to a string or list of strings. See [Tag filter format](#tag-filter-format). | -| `time_range` | string | `"30d"` | Historical window. Format: ``. Units: `s`, `min`, `h`, `d`, `w`, `m`, `q`, `y` (`m`/`q`/`y` are approximate). | +| `time_range` | string | `"30d"` | Historical window. Format: ``. Units: `us`, `ms`, `s`, `min`, `h`, `d`, `w`, `m`, `q`, `y` (`m`/`q`/`y` are approximate). | | `forecast_horizon` | string | `"7d"` | Forecast duration. Format: `` (same units as `time_range`) or ` points`. | | `model` | string | `"sfm-tabular"` | Synthefy model identifier (e.g., `sfm-tabular`, `Migas-latest`) | | `output_measurement` | string | `"{measurement}_forecast"` | Destination measurement for forecast results | | `metadata_fields` | string \| list | `""` | Trigger args: space-separated list of field names (`"humidity pressure"`). Request body: JSON list of strings. Used as covariates. | +| `max_forecast_points`| integer | `10000` | Upper bound on the forecast points one request may produce. A time-based `forecast_horizon` is divided by the series' own step, so a dense series with a long horizon builds a very large payload. | | `database` | string | `""` | Optional override database for **writes only**. If unset, forecasts are written to the trigger's database. Reads always go to the trigger's database. | #### Tag filter format @@ -54,13 +55,14 @@ The plugin supports multi-value tag filters that are translated to `tag IN ('a', - `.` separates `key:value` pairs - `:` separates the tag name from its value(s) - `@` separates multiple values for the same tag -- Quote a value with `'...'` or `"..."` if it contains special characters such as `:`, `@`, `.` or `'` +- Quote a value with `'...'` or `"..."` if it contains `:`, `@` or `.`. A quote inside a value, as in `Bob's`, needs no escaping. An unclosed quote is rejected. Examples: ``` tags="room:Bedroom" tags="room:Bedroom@Kitchen.location:Hall" tags="room:'Some other room'@Bedroom.device:sensor1" +tags="owner:Bob's.room:Bedroom" ``` **Request body (JSON form)**: @@ -68,6 +70,8 @@ tags="room:'Some other room'@Bedroom.device:sensor1" { "tags": { "room": ["Bedroom", "Kitchen"], "location": "Hall" } } ``` +The body also accepts the string form above. Send `{}` to clear the tag filters configured on the trigger. + #### Forecast points and tags When a tag filter has a single value, that value is added as a tag on every forecast point. When it has multiple values (an `IN (...)` filter), no value is written for that tag — the response covers several tag values at once. @@ -76,9 +80,10 @@ When a tag filter has a single value, that value is added as a tag on every fore ### Dependencies -- Python 3.9 or higher +- Python 3.11 or higher - `pandas` — Data manipulation - `requests` — HTTP client for the Synthefy API +- `influxdata-plugin-utils` — Shared configuration, schema and write helpers ### Installation steps @@ -87,6 +92,7 @@ Using the InfluxDB 3 package manager: ```bash influxdb3 install package pandas influxdb3 install package requests +influxdb3 install package influxdata-plugin-utils ``` ### Prerequisites @@ -320,7 +326,7 @@ Check the [Synthefy documentation](https://docs.synthefy.com) for the most up-to - `process_request(influxdb3_local, query_parameters, request_headers, request_body, args=None)`: handles HTTP requests, merges trigger arguments with request-body overrides, validates input, calls Synthefy, and writes forecast points. - `build_history_query(measurement, field, metadata_fields, tag_filters, start_time)`: builds the parameterized SQL query for historical data. -- `dataframe_to_synthefy_request(df, field, forecast_horizon, metadata_fields, model, task_id)`: converts InfluxDB query rows into the Synthefy forecast request payload. +- `dataframe_to_synthefy_request(influxdb3_local, df, field, forecast_horizon, metadata_fields, model, max_forecast_points, task_id)`: converts InfluxDB query rows into the Synthefy forecast request payload and enforces the point limit. - `forecast_response_to_line_builders(influxdb3_local, forecast_response, output_measurement, tag_filters, model, field_name, task_id)`: converts Synthefy forecast results into InfluxDB line protocol builders. ## Troubleshooting @@ -345,7 +351,19 @@ Check the [Synthefy documentation](https://docs.synthefy.com) for the most up-to ### Invalid interval format -`Invalid interval format: ''. Expected ''.` — `time_range` and `forecast_horizon` must be `` (units: `s`, `min`, `h`, `d`, `w`, `m`, `q`, `y`) or, for `forecast_horizon`, ` points`. +`Invalid interval format: ''. Expected ''.` — `time_range` and `forecast_horizon` must be `` (units: `us`, `ms`, `s`, `min`, `h`, `d`, `w`, `m`, `q`, `y`) or, for `forecast_horizon`, ` points`. + +### Forecast horizon too large + +`forecast_horizon '' resolves to N points … above the max_forecast_points limit` — the horizon is divided by the interval between the last two historical points, so a dense series produces many points. Shorten `forecast_horizon`, use the ` points` form, or raise `max_forecast_points`. + +### Timestamps collapse onto each other + +`N history timestamps differ by less than a microsecond …` or `The series' step of … is finer than a microsecond …` — timestamps are sent to Synthefy with microsecond precision, so a series stepping in nanoseconds cannot be represented. Resample it to a coarser step. + +### History holds repeated timestamps + +`History holds N repeated timestamps, so the window covers more than one series …` — the query matched several tag series and their values are interleaved in one input sequence. Add a `tags` filter that selects a single series. ### Synthefy API errors @@ -361,7 +379,7 @@ If writes fail: - Ensure the database exists (the trigger database, or the override `database` if used) - Ensure the plugin has write permissions -- Check the `[task_id] Error writing forecasts attempt N/M: …` warnings in the InfluxDB logs +- Check the `[task_id] Failed to write forecasts after N attempts: …` error in the InfluxDB logs ### Query file limit exceeded (InfluxDB 3 Core) @@ -369,9 +387,10 @@ If you see "Query would scan X Parquet files, exceeding the file limit" errors, ## Limitations -- Currently supports a single time series per request (one `field` plus optional covariates). +- Currently supports a single time series per request (one `field` plus optional covariates). A request whose window matches several tag series is logged as a warning. - Forecast horizon calculation assumes regular time intervals. -- Tag values containing `:`, `@`, `.` or `'` are only fully supported via the JSON request body or quoted values in the trigger-arguments string form. +- Timestamps are exchanged with microsecond precision; a series whose step is finer is rejected. +- In the trigger-arguments string form, a tag value containing `:`, `@` or `.` must be quoted; the JSON request body needs no quoting. ## License diff --git a/influxdata/synthefy_forecasting/manifest.toml b/influxdata/synthefy_forecasting/manifest.toml index fbbe0a6..438a0e1 100644 --- a/influxdata/synthefy_forecasting/manifest.toml +++ b/influxdata/synthefy_forecasting/manifest.toml @@ -2,7 +2,7 @@ manifest_schema_version = "1.2" [plugin] name = "synthefy_forecasting" -version = "0.1.0" +version = "0.2.0" description = "Integrates Synthefy Forecasting API with InfluxDB 3 for on-demand time series forecasting via HTTP. Reads data from InfluxDB, generates forecasts with Synthefy models, and writes results back." triggers = ["process_request"] homepage = "https://www.influxdata.com/" @@ -16,4 +16,4 @@ exclude = [ [dependencies] database_version = ">=3.8.2" -python = ["pandas", "requests"] +python = ["pandas", "requests", "influxdata-plugin-utils>=0.3.0"] diff --git a/influxdata/synthefy_forecasting/requirements-dev.txt b/influxdata/synthefy_forecasting/requirements-dev.txt new file mode 100644 index 0000000..64e33b3 --- /dev/null +++ b/influxdata/synthefy_forecasting/requirements-dev.txt @@ -0,0 +1,4 @@ +pytest +pandas +requests +influxdata-plugin-utils>=0.3.0 diff --git a/influxdata/synthefy_forecasting/requirements.txt b/influxdata/synthefy_forecasting/requirements.txt index a94cf69..32e549d 100644 --- a/influxdata/synthefy_forecasting/requirements.txt +++ b/influxdata/synthefy_forecasting/requirements.txt @@ -1,2 +1,3 @@ requests -pandas \ No newline at end of file +pandas +influxdata-plugin-utils>=0.3.0 diff --git a/influxdata/synthefy_forecasting/synthefy_forecasting.py b/influxdata/synthefy_forecasting/synthefy_forecasting.py index baf5aaf..a5e69c4 100644 --- a/influxdata/synthefy_forecasting/synthefy_forecasting.py +++ b/influxdata/synthefy_forecasting/synthefy_forecasting.py @@ -23,13 +23,13 @@ { "name": "time_range", "example": "30d", - "description": "Historical data window. Format: '' where unit is one of s, min, h, d, w, m, q, y.", + "description": "Historical data window. Format: '' where unit is one of us, ms, s, min, h, d, w, m, q, y.", "required": false }, { "name": "forecast_horizon", "example": "7d", - "description": "Forecast duration. Format: '' (units: s, min, h, d, w, m, q, y) or ' points'.", + "description": "Forecast duration. Format: '' (units: us, ms, s, min, h, d, w, m, q, y) or ' points'.", "required": false }, { @@ -50,28 +50,111 @@ "description": "Space-separated list of metadata field names to use as covariates. In request body, may also be a JSON list of strings.", "required": false }, + { + "name": "max_forecast_points", + "example": "10000", + "description": "Maximum number of forecast points one request may produce (default: 10000). The horizon is converted to points using the series' own step, so a dense series with a long horizon would otherwise build a very large payload.", + "required": false + }, { "name": "database", "example": "mydb", "description": "Optional override database for writing forecasts. Reads always go to the trigger's database.", "required": false } + ], + "http_body_config": [ + { + "name": "measurement", + "example": "temperature", + "description": "InfluxDB measurement name to read from. Required unless set in the trigger arguments.", + "required": false + }, + { + "name": "field", + "example": "value", + "description": "Field name containing the time series values", + "required": false + }, + { + "name": "tags", + "example": "{'room': ['Bedroom', 'Kitchen'], 'location': 'Hall'}", + "description": "Tag filters as a JSON object mapping tag name to a value or list of values. The dot-separated string form of the trigger arguments is also accepted. Send {} to clear the filters configured on the trigger; null means 'not set', so the trigger argument applies.", + "required": false + }, + { + "name": "time_range", + "example": "30d", + "description": "Historical data window. Format: '' where unit is one of us, ms, s, min, h, d, w, m, q, y.", + "required": false + }, + { + "name": "forecast_horizon", + "example": "7d", + "description": "Forecast duration. Format: '' (units: us, ms, s, min, h, d, w, m, q, y) or ' points'.", + "required": false + }, + { + "name": "model", + "example": "sfm-tabular", + "description": "Synthefy model to use (e.g., 'sfm-tabular', 'Migas-latest'). See README for supported models.", + "required": false + }, + { + "name": "output_measurement", + "example": "temperature_forecast", + "description": "Output measurement name (default: '{measurement}_forecast')", + "required": false + }, + { + "name": "metadata_fields", + "example": "['humidity', 'pressure']", + "description": "JSON list of metadata field names to use as covariates. A space-separated string is also accepted. Send [] to clear the list configured on the trigger; null means 'not set', so the trigger argument applies.", + "required": false + }, + { + "name": "max_forecast_points", + "example": "10000", + "description": "Maximum number of forecast points one request may produce (default: 10000). Accepts a JSON number or a string.", + "required": false + }, + { + "name": "database", + "example": "mydb", + "description": "Optional override database for writing forecasts. Reads always go to the trigger's database.", + "required": false + } + ], + "http_headers_config": [ + { + "name": "X-Synthefy-Api-Key", + "example": "", + "description": "Synthefy API key. Required unless the SYNTHEFY_API_KEY environment variable is set on the InfluxDB process; the header wins when both are present.", + "required": false + } ] } """ import json +import math import os -import random import re -import time import uuid from datetime import datetime, timedelta, timezone from json import JSONDecodeError -from typing import Any, Iterable, Optional, Protocol +from typing import Any import pandas as pd import requests +from influxdata_plugin_utils.config import Validator, load_plugin_config +from influxdata_plugin_utils.introspection import get_field_names, get_tag_names +from influxdata_plugin_utils.parsing import ( + parse_delimited_list, + parse_int, + parse_timedelta, +) +from influxdata_plugin_utils.write import build_line, write_data # Note: LineBuilder is provided by the InfluxDB 3 plugin framework at runtime. @@ -79,35 +162,34 @@ API_KEY_HEADER = "X-Synthefy-Api-Key" API_KEY_ENV_VAR = "SYNTHEFY_API_KEY" +DEFAULT_MAX_FORECAST_POINTS = 10000 -class _LineBuilderInterface(Protocol): - def build(self) -> str: ... - - -class _BatchLines: - """ - Wraps multiple LineBuilder objects into a single object with a build() - method that returns a newline-separated string. Allows batched writes - through the write_sync / write_sync_to_db APIs. - """ +# Calendar units have no fixed length, so they are approximated in days. +CALENDAR_UNIT_DAYS = {"m": 30.42, "q": 91.25, "y": 365.0} - def __init__(self, line_builders: Iterable[_LineBuilderInterface]): - self._line_builders = list(line_builders) - self._built: Optional[str] = None +QUOTE_CHARS = ("'", '"') +# Separators after which a quoted tag value may start. +VALUE_START_CHARS = ":@" - def _coerce_builder(self, builder: _LineBuilderInterface) -> str: - build_fn = getattr(builder, "build", None) - if not callable(build_fn): - raise TypeError("line_builder is missing a callable build()") - return str(build_fn()) +# Synthefy accepts sub-second timestamps. +TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ" - def build(self) -> str: - if self._built is None: - lines = [self._coerce_builder(b) for b in self._line_builders] - if not lines: - raise ValueError("batch_write received no lines to build") - self._built = "\n".join(lines) - return self._built +VALIDATORS: list = [ + Validator("measurement", default="", cast=str), + Validator("field", default="value", cast=str), + Validator("tags", default=None), + Validator("metadata_fields", default=None), + Validator("time_range", default="30d", cast=str), + Validator("forecast_horizon", default="7d", cast=str), + Validator("model", default="sfm-tabular", cast=str), + Validator("output_measurement", default="", cast=str), + Validator("database", default="", cast=str), + Validator( + "max_forecast_points", + default=DEFAULT_MAX_FORECAST_POINTS, + cast=lambda raw: parse_int(raw, minimum=1), + ), +] def quote_identifier(name: str) -> str: @@ -118,96 +200,95 @@ def escape_string_literal(value: str) -> str: return value.replace("'", "''") -def parse_time_interval(raw: str, task_id: str) -> timedelta: +def _load_config(args: dict | None, body: dict | None) -> dict: """ - Parse an interval string ('10min', '2d', '1y', ...) into a timedelta. + Merge trigger arguments with the request body and validate the result. - Supported units: s, min, h, d, w, m (≈30.42d), q (≈91.25d), y (365d). + Body values override trigger arguments. An explicit JSON null means "not set", + so the validator default applies. """ - unit_mapping = { - "s": "seconds", - "min": "minutes", - "h": "hours", - "d": "days", - "w": "weeks", - "m": "days", - "q": "days", - "y": "days", - } - day_conversions = { - "m": 30.42, - "q": 91.25, - "y": 365.0, + args = args or {} + body = body or {} + merged = { + **args, + **{key: value for key, value in body.items() if value is not None}, } + try: + settings = load_plugin_config(merged, validators=VALIDATORS, source="args") + except Exception as e: + raise Exception(f"Invalid configuration: {e}") from e + return {key.lower(): value for key, value in settings.as_dict().items()} + + +def parse_time_interval(raw: str, task_id: str) -> timedelta: + """ + Parse an interval string ('10min', '2d', '1y', ...) into a timedelta. + Supported units: us, ms, s, min, h, d, w, plus the approximate calendar units + m (30.42d), q (91.25d) and y (365d). + """ if not isinstance(raw, str): raise Exception( f"[{task_id}] Invalid interval type: expected string like '10min', got {type(raw).__name__}" ) - match = re.fullmatch(r"(\d+)([a-zA-Z]+)", raw.strip()) - if not match: - raise Exception( - f"[{task_id}] Invalid interval format: '{raw}'. Expected '', e.g. '10min', '2d'." - ) - - number_part, unit = match.groups() - magnitude = int(number_part) - unit = unit.lower() - if unit not in unit_mapping: - raise Exception(f"[{task_id}] Unsupported interval unit '{unit}' in '{raw}'.") - - if unit in day_conversions: - days_approx = int(magnitude * day_conversions[unit]) - if days_approx < 1: + match = re.fullmatch(r"\s*(\d+)\s*([a-zA-Z]+)\s*", raw) + if match and match.group(2).lower() in CALENDAR_UNIT_DAYS: + magnitude = int(match.group(1)) + unit = match.group(2).lower() + days = int(magnitude * CALENDAR_UNIT_DAYS[unit]) + if days < 1: raise Exception( f"[{task_id}] Computed days < 1 for {magnitude}{unit} in '{raw}'." ) - return timedelta(days=days_approx) - - if unit == "s": - return timedelta(seconds=magnitude) - if unit == "min": - return timedelta(minutes=magnitude) - if unit == "h": - return timedelta(hours=magnitude) - if unit == "d": - return timedelta(days=magnitude) - if unit == "w": - return timedelta(weeks=magnitude) - raise Exception(f"[{task_id}] Unsupported interval unit '{unit}' in '{raw}'.") - - -def get_tag_names(influxdb3_local, measurement: str, task_id: str) -> list[str]: - """Return tag column names for `measurement`, or an empty list if none/no schema.""" - query = """ - SELECT column_name - FROM information_schema.columns - WHERE table_name = $measurement - AND data_type = 'Dictionary(Int32, Utf8)' + return timedelta(days=days) + + try: + return parse_timedelta(raw) + except ValueError as e: + raise Exception( + f"[{task_id}] Invalid interval format: '{raw}' ({e}). " + f"Expected '', e.g. '10min', '2d', '1y'." + ) from e + + +def split_unquoted(text: str, separator: str) -> list[str]: """ - res = influxdb3_local.query(query, {"measurement": measurement}) - if not res: - influxdb3_local.info( - f"[{task_id}] No tags found for measurement '{measurement}'." - ) - return [] - return [row["column_name"] for row in res] + Split on `separator`, ignoring separators inside '...' or "..." quotes. + A quote is only special where a value may start: at the beginning of a part + or right after ':' or '@'. Elsewhere it is data, so "Bob's" stays intact. -def get_field_names(influxdb3_local, measurement: str, task_id: str) -> list[str]: - """Return non-tag, non-time field column names for `measurement`.""" - query = """ - SELECT column_name - FROM information_schema.columns - WHERE table_name = $measurement - AND data_type != 'Dictionary(Int32, Utf8)' - AND column_name != 'time' + Raises: + ValueError: if a quote is never closed. """ - res = influxdb3_local.query(query, {"measurement": measurement}) - if not res: - return [] - return [row["column_name"] for row in res] + parts: list[str] = [] + current: list[str] = [] + quote = "" + for char in text: + if quote: + current.append(char) + if char == quote: + quote = "" + elif char in QUOTE_CHARS and (not current or current[-1] in VALUE_START_CHARS): + quote = char + current.append(char) + elif char == separator: + parts.append("".join(current)) + current = [] + else: + current.append(char) + if quote: + raise ValueError(f"unterminated {quote} quote in '{text}'") + parts.append("".join(current)) + return parts + + +def strip_quotes(value: str) -> str: + """Remove one matching pair of surrounding single or double quotes.""" + if len(value) >= 2 and value[0] == value[-1] and value[0] in QUOTE_CHARS: + return value[1:-1] + return value def parse_tags_from_args( @@ -220,7 +301,9 @@ def parse_tags_from_args( - '.' separates pairs - ':' separates the tag key from its value(s) - '@' separates multiple values for one key - - quoted values ('...' or "...") are stripped of their quotes + - a value wrapped in '...' or "..." is stripped of its quotes, and any + separator inside the quotes is treated as part of the value + - a quote inside a value, as in "Bob's", needs no escaping """ if raw is None or raw == "": return {} @@ -230,21 +313,23 @@ def parse_tags_from_args( ) result: dict[str, list[str]] = {} - for pair in raw.split("."): + try: + pairs = split_unquoted(raw, ".") + except ValueError as e: + raise Exception( + f"[{task_id}] Invalid 'tags' string in trigger args: {e}." + ) from e + + for pair in pairs: if not pair: continue - parts = pair.split(":") + parts = split_unquoted(pair, ":") if len(parts) != 2: raise Exception( f"[{task_id}] Invalid tag-value pair: '{pair}' (must contain exactly one ':'; quote values containing ':')" ) tag_name, value_str = parts - values: list[str] = [] - for v in value_str.split("@"): - if len(v) >= 2 and v[0] == v[-1] and v[0] in ("'", '"'): - values.append(v[1:-1]) - else: - values.append(v) + values = [strip_quotes(value) for value in split_unquoted(value_str, "@")] if tag_name not in tag_names: influxdb3_local.warn( @@ -300,6 +385,23 @@ def parse_tags_from_body( return result +def parse_tags( + influxdb3_local, raw: Any, measurement: str, tag_names: list[str], task_id: str +) -> dict[str, list[str]]: + """Dispatch to the JSON-object form (request body) or the string form (trigger args).""" + if isinstance(raw, dict): + return parse_tags_from_body( + influxdb3_local, raw, measurement, tag_names, task_id + ) + if raw is None or isinstance(raw, str): + return parse_tags_from_args( + influxdb3_local, raw, measurement, tag_names, task_id + ) + raise Exception( + f"[{task_id}] Invalid 'tags' format: expected a string or JSON object, got {type(raw).__name__}." + ) + + def parse_metadata_fields( influxdb3_local, raw: Any, @@ -313,17 +415,13 @@ def parse_metadata_fields( """ if raw is None or raw == "": return [] - if isinstance(raw, str): - items = raw.split() - elif isinstance(raw, list): - items = [str(x) for x in raw] - else: + if not isinstance(raw, (str, list)): raise Exception( f"[{task_id}] Invalid 'metadata_fields' format: expected string or list, got {type(raw).__name__}." ) result: list[str] = [] - for item in items: + for item in parse_delimited_list(raw): if item not in field_names: influxdb3_local.warn( f"[{task_id}] Metadata field '{item}' does not exist in '{measurement}'; ignoring." @@ -351,7 +449,7 @@ def build_history_query( select_columns.append(quote_identifier(mf)) select_clause = ", ".join(select_columns) - start_iso = start_time.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + start_iso = start_time.astimezone(timezone.utc).strftime(TIMESTAMP_FORMAT) params: dict[str, Any] = {} where_parts = [f"time >= '{start_iso}'"] @@ -386,11 +484,13 @@ def build_history_query( def dataframe_to_synthefy_request( + influxdb3_local, df: pd.DataFrame, field: str, forecast_horizon: str, metadata_fields: list[str], model: str, + max_forecast_points: int, task_id: str, ) -> dict[str, Any]: """ @@ -405,9 +505,26 @@ def dataframe_to_synthefy_request( df["time"] = pd.to_datetime(df["time"]) df = df.sort_values("time").reset_index(drop=True) - history_timestamps = df["time"].dt.strftime("%Y-%m-%dT%H:%M:%SZ").tolist() + repeated_timestamps = int(df["time"].duplicated().sum()) + if repeated_timestamps: + influxdb3_local.warn( + f"[{task_id}] History holds {repeated_timestamps} repeated timestamps, so the " + f"window covers more than one series and their values are interleaved. " + f"Set 'tags' to select a single series." + ) + + history_timestamps = df["time"].dt.strftime(TIMESTAMP_FORMAT).tolist() history_values = [None if pd.isna(v) else v for v in df[field].tolist()] + collapsed = ( + len(history_timestamps) - len(set(history_timestamps)) - repeated_timestamps + ) + if collapsed > 0: + raise Exception( + f"[{task_id}] {collapsed} history timestamps differ by less than a microsecond " + f"and collapse onto each other. Resample the series to a coarser step." + ) + if len(df) >= 2: time_step = df["time"].iloc[-1] - df["time"].iloc[-2] if time_step <= timedelta(0): @@ -418,23 +535,34 @@ def dataframe_to_synthefy_request( fh = forecast_horizon.strip() if fh.endswith(" points"): try: - num_points = int(fh.replace(" points", "").strip()) - except ValueError: + num_points = parse_int(fh.removesuffix(" points"), minimum=1) + except ValueError as e: raise Exception( - f"[{task_id}] Invalid forecast_horizon: '{forecast_horizon}'." - ) - if num_points < 1: - raise Exception(f"[{task_id}] forecast_horizon must be >= 1 point.") + f"[{task_id}] Invalid forecast_horizon: '{forecast_horizon}' ({e})." + ) from e else: forecast_td = parse_time_interval(fh, task_id) num_points = max(1, int(forecast_td / time_step)) + if num_points > max_forecast_points: + raise Exception( + f"[{task_id}] forecast_horizon '{forecast_horizon}' resolves to {num_points} points " + f"at the series' step of {time_step}, above the max_forecast_points limit of " + f"{max_forecast_points}. Shorten the horizon or raise max_forecast_points." + ) + target_timestamps: list[str] = [] current_time = df["time"].iloc[-1] + time_step for _ in range(num_points): - target_timestamps.append(current_time.strftime("%Y-%m-%dT%H:%M:%SZ")) + target_timestamps.append(current_time.strftime(TIMESTAMP_FORMAT)) current_time += time_step + if len(set(target_timestamps)) != len(target_timestamps): + raise Exception( + f"[{task_id}] The series' step of {time_step} is finer than a microsecond, so " + f"forecast timestamps collapse onto each other. Resample to a coarser step." + ) + target_values = [None] * len(target_timestamps) forecast_sample = { @@ -502,6 +630,19 @@ def call_synthefy_api( raise +def _timestamp_ns(raw: Any) -> int: + """Convert a forecast timestamp to integer nanoseconds; naive values are UTC.""" + ts = pd.Timestamp(raw) + if ts.tz is None: + ts = ts.tz_localize("UTC") + return int(ts.value) + + +def _is_non_finite(value: Any) -> bool: + """True for NaN/inf floats, which would make InfluxDB reject the whole batch.""" + return isinstance(value, float) and not math.isfinite(value) + + def forecast_response_to_line_builders( influxdb3_local, forecast_response: dict[str, Any], @@ -527,7 +668,7 @@ def forecast_response_to_line_builders( forecast_row = forecasts[0] - forecast_payload: Optional[dict] = None + forecast_payload: dict | None = None for f in forecast_row: if isinstance(f, dict) and "timestamps" in f and "values" in f: forecast_payload = f @@ -543,90 +684,86 @@ def forecast_response_to_line_builders( quantiles = forecast_payload.get("quantiles") or {} output_field_name = field_name or forecast_payload.get("sample_id", "value") + static_tags = { + tag_key: tag_values[0] + for tag_key, tag_values in tag_filters.items() + if len(tag_values) == 1 + } + static_tags["model"] = model + builders: list[Any] = [] for i, (ts_str, value) in enumerate(zip(timestamps, values)): if value is None: continue + if _is_non_finite(value): + influxdb3_local.warn( + f"[{task_id}] Non-finite forecast value at '{ts_str}'; skipping point." + ) + continue try: - ts = pd.to_datetime(ts_str) - ts_ns = int(ts.timestamp() * 1e9) + ts_ns = _timestamp_ns(ts_str) except Exception: influxdb3_local.warn( f"[{task_id}] Could not parse timestamp '{ts_str}'; skipping point." ) continue - builder = LineBuilder(output_measurement) - builder.time_ns(ts_ns) - - for tag_key, tag_values in tag_filters.items(): - if len(tag_values) == 1: - builder.tag(tag_key, tag_values[0]) - builder.tag("model", model) - - _set_field(builder, output_field_name, value) + fields: dict[str, Any] = {output_field_name: value} for q_level, q_values in quantiles.items(): - if i < len(q_values) and q_values[i] is not None: - _set_field(builder, f"value_{q_level}", q_values[i]) - - builders.append(builder) + if i >= len(q_values): + continue + q_value = q_values[i] + if q_value is None or _is_non_finite(q_value): + continue + fields[f"value_{q_level}"] = q_value + + builders.append( + build_line( + LineBuilder, + output_measurement, + tags=static_tags, + fields=fields, + time_ns=ts_ns, + ) + ) return builders -def _set_field(builder: Any, name: str, value: Any) -> None: - if isinstance(value, bool): - builder.string_field(name, str(value)) - elif isinstance(value, int): - builder.int64_field(name, value) - elif isinstance(value, float): - builder.float64_field(name, value) - else: - builder.string_field(name, str(value)) - - def write_forecasts_to_influxdb( influxdb3_local, builders: list[Any], - database: Optional[str], + database: str | None, task_id: str, max_retries: int = 3, ) -> None: """ - Write forecast points using write_sync (or write_sync_to_db when `database` - is set), batched into a single line-protocol payload, with exponential backoff retries. + Write forecast points as a single batched, synchronous payload, retrying with + exponential backoff. Writes go to `database` when set, otherwise to the + trigger's own database. """ if not builders: influxdb3_local.warn(f"[{task_id}] No forecast points to write.") return + target = f"database {database}" if database else "trigger database" influxdb3_local.info( - f"[{task_id}] Writing {len(builders)} forecast points to " - f"{'database ' + database if database else 'trigger database'}." + f"[{task_id}] Writing {len(builders)} forecast points to {target}." ) - - batch = _BatchLines(builders) - for attempt in range(max_retries): - try: - if database: - influxdb3_local.write_sync_to_db(database, batch, no_sync=True) - else: - influxdb3_local.write_sync(batch, no_sync=True) - influxdb3_local.info( - f"[{task_id}] Wrote {len(builders)} forecast points (attempt {attempt + 1})." - ) - return - except Exception as e: - influxdb3_local.warn( - f"[{task_id}] Error writing forecasts attempt {attempt + 1}/{max_retries}: {e}" - ) - if attempt < max_retries - 1: - wait_time = (2**attempt) + random.random() - time.sleep(wait_time) - else: - influxdb3_local.error( - f"[{task_id}] Failed to write forecasts after {max_retries} attempts: {e}" - ) - raise + try: + write_data( + influxdb3_local, + builders, + batch=True, + retries=max_retries - 1, + no_sync=True, + database=database, + ) + except Exception as e: + influxdb3_local.error( + f"[{task_id}] Failed to write forecasts after {max_retries} attempts: {e}" + ) + raise + influxdb3_local.info(f"[{task_id}] Wrote {len(builders)} forecast points.") def _decode_request_body(request_body: Any, task_id: str) -> dict: @@ -635,18 +772,19 @@ def _decode_request_body(request_body: Any, task_id: str) -> dict: return {} if isinstance(request_body, dict): return request_body - if isinstance(request_body, bytes): - body_str = request_body.decode("utf-8") - elif isinstance(request_body, str): - body_str = request_body - else: + if not isinstance(request_body, (bytes, str)): raise Exception( f"[{task_id}] Unsupported request_body type: {type(request_body).__name__}" ) + body_str = ( + request_body.decode("utf-8") + if isinstance(request_body, bytes) + else request_body + ) return json.loads(body_str) -def _get_api_key(request_headers: Optional[dict]) -> Optional[str]: +def _get_api_key(request_headers: dict | None) -> str | None: """Return the API key from the request header or env var, or None.""" if request_headers: for key, value in request_headers.items(): @@ -660,7 +798,7 @@ def process_request( query_parameters: dict, request_headers: dict, request_body: Any, - args: Optional[dict] = None, + args: dict | None = None, ) -> dict: """ HTTP entry point. Reads historical data, calls Synthefy, writes the forecast back. @@ -696,24 +834,21 @@ def process_request( return {"message": "Missing API key"} try: - merged_args = {**args, **body_dict} + config = _load_config(args, body_dict) - measurement = merged_args.get("measurement") + measurement = config["measurement"] if not measurement: influxdb3_local.error(f"[{task_id}] 'measurement' argument is required.") return {"message": "'measurement' argument is required"} - field = merged_args.get("field", "value") - time_range_str = merged_args.get("time_range", "30d") - forecast_horizon_str = merged_args.get("forecast_horizon", "7d") - model = merged_args.get("model", "sfm-tabular") - output_measurement = ( - merged_args.get("output_measurement") or f"{measurement}_forecast" - ) - database = merged_args.get("database") or None + field = config["field"] + model = config["model"] + output_measurement = config["output_measurement"] or f"{measurement}_forecast" + database = config["database"] or None + max_forecast_points = config["max_forecast_points"] - field_names = get_field_names(influxdb3_local, measurement, task_id) - tag_names = get_tag_names(influxdb3_local, measurement, task_id) + field_names = get_field_names(influxdb3_local, measurement, use_cache=False) + tag_names = get_tag_names(influxdb3_local, measurement, use_cache=False) if not field_names and not tag_names: influxdb3_local.error( @@ -727,41 +862,14 @@ def process_request( ) return {"message": f"Field '{field}' does not exist in '{measurement}'"} - if "tags" in body_dict: - tag_filters = parse_tags_from_body( - influxdb3_local, - body_dict.get("tags"), - measurement, - tag_names, - task_id, - ) - else: - tag_filters = parse_tags_from_args( - influxdb3_local, - args.get("tags"), - measurement, - tag_names, - task_id, - ) - - if "metadata_fields" in body_dict: - metadata_fields = parse_metadata_fields( - influxdb3_local, - body_dict.get("metadata_fields"), - measurement, - field_names, - task_id, - ) - else: - metadata_fields = parse_metadata_fields( - influxdb3_local, - args.get("metadata_fields"), - measurement, - field_names, - task_id, - ) + tag_filters = parse_tags( + influxdb3_local, config["tags"], measurement, tag_names, task_id + ) + metadata_fields = parse_metadata_fields( + influxdb3_local, config["metadata_fields"], measurement, field_names, task_id + ) - time_range_td = parse_time_interval(time_range_str, task_id) + time_range_td = parse_time_interval(config["time_range"], task_id) start_time = datetime.now(timezone.utc) - time_range_td query, params = build_history_query( @@ -781,7 +889,14 @@ def process_request( return {"message": "No data found"} synthefy_request = dataframe_to_synthefy_request( - df, field, forecast_horizon_str, metadata_fields, model, task_id + influxdb3_local, + df, + field, + config["forecast_horizon"], + metadata_fields, + model, + max_forecast_points, + task_id, ) forecast_response = call_synthefy_api( influxdb3_local, synthefy_request, api_key, task_id diff --git a/influxdata/synthefy_forecasting/test_synthefy_forecasting.py b/influxdata/synthefy_forecasting/test_synthefy_forecasting.py new file mode 100644 index 0000000..79073cd --- /dev/null +++ b/influxdata/synthefy_forecasting/test_synthefy_forecasting.py @@ -0,0 +1,868 @@ +"""Unit and integration tests for the synthefy_forecasting plugin.""" + +import json +import os +import sys +from collections import namedtuple + +import pandas as pd +import pytest +from influxdata_plugin_utils import write as utils_write + +sys.path.insert(0, os.path.dirname(__file__)) +import synthefy_forecasting as sf + +TAG_TYPE = "Dictionary(Int32, Utf8)" + +COLUMNS = { + "time": "Timestamp(Nanosecond, None)", + "value": "Float64", + "humidity": "Float64", + "pressure": "Float64", + "room": TAG_TYPE, + "site": TAG_TYPE, +} + + +# --------------------------------------------------------------------------- +# 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 + + +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 int64_field(self, key, value): + self.fields[key] = f"{value}i" + return self + + def uint64_field(self, key, value): + self.fields[key] = f"{value}u" + return self + + def float64_field(self, key, value): + self.fields[key] = f"{int(value)}.0" if value % 1 == 0 else str(value) + return self + + def bool_field(self, key, value): + self.fields[key] = "true" if value else "false" + return self + + def string_field(self, key, value): + self.fields[key] = f'"{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}={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_field(raw): + if raw.startswith('"'): + return raw[1:-1] + if raw in ("true", "false"): + return raw == "true" + if raw[-1] in ("i", "u"): + return int(raw[:-1]) + return float(raw) + + +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: _parse_field(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, columns=None, rows=None, write_failures=0): + self.cache = FakeCache() + self.columns = COLUMNS if columns is None else columns + self.rows = [] if rows is None else rows + self.write_failures = write_failures + self.queries = [] + self.writes = [] # (db_name | None, Record) per emitted point + self.infos = [] + self.warns = [] + self.errors = [] + + def query(self, query, args=None): + self.queries.append((query, args)) + if "information_schema.columns" not in query: + return self.rows + rows = [{"column_name": n, "data_type": t} for n, t in self.columns.items()] + wanted = (args or {}).get("data_type") + return [r for r in rows if wanted is None or r["data_type"] == wanted] + + 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. + if self.write_failures: + self.write_failures -= 1 + raise RuntimeError("simulated write failure") + 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(self, *args): + raise AssertionError("buffered write must not be used") + + def write_sync(self, batch, no_sync=False): + self._record_batch(None, batch) + + def write_sync_to_db(self, db_name, batch, no_sync=False): + self._record_batch(db_name, batch) + + +class FakeResponse: + def __init__(self, payload): + self._payload = payload + + def raise_for_status(self): + return None + + def json(self): + return self._payload + + +class FakeRequests: + """Stand-in for the `requests` module inside the plugin.""" + + def __init__(self, payload=None, error=None): + self.payload = payload + self.error = error + self.calls = [] + + def post(self, url, json=None, headers=None, timeout=None): + self.calls.append({"url": url, "body": json, "headers": headers, "timeout": timeout}) + if self.error is not None: + raise self.error + return FakeResponse(self.payload) + + +@pytest.fixture(autouse=True) +def _plugin_env(monkeypatch): + monkeypatch.setattr(sf, "LineBuilder", FakeLineBuilder, raising=False) + monkeypatch.setattr(utils_write.time, "sleep", lambda _: None) + yield + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +START_NS = 1_700_000_000_000_000_000 +HOUR_NS = 3_600_000_000_000 + + +def history_rows(count=5, field="value", step_ns=HOUR_NS, extra=None): + rows = [] + for i in range(count): + row = {"time": START_NS + i * step_ns, field: float(i)} + row.update(extra or {}) + rows.append(row) + return rows + + +def forecast_response(timestamps, values, quantiles=None, sample_id="value"): + payload = {"sample_id": sample_id, "timestamps": timestamps, "values": values} + if quantiles is not None: + payload["quantiles"] = quantiles + return {"forecasts": [[payload]]} + + +def run(local, body=None, args=None, headers=None, requests_stub=None, monkeypatch=None): + if requests_stub is not None: + monkeypatch.setattr(sf, "requests", requests_stub) + return sf.process_request( + local, + {}, + {"X-Synthefy-Api-Key": "k"} if headers is None else headers, + json.dumps(body or {}), + args or {}, + ) + + +# --------------------------------------------------------------------------- +# M1 — plugin metadata +# --------------------------------------------------------------------------- + + +def test_docstring_header_is_valid_json_with_expected_args(): + header = json.loads(sf.__doc__) + assert header["plugin_type"] == ["http"] + names = [arg["name"] for arg in header["http_args_config"]] + assert set(names) == { + "measurement", "field", "tags", "time_range", "forecast_horizon", "model", + "output_measurement", "metadata_fields", "max_forecast_points", "database", + } + # every argument is also accepted in the request body, in the same order + assert [arg["name"] for arg in header["http_body_config"]] == names + assert [h["name"] for h in header["http_headers_config"]] == [sf.API_KEY_HEADER] + for section in ("http_args_config", "http_body_config", "http_headers_config"): + for entry in header[section]: + assert set(entry) == {"name", "example", "description", "required"} + + +# --------------------------------------------------------------------------- +# M2 — configuration +# --------------------------------------------------------------------------- + + +def test_config_defaults(): + cfg = sf._load_config({"measurement": "t"}, {}) + assert cfg["field"] == "value" + assert cfg["time_range"] == "30d" + assert cfg["forecast_horizon"] == "7d" + assert cfg["model"] == "sfm-tabular" + assert cfg["output_measurement"] == "" + assert cfg["database"] == "" + assert cfg["max_forecast_points"] == sf.DEFAULT_MAX_FORECAST_POINTS + + +def test_config_body_overrides_args_and_null_falls_back(): + cfg = sf._load_config({"measurement": "t", "model": "from-args"}, {"model": "from-body"}) + assert cfg["model"] == "from-body" + cfg = sf._load_config({"measurement": "t", "model": "from-args"}, {"model": None}) + assert cfg["model"] == "from-args" + + +def test_config_never_reads_a_toml_file(): + cfg = sf._load_config({"measurement": "t", "config_file_path": "/nonexistent.toml"}, {}) + assert "config_file_path" not in cfg + assert cfg["measurement"] == "t" + + +def test_config_leaves_dynaconf_tokens_literal(): + cfg = sf._load_config({"measurement": "@format {env[HOME]}"}, {}) + assert cfg["measurement"] == "@format {env[HOME]}" + + +@pytest.mark.parametrize( + "value, fragment", + [("0", "below minimum 1"), ("-5", "below minimum 1"), ("junk", "Invalid integer")], +) +def test_config_rejects_bad_max_forecast_points(value, fragment): + with pytest.raises(Exception) as excinfo: + sf._load_config({"measurement": "t", "max_forecast_points": value}, {}) + assert "Invalid configuration" in str(excinfo.value) + assert fragment in str(excinfo.value) + + +# --------------------------------------------------------------------------- +# M3 — interval parsing +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "raw, seconds", + [ + ("500us", 0.0005), + ("100ms", 0.1), + ("30s", 30), + ("10min", 600), + ("2h", 7200), + ("30d", 2_592_000), + ("1w", 604_800), + ("1m", 30 * 86_400), + ("2q", 182 * 86_400), + ("1y", 365 * 86_400), + ], +) +def test_parse_time_interval_units(raw, seconds): + assert sf.parse_time_interval(raw, "T").total_seconds() == pytest.approx(seconds) + + +@pytest.mark.parametrize( + "raw, fragment", + [ + ("5x", "Invalid interval format"), + ("abc", "Invalid interval format"), + ("", "Invalid interval format"), + ("0y", "Computed days < 1"), + (30, "Invalid interval type"), + ], +) +def test_parse_time_interval_rejections(raw, fragment): + with pytest.raises(Exception, match=fragment): + sf.parse_time_interval(raw, "T") + + +# --------------------------------------------------------------------------- +# M4 — tag filters +# --------------------------------------------------------------------------- + +TAG_NAMES = ["room", "site", "path"] + + +@pytest.mark.parametrize( + "raw, expected", + [ + ("", {}), + ("room:Bedroom", {"room": ["Bedroom"]}), + ("room:Bedroom@Kitchen.site:north", {"room": ["Bedroom", "Kitchen"], "site": ["north"]}), + ("room:'Some other room'@Bedroom", {"room": ["Some other room", "Bedroom"]}), + ("room:A.room:B", {"room": ["A", "B"]}), + # quoting protects every separator, as documented in the README + ("path:'a:b'", {"path": ["a:b"]}), + ("path:'a.b'", {"path": ["a.b"]}), + ("path:'a@b'", {"path": ["a@b"]}), + ('path:"a:b@c.d"', {"path": ["a:b@c.d"]}), + ("path:Bob's.room:A", {"path": ["Bob's"], "room": ["A"]}), + ('path:5".room:A', {"path": ['5"'], "room": ["A"]}), + ], +) +def test_parse_tags_from_args(raw, expected): + local = FakeLocal() + assert sf.parse_tags_from_args(local, raw, "m", TAG_NAMES, "T") == expected + + +def test_parse_tags_from_args_rejects_ambiguous_pair(): + with pytest.raises(Exception, match="Invalid tag-value pair"): + sf.parse_tags_from_args(FakeLocal(), "path:a:b:c", "m", TAG_NAMES, "T") + with pytest.raises(Exception, match="expected string"): + sf.parse_tags_from_args(FakeLocal(), ["room:A"], "m", TAG_NAMES, "T") + with pytest.raises(Exception, match="unterminated ' quote"): + sf.parse_tags_from_args(FakeLocal(), "room:'Living room", "m", TAG_NAMES, "T") + + +def test_parse_tags_from_args_warns_on_unknown_tag(): + local = FakeLocal() + assert sf.parse_tags_from_args(local, "nope:x.room:A", "m", TAG_NAMES, "T") == {"room": ["A"]} + assert any("Tag 'nope' does not exist" in w for w in local.warns) + + +@pytest.mark.parametrize( + "raw, expected", + [ + (None, {}), + ({}, {}), + ({"room": "Bedroom"}, {"room": ["Bedroom"]}), + ({"room": ["Bedroom", "Kitchen"]}, {"room": ["Bedroom", "Kitchen"]}), + ({"room": [1, 2]}, {"room": ["1", "2"]}), + ({"room": []}, {}), + ], +) +def test_parse_tags_from_body(raw, expected): + assert sf.parse_tags_from_body(FakeLocal(), raw, "m", TAG_NAMES, "T") == expected + + +@pytest.mark.parametrize( + "raw, fragment", + [(["Bedroom"], "expected JSON object"), ({"room": 5}, "expected string or list")], +) +def test_parse_tags_from_body_rejections(raw, fragment): + with pytest.raises(Exception, match=fragment): + sf.parse_tags_from_body(FakeLocal(), raw, "m", TAG_NAMES, "T") + + +def test_parse_tags_dispatches_on_the_value_type(): + local = FakeLocal() + assert sf.parse_tags(local, {"room": "A"}, "m", TAG_NAMES, "T") == {"room": ["A"]} + assert sf.parse_tags(local, "room:A", "m", TAG_NAMES, "T") == {"room": ["A"]} + assert sf.parse_tags(local, None, "m", TAG_NAMES, "T") == {} + with pytest.raises(Exception, match="expected a string or JSON object"): + sf.parse_tags(local, ["room:A"], "m", TAG_NAMES, "T") + + +# --------------------------------------------------------------------------- +# M5 — metadata fields +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "raw, expected", + [ + ("", []), + ("humidity pressure", ["humidity", "pressure"]), + (["humidity", "pressure"], ["humidity", "pressure"]), + ("humidity pressure", ["humidity", "pressure"]), + ], +) +def test_parse_metadata_fields(raw, expected): + names = ["humidity", "pressure"] + assert sf.parse_metadata_fields(FakeLocal(), raw, "m", names, "T") == expected + + +def test_parse_metadata_fields_drops_unknown_with_warning(): + local = FakeLocal() + result = sf.parse_metadata_fields(local, "humidity nope", "m", ["humidity"], "T") + assert result == ["humidity"] + assert any("Metadata field 'nope' does not exist" in w for w in local.warns) + + +def test_parse_metadata_fields_rejects_other_types(): + with pytest.raises(Exception, match="expected string or list"): + sf.parse_metadata_fields(FakeLocal(), 5, "m", [], "T") + + +# --------------------------------------------------------------------------- +# M6 — history query +# --------------------------------------------------------------------------- + + +def test_history_query_binds_tag_values_and_quotes_identifiers(): + start = pd.Timestamp("2026-01-01T00:00:00Z").to_pydatetime() + query, params = sf.build_history_query( + "temp", "va\"lue", ["humidity"], {"room": ["A"], "site": ["x", "y"]}, start + ) + assert '"va""lue"' in query and '"humidity"' in query + assert '"room" = $tag_val_0' in query + assert '"site" IN ($tag_val_1, $tag_val_2)' in query + assert params == {"tag_val_0": "A", "tag_val_1": "x", "tag_val_2": "y"} + assert "time >= '2026-01-01T00:00:00.000000Z'" in query + assert "ORDER BY time" in query + + +# --------------------------------------------------------------------------- +# M7 — Synthefy request payload +# --------------------------------------------------------------------------- + + +def test_request_payload_derives_step_and_targets(): + df = pd.DataFrame(history_rows(4)) + request = sf.dataframe_to_synthefy_request(FakeLocal(), df, "value", "3h", [], "sfm-tabular", 10_000, "T") + sample = request["samples"][0][0] + assert request["model"] == "sfm-tabular" + assert sample["forecast"] is True and sample["metadata"] is False + assert len(sample["history_timestamps"]) == 4 + assert sample["history_timestamps"][-1] == "2023-11-15T01:13:20.000000Z" + # the horizon continues the series at its own step, starting after the last point + assert sample["target_timestamps"] == [ + "2023-11-15T02:13:20.000000Z", + "2023-11-15T03:13:20.000000Z", + "2023-11-15T04:13:20.000000Z", + ] + assert sample["target_values"] == [None, None, None] + + +def test_request_payload_point_form_and_covariates(): + rows = history_rows(3, extra={"humidity": 1.0}) + request = sf.dataframe_to_synthefy_request( + FakeLocal(), pd.DataFrame(rows), "value", "2 points", ["humidity"], "m", 10_000, "T" + ) + samples = request["samples"][0] + assert len(samples[0]["target_timestamps"]) == 2 + assert len(samples) == 2 + assert samples[1]["sample_id"] == "humidity" + assert samples[1]["metadata"] is True and samples[1]["forecast"] is False + + +@pytest.mark.parametrize( + "horizon, cap, fragment", + [ + ("7d", 10_000, "above the max_forecast_points limit"), + ("50000 points", 10_000, "above the max_forecast_points limit"), + ("0 points", 10_000, "below minimum 1"), + ("many points", 10_000, "Invalid forecast_horizon"), + ("2 points points", 10_000, "Invalid forecast_horizon"), + ], +) +def test_request_payload_rejections(horizon, cap, fragment): + df = pd.DataFrame(history_rows(3, step_ns=1_000_000_000)) + with pytest.raises(Exception, match=fragment): + sf.dataframe_to_synthefy_request(FakeLocal(), df, "value", horizon, [], "m", cap, "T") + + +def test_request_payload_allows_a_raised_cap(): + df = pd.DataFrame(history_rows(3, step_ns=1_000_000_000)) + request = sf.dataframe_to_synthefy_request(FakeLocal(), df, "value", "1h", [], "m", 10_000, "T") + assert len(request["samples"][0][0]["target_timestamps"]) == 3600 + + +def test_request_payload_warns_when_the_window_holds_several_series(): + # two series without a tag filter: every timestamp appears twice + rows = history_rows(3) + history_rows(3, field="value") + local = FakeLocal() + sf.dataframe_to_synthefy_request(local, pd.DataFrame(rows), "value", "1 points", [], "m", 10, "T") + assert any("3 repeated timestamps" in w and "Set 'tags'" in w for w in local.warns) + + local = FakeLocal() + sf.dataframe_to_synthefy_request( + local, pd.DataFrame(history_rows(3)), "value", "1 points", [], "m", 10, "T" + ) + assert local.warns == [] + + +def test_request_payload_keeps_sub_second_steps(): + df = pd.DataFrame(history_rows(6, step_ns=100_000_000)) + sample = sf.dataframe_to_synthefy_request( + FakeLocal(), df, "value", "500ms", [], "m", 10_000, "T" + )["samples"][0][0] + assert sample["history_timestamps"][1].endswith(".100000Z") + assert len(set(sample["target_timestamps"])) == 5 + + +def test_request_payload_rejects_steps_finer_than_a_microsecond(): + df = pd.DataFrame(history_rows(4, step_ns=500)) + with pytest.raises(Exception, match="less than a microsecond"): + sf.dataframe_to_synthefy_request(FakeLocal(), df, "value", "2 points", [], "m", 10, "T") + + +# --------------------------------------------------------------------------- +# M8 — forecast response to line protocol +# --------------------------------------------------------------------------- + + +def test_response_writes_tags_quantiles_and_exact_nanoseconds(): + response = forecast_response( + ["2026-01-01T00:00:00.123456789Z", "2026-01-01T01:00:00Z"], + [1.5, 2.5], + quantiles={"0.1": [1.0, 2.0], "0.9": [2.0, 3.0]}, + ) + builders = sf.forecast_response_to_line_builders( + FakeLocal(), response, "temp_forecast", {"room": ["A"], "site": ["x", "y"]}, + "sfm-tabular", "temp", "T", + ) + first = _parse_lp(builders[0].build()) + assert first.measurement == "temp_forecast" + # a single-valued filter is written as a tag; a multi-valued one is not + assert first.tags == {"room": "A", "model": "sfm-tabular"} + assert first.fields == {"temp": 1.5, "value_0.1": 1.0, "value_0.9": 2.0} + assert first.timestamp == 1767225600123456789 + assert len(builders) == 2 + + +def test_response_treats_naive_timestamps_as_utc(): + aware = forecast_response(["2026-01-01T00:00:00Z"], [1.0]) + naive = forecast_response(["2026-01-01T00:00:00"], [1.0]) + build = lambda r: sf.forecast_response_to_line_builders( + FakeLocal(), r, "m", {}, "mdl", "v", "T" + )[0].timestamp + assert build(aware) == build(naive) + + +def test_response_skips_unusable_points_but_keeps_the_rest(): + local = FakeLocal() + response = forecast_response( + ["2026-01-01T00:00:00Z", "2026-01-01T01:00:00Z", "not-a-time", "2026-01-01T03:00:00Z"], + [1.0, None, 3.0, float("nan")], + ) + builders = sf.forecast_response_to_line_builders( + local, response, "m", {}, "mdl", "v", "T" + ) + assert len(builders) == 1 + assert any("Non-finite forecast value" in w for w in local.warns) + assert any("Could not parse timestamp" in w for w in local.warns) + + +def test_response_drops_non_finite_quantiles_only(): + response = forecast_response( + ["2026-01-01T00:00:00Z"], [1.0], quantiles={"0.1": [float("inf")], "0.9": [2.0]} + ) + builders = sf.forecast_response_to_line_builders( + FakeLocal(), response, "m", {}, "mdl", "v", "T" + ) + assert _parse_lp(builders[0].build()).fields == {"v": 1.0, "value_0.9": 2.0} + + +@pytest.mark.parametrize( + "response, fragment", + [ + ({}, "missing 'forecasts' field"), + ({"forecasts": []}, "No forecasts in response"), + ({"forecasts": [[{"nope": 1}]]}, "No forecast payload"), + ], +) +def test_response_rejections(response, fragment): + with pytest.raises(ValueError, match=fragment): + sf.forecast_response_to_line_builders( + FakeLocal(), response, "m", {}, "mdl", "v", "T" + ) + + +# --------------------------------------------------------------------------- +# M9 — writes +# --------------------------------------------------------------------------- + + +def _two_builders(): + return sf.forecast_response_to_line_builders( + FakeLocal(), + forecast_response(["2026-01-01T00:00:00Z", "2026-01-01T01:00:00Z"], [1.0, 2.0]), + "m", {}, "mdl", "v", "T", + ) + + +def test_write_batches_all_points_into_one_payload(): + local = FakeLocal() + calls = [] + original = local._record_batch + local._record_batch = lambda db, batch: (calls.append(db), original(db, batch)) + + sf.write_forecasts_to_influxdb(local, _two_builders(), None, "T") + assert calls == [None] # a single batched write, not one call per point + assert len(local.writes) == 2 + + +def test_write_routes_to_the_override_database(): + local = FakeLocal() + sf.write_forecasts_to_influxdb(local, _two_builders(), "other", "T") + assert [db for db, _ in local.writes] == ["other", "other"] + assert any("database other" in i for i in local.infos) + + +def test_write_retries_then_succeeds(): + local = FakeLocal(write_failures=2) + sf.write_forecasts_to_influxdb(local, _two_builders(), None, "T") + assert len(local.writes) == 2 + assert local.errors == [] + + +def test_write_reports_and_reraises_after_exhausting_retries(): + local = FakeLocal(write_failures=3) + with pytest.raises(RuntimeError): + sf.write_forecasts_to_influxdb(local, _two_builders(), None, "T") + assert any("Failed to write forecasts after 3 attempts" in e for e in local.errors) + + +def test_write_skips_an_empty_result(): + local = FakeLocal() + sf.write_forecasts_to_influxdb(local, [], None, "T") + assert local.writes == [] + assert any("No forecast points to write" in w for w in local.warns) + + +# --------------------------------------------------------------------------- +# M10 — request body decoding +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "body, expected", + [ + (None, {}), + ("", {}), + (b"", {}), + ({"a": 1}, {"a": 1}), + ('{"a": 1}', {"a": 1}), + (b'{"a": 1}', {"a": 1}), + ], +) +def test_decode_request_body(body, expected): + assert sf._decode_request_body(body, "T") == expected + + +def test_decode_request_body_rejects_unsupported_type(): + with pytest.raises(Exception, match="Unsupported request_body type"): + sf._decode_request_body(42, "T") + + +# --------------------------------------------------------------------------- +# M11 — process_request +# --------------------------------------------------------------------------- + + +def test_full_flow_reads_forecasts_and_writes(monkeypatch): + local = FakeLocal(rows=history_rows(5, extra={"humidity": 1.0})) + stub = FakeRequests( + forecast_response(["2026-01-01T00:00:00Z", "2026-01-01T01:00:00Z"], [10.0, 11.0]) + ) + result = run( + local, + body={ + "measurement": "sf_temp", + "tags": {"room": "Bedroom"}, + "metadata_fields": ["humidity"], + "forecast_horizon": "2 points", + }, + requests_stub=stub, + monkeypatch=monkeypatch, + ) + + assert result == { + "message": "Forecast generated and written to InfluxDB. 2 forecast points written." + } + call = stub.calls[0] + assert call["url"] == "https://forecast.synthefy.com/v2/forecast" + assert call["headers"]["X-API-Key"] == "k" + assert [s["sample_id"] for s in call["body"]["samples"][0]] == ["value", "humidity"] + + written = [record for _, record in local.writes] + assert [r.measurement for r in written] == ["sf_temp_forecast"] * 2 + assert written[0].tags == {"room": "Bedroom", "model": "sfm-tabular"} + assert [r.fields["value"] for r in written] == [10.0, 11.0] + assert local.errors == [] + + +def test_full_flow_honours_output_measurement_and_database(monkeypatch): + local = FakeLocal(rows=history_rows(3)) + stub = FakeRequests(forecast_response(["2026-01-01T00:00:00Z"], [10.0])) + run( + local, + args={"measurement": "sf_temp", "output_measurement": "my_fc", "database": "other"}, + body={"forecast_horizon": "1 points"}, + requests_stub=stub, + monkeypatch=monkeypatch, + ) + db, record = local.writes[0] + assert (db, record.measurement) == ("other", "my_fc") + + +def test_body_overrides_trigger_arguments(monkeypatch): + local = FakeLocal(rows=history_rows(3)) + stub = FakeRequests(forecast_response(["2026-01-01T00:00:00Z"], [10.0])) + run( + local, + args={"measurement": "sf_temp", "model": "from-args"}, + body={"model": "from-body", "forecast_horizon": "1 points"}, + requests_stub=stub, + monkeypatch=monkeypatch, + ) + assert stub.calls[0]["body"]["model"] == "from-body" + assert local.writes[0][1].tags["model"] == "from-body" + + +@pytest.mark.parametrize( + "body_extra, expected_room, expected_samples", + [ + ({}, "Bedroom", ["value", "humidity"]), + # a null means "not set", so the trigger argument still applies + ({"tags": None, "metadata_fields": None}, "Bedroom", ["value", "humidity"]), + # an empty value clears the trigger argument + ({"tags": {}, "metadata_fields": []}, None, ["value"]), + ({"tags": {"room": "Hall"}}, "Hall", ["value", "humidity"]), + # the body accepts the trigger-argument string form too + ({"tags": "room:Hall"}, "Hall", ["value", "humidity"]), + ], +) +def test_tags_and_covariates_merge_with_trigger_arguments( + body_extra, expected_room, expected_samples, monkeypatch +): + local = FakeLocal(rows=history_rows(3, extra={"humidity": 1.0})) + stub = FakeRequests(forecast_response(["2026-01-01T00:00:00Z"], [10.0])) + run( + local, + args={"measurement": "sf_temp", "tags": "room:Bedroom", "metadata_fields": "humidity"}, + body={"forecast_horizon": "1 points", **body_extra}, + requests_stub=stub, + monkeypatch=monkeypatch, + ) + assert local.writes[0][1].tags.get("room") == expected_room + assert [s["sample_id"] for s in stub.calls[0]["body"]["samples"][0]] == expected_samples + + +@pytest.mark.parametrize( + "body, message", + [ + ({}, "'measurement' argument is required"), + ({"measurement": "nope"}, "Measurement 'nope' not found"), + ({"measurement": "sf_temp", "field": "nope"}, + "Field 'nope' does not exist in 'sf_temp'"), + ], +) +def test_request_rejections_before_the_api_call(body, message, monkeypatch): + columns = {} if body.get("measurement") == "nope" else COLUMNS + local = FakeLocal(columns=columns, rows=history_rows(3)) + stub = FakeRequests(error=AssertionError("API must not be called")) + assert run(local, body=body, requests_stub=stub, monkeypatch=monkeypatch) == { + "message": message + } + assert stub.calls == [] + + +def test_missing_api_key_stops_before_touching_the_database(): + local = FakeLocal(rows=history_rows(3)) + result = sf.process_request(local, {}, {}, '{"measurement": "sf_temp"}', {}) + assert result == {"message": "Missing API key"} + assert local.queries == [] + + +def test_api_key_falls_back_to_the_environment(monkeypatch): + monkeypatch.setenv(sf.API_KEY_ENV_VAR, "env-key") + local = FakeLocal(rows=history_rows(3)) + stub = FakeRequests(forecast_response(["2026-01-01T00:00:00Z"], [10.0])) + run( + local, + body={"measurement": "sf_temp", "forecast_horizon": "1 points"}, + headers={}, + requests_stub=stub, + monkeypatch=monkeypatch, + ) + assert stub.calls[0]["headers"]["X-API-Key"] == "env-key" + + +def test_empty_history_returns_no_data_without_calling_the_api(monkeypatch): + local = FakeLocal(rows=[]) + stub = FakeRequests(error=AssertionError("API must not be called")) + result = run(local, body={"measurement": "sf_temp"}, requests_stub=stub, monkeypatch=monkeypatch) + assert result == {"message": "No data found"} + assert stub.calls == [] + + +def test_invalid_json_body_is_reported(): + local = FakeLocal() + result = sf.process_request(local, {}, {"X-Synthefy-Api-Key": "k"}, "{oops", {}) + assert result == {"message": "Invalid JSON in request body"} + assert any("Invalid JSON in request body" in e for e in local.errors) + + +def test_api_failure_is_logged_and_returned(monkeypatch): + local = FakeLocal(rows=history_rows(3)) + stub = FakeRequests(error=RuntimeError("503 Service Unavailable")) + result = run( + local, + body={"measurement": "sf_temp", "forecast_horizon": "1 points"}, + requests_stub=stub, + monkeypatch=monkeypatch, + ) + assert result == {"message": "Error: 503 Service Unavailable"} + assert any("Synthefy API call failed" in e for e in local.errors) + assert local.writes == [] + + +def test_configuration_error_is_returned_not_raised(monkeypatch): + local = FakeLocal(rows=history_rows(3)) + stub = FakeRequests(error=AssertionError("API must not be called")) + result = run( + local, + body={"measurement": "sf_temp", "max_forecast_points": "junk"}, + requests_stub=stub, + monkeypatch=monkeypatch, + ) + assert "Invalid configuration" in result["message"] + assert any("HTTP request forecast failed" in e for e in local.errors)