Condition Monitoring Trend and Time-to-Alarm Python Script Generator

import csv
from datetime import datetime

CSV_PATH = "pump_01_vibration.csv"
TIME_COL = "timestamp"
VALUE_COL = "velocity_mm_s"
ALERT = 4.5
DANGER = 7.1
WINDOW = 30


def load(path):
    rows = []
    with open(path, newline="") as fh:
        for rec in csv.DictReader(fh):
            t = datetime.fromisoformat(rec[TIME_COL])
            rows.append((t, float(rec[VALUE_COL])))
    rows.sort(key=lambda r: r[0])
    return rows


def fit_line(points):
    # least squares fit of value = a + b * days
    t0 = points[0][0]
    xs = [(t - t0).total_seconds() / 86400.0 for t, _ in points]
    ys = [v for _, v in points]
    n = len(xs)
    mx = sum(xs) / n
    my = sum(ys) / n
    sxx = sum((x - mx) ** 2 for x in xs)
    if sxx == 0:
        return my, 0.0, xs[-1]
    b = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / sxx
    a = my - b * mx
    return a, b, xs[-1]


def days_to(limit, a, b, now_x):
    if b <= 0:
        return None
    x_cross = (limit - a) / b
    return max(0.0, x_cross - now_x)


rows = load(CSV_PATH)
if len(rows) < 5:
    raise SystemExit("Need at least 5 readings")
recent = rows[-WINDOW:]
a, b, now_x = fit_line(recent)
last_value = recent[-1][1]

print("Latest reading:", round(last_value, 3))
print("Trend per day:", round(b, 4))
for name, limit in (("alert", ALERT), ("danger", DANGER)):
    if last_value >= limit:
        print(name, "limit already exceeded")
        continue
    d = days_to(limit, a, b, now_x)
    if d is None:
        print(name, "limit: no upward trend")
    else:
        print(name, "limit in about", round(d, 1), "days")

timestamp,velocity_mm_s
2026-09-01T06:00:00,2.10
2026-09-02T06:00:00,2.14
2026-09-03T06:00:00,2.19