48 lines
1.1 KiB
Python
48 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
|
|
import pandas as pd
|
|
|
|
from ..models import Anomaly
|
|
|
|
|
|
class Rule(ABC):
|
|
"""Base class for detection rules.
|
|
|
|
Subclasses declare class attributes `name`, `topic`, `severity`, then
|
|
implement `detect(df)` which receives the topic-specific DataFrame and
|
|
returns a list of Anomaly instances.
|
|
"""
|
|
|
|
name: str = "rule"
|
|
topic: str = ""
|
|
severity: str = "warn"
|
|
|
|
def __init__(self, subject: str = "AUV000") -> None:
|
|
self.subject = subject
|
|
|
|
def bind(self, subject: str) -> "Rule":
|
|
self.subject = subject
|
|
return self
|
|
|
|
@abstractmethod
|
|
def detect(self, df: pd.DataFrame) -> list[Anomaly]:
|
|
...
|
|
|
|
def _make(
|
|
self,
|
|
ts: float,
|
|
value: float | None,
|
|
context: dict,
|
|
) -> Anomaly:
|
|
return Anomaly(
|
|
rule=self.name,
|
|
severity=self.severity,
|
|
timestamp=float(ts),
|
|
subject=self.subject,
|
|
topic=self.topic,
|
|
value=value,
|
|
context=context,
|
|
)
|