You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
129 lines
4.5 KiB
129 lines
4.5 KiB
from data_provider import DataProvider |
|
import logging |
|
from urllib.parse import urlparse |
|
import os.path |
|
import json |
|
import config |
|
|
|
import pandas as pd |
|
|
|
|
|
logger = logging.getLogger('analytic_toolset') |
|
|
|
|
|
def segments_box(segments): |
|
max_time = 0 |
|
min_time = float("inf") |
|
for segment in segments: |
|
min_time = min(min_time, segment['start']) |
|
max_time = max(max_time, segment['finish']) |
|
min_time = pd.to_datetime(min_time, unit='ms') |
|
max_time = pd.to_datetime(max_time, unit='ms') |
|
return min_time, max_time |
|
|
|
|
|
class PatternDetectionModel: |
|
|
|
def __init__(self, anomaly_id, pattern): |
|
self.anomaly_id = anomaly_id |
|
self.pattern = pattern |
|
|
|
self.__load_anomaly_config() |
|
|
|
parsedUrl = urlparse(self.anomaly_config['panelUrl']) |
|
origin = parsedUrl.scheme + '://' + parsedUrl.netloc |
|
|
|
datasource = self.anomaly_config['datasource'] |
|
metric_name = self.anomaly_config['metric']['targets'][0] |
|
|
|
target_filename = os.path.join(config.METRICS_FOLDER, metric_name + ".json") |
|
datasource['origin'] = origin |
|
dataset_filename = os.path.join(config.DATASET_FOLDER, metric_name + ".csv") |
|
|
|
with open(target_filename, 'r') as file: |
|
target = json.load(file) |
|
|
|
self.data_prov = DataProvider(datasource, target, dataset_filename) |
|
|
|
self.model = None |
|
self.__load_model(pattern) |
|
|
|
def learn(self, segments): |
|
self.model = self.__create_model(self.pattern) |
|
window_size = 200 |
|
|
|
dataframe = self.data_prov.get_dataframe() |
|
start_index, stop_index = 0, len(dataframe) |
|
if len(segments) > 0: |
|
min_time, max_time = segments_box(segments) |
|
try: |
|
start_index = dataframe[dataframe['timestamp'] >= min_time].index[0] |
|
stop_index = dataframe[dataframe['timestamp'] > max_time].index[0] |
|
start_index = max(start_index - window_size, 0) |
|
stop_index = min(stop_index + window_size, len(dataframe)) |
|
except IndexError: |
|
pass |
|
|
|
dataframe = dataframe[start_index:stop_index] |
|
|
|
segments = self.data_prov.transform_anomalies(segments) |
|
self.model.fit(dataframe, segments) |
|
self.__save_model() |
|
return 0 |
|
# return last_prediction_time |
|
|
|
def predict(self, last_prediction_time): |
|
if self.model is None: |
|
return [], last_prediction_time |
|
|
|
window_size = 100 |
|
last_prediction_time = pd.to_datetime(last_prediction_time, unit='ms') |
|
|
|
start_index = self.data_prov.get_upper_bound(last_prediction_time) |
|
start_index = max(0, start_index - window_size) |
|
dataframe = self.data_prov.get_data_range(start_index) |
|
|
|
predicted_indexes = self.model.predict(dataframe) |
|
predicted_indexes = [(x, y) for (x, y) in predicted_indexes if x >= start_index and y >= start_index] |
|
|
|
predicted_times = self.data_prov.inverse_transform_indexes(predicted_indexes) |
|
segments = [] |
|
for time_value in predicted_times: |
|
ts1 = int(time_value[0].timestamp() * 1000) |
|
ts2 = int(time_value[1].timestamp() * 1000) |
|
segments.append({ |
|
'start': min(ts1, ts2), |
|
'finish': max(ts1, ts2) |
|
}) |
|
|
|
last_dataframe_time = dataframe.iloc[- 1]['timestamp'] |
|
last_prediction_time = int(last_dataframe_time.timestamp() * 1000) |
|
return segments, last_prediction_time |
|
# return predicted_anomalies, last_prediction_time |
|
|
|
def synchronize_data(self): |
|
self.data_prov.synchronize() |
|
|
|
def __create_model(self, pattern): |
|
if pattern == "peaks": |
|
from peaks_detector import PeaksDetector |
|
return PeaksDetector() |
|
if pattern == "jumps" or pattern == "drops": |
|
from step_detector import StepDetector |
|
return StepDetector(pattern) |
|
|
|
def __load_anomaly_config(self): |
|
with open(os.path.join(config.ANOMALIES_FOLDER, self.anomaly_id + ".json"), 'r') as config_file: |
|
self.anomaly_config = json.load(config_file) |
|
|
|
def __save_model(self): |
|
logger.info("Save model '%s'" % self.anomaly_id) |
|
model_filename = os.path.join(config.MODELS_FOLDER, self.anomaly_id + ".m") |
|
self.model.save(model_filename) |
|
|
|
def __load_model(self, pattern): |
|
logger.info("Load model '%s'" % self.anomaly_id) |
|
model_filename = os.path.join(config.MODELS_FOLDER, self.pattern + ".m") |
|
if os.path.exists(model_filename): |
|
self.model = self.__create_model(pattern) |
|
self.model.load(model_filename)
|
|
|