hastic
/
hastic-server
4
0
Fork 0
Code Issues Pull Requests Projects Releases 1 Wiki Activity
Browse Source

LF eol in all files & add doc about that & setting to settings.json

pull/1/head
Alexey Velikiy 6 years ago
parent
aba8f97a8d
commit
72e8c4520e
9 changed files with 320 additions and 315 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 3
      analytics/.vscode/settings.json
  2. 40
      analytics/Codestyle.md
  3. 74
      analytics/config.py
  4. 10
      analytics/detectors/__init__.py
  5. 276
      analytics/detectors/jump_detector.py
  6. 128
      analytics/server.py
  7. 4
      analytics/services/__init__.py
  8. 16
      analytics/services/data_service.py
  9. 84
      analytics/services/server_service.py

3
analytics/.vscode/settings.json vendored
Unescape View File

@ -3,5 +3,6 @@
"python.linting.enabled": false, "python.linting.enabled": false,
"terminal.integrated.shell.windows": "C:\\WINDOWS\\System32\\WindowsPowerShell\\v1.0\\powershell.exe", "terminal.integrated.shell.windows": "C:\\WINDOWS\\System32\\WindowsPowerShell\\v1.0\\powershell.exe",
"editor.tabSize": 4, "editor.tabSize": 4,
"editor.insertSpaces": true "editor.insertSpaces": true,
"files.eol": "\n"
} }

40
analytics/Codestyle.md
Unescape View File

@ -1,19 +1,23 @@
# Imports # Line endings
You import local files first, than spesific liba and then standart libs. We use CRLS everywhere
So you import from something very scecific to something very common.
It allows you to pay attention on most important things from beginning. # Imports
``` You import local files first, than spesific liba and then standart libs.
So you import from something very scecific to something very common.
from data_provider import DataProvider It allows you to pay attention on most important things from beginning.
from anomaly_model import AnomalyModel
from pattern_detection_model import PatternDetectionModel ```
import numpy as np from data_provider import DataProvider
from anomaly_model import AnomalyModel
from scipy.signal import argrelextrema from pattern_detection_model import PatternDetectionModel
import pickle import numpy as np
from scipy.signal import argrelextrema
import pickle
``` ```

74
analytics/config.py
Unescape View File

@ -1,37 +1,37 @@
import os import os
import json import json
PARENT_FOLDER = os.path.dirname(os.path.dirname(os.path.realpath(__file__))) PARENT_FOLDER = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
DATA_FOLDER = os.path.join(PARENT_FOLDER, 'data') DATA_FOLDER = os.path.join(PARENT_FOLDER, 'data')
CONFIG_FILE = os.path.join(PARENT_FOLDER, 'config.json') CONFIG_FILE = os.path.join(PARENT_FOLDER, 'config.json')
config_exists = os.path.isfile(CONFIG_FILE) config_exists = os.path.isfile(CONFIG_FILE)
if config_exists: if config_exists:
with open(CONFIG_FILE) as f: with open(CONFIG_FILE) as f:
config = json.load(f) config = json.load(f)
def get_config_field(field, default_val = None): def get_config_field(field, default_val = None):
if field in os.environ: if field in os.environ:
return os.environ[field] return os.environ[field]
if config_exists and field in config: if config_exists and field in config:
return config[field] return config[field]
if default_val is not None: if default_val is not None:
return default_val return default_val
raise Exception('Please configure {}'.format(field)) raise Exception('Please configure {}'.format(field))
DATASET_FOLDER = os.path.join(DATA_FOLDER, 'datasets') DATASET_FOLDER = os.path.join(DATA_FOLDER, 'datasets')
ANALYTIC_UNITS_FOLDER = os.path.join(DATA_FOLDER, 'analytic_units') ANALYTIC_UNITS_FOLDER = os.path.join(DATA_FOLDER, 'analytic_units')
MODELS_FOLDER = os.path.join(DATA_FOLDER, 'models') MODELS_FOLDER = os.path.join(DATA_FOLDER, 'models')
METRICS_FOLDER = os.path.join(DATA_FOLDER, 'metrics') METRICS_FOLDER = os.path.join(DATA_FOLDER, 'metrics')
HASTIC_API_KEY = get_config_field('HASTIC_API_KEY') HASTIC_API_KEY = get_config_field('HASTIC_API_KEY')
ZEROMQ_CONNECTION_STRING = get_config_field('ZEROMQ_CONNECTION_STRING', 'tcp://*:8002') ZEROMQ_CONNECTION_STRING = get_config_field('ZEROMQ_CONNECTION_STRING', 'tcp://*:8002')

10
analytics/detectors/__init__.py
Unescape View File

@ -1,5 +1,5 @@
from detectors.general_detector import GeneralDetector from detectors.general_detector import GeneralDetector
from detectors.pattern_detection_model import PatternDetectionModel from detectors.pattern_detection_model import PatternDetectionModel
from detectors.peaks_detector import PeaksDetector from detectors.peaks_detector import PeaksDetector
from detectors.step_detector import StepDetector from detectors.step_detector import StepDetector
from detectors.jump_detector import Jumpdetector from detectors.jump_detector import Jumpdetector

276
analytics/detectors/jump_detector.py
Unescape View File

@ -1,138 +1,138 @@
import numpy as np import numpy as np
import pickle import pickle
import scipy.signal import scipy.signal
from scipy.fftpack import fft from scipy.fftpack import fft
from scipy.signal import argrelextrema from scipy.signal import argrelextrema
import math import math
def is_intersect(target_segment, segments): def is_intersect(target_segment, segments):
for segment in segments: for segment in segments:
start = max(segment['start'], target_segment[0]) start = max(segment['start'], target_segment[0])
finish = min(segment['finish'], target_segment[1]) finish = min(segment['finish'], target_segment[1])
if start <= finish: if start <= finish:
return True return True
return False return False
def exponential_smoothing(series, alpha): def exponential_smoothing(series, alpha):
result = [series[0]] result = [series[0]]
for n in range(1, len(series)): for n in range(1, len(series)):
result.append(alpha * series[n] + (1 - alpha) * result[n-1]) result.append(alpha * series[n] + (1 - alpha) * result[n-1])
return result return result
class Jumpdetector: class Jumpdetector:
def __init__(self, pattern): def __init__(self, pattern):
self.pattern = pattern self.pattern = pattern
self.segments = [] self.segments = []
self.confidence = 1.5 self.confidence = 1.5
self.convolve_max = 120 self.convolve_max = 120
def fit(self, dataframe, segments): def fit(self, dataframe, segments):
data = dataframe['value'] data = dataframe['value']
confidences = [] confidences = []
convolve_list = [] convolve_list = []
for segment in segments: for segment in segments:
if segment['labeled']: if segment['labeled']:
segment_data = data[segment['start'] : segment['finish'] + 1] segment_data = data[segment['start'] : segment['finish'] + 1]
segment_min = min(segment_data) segment_min = min(segment_data)
segment_max = max(segment_data) segment_max = max(segment_data)
confidences.append(0.20 * (segment_max - segment_min)) confidences.append(0.20 * (segment_max - segment_min))
flat_segment = segment_data.rolling(window=5).mean() #сглаживаем сегмент flat_segment = segment_data.rolling(window=5).mean() #сглаживаем сегмент
# в идеале нужно посмотреть гистограмму сегмента и выбрать среднее значение, # в идеале нужно посмотреть гистограмму сегмента и выбрать среднее значение,
# далее от него брать + -120 # далее от него брать + -120
segment_summ = 0 segment_summ = 0
for val in flat_segment: for val in flat_segment:
segment_summ += val segment_summ += val
segment_mid = segment_summ / len(flat_segment) #посчитать нормально среднее значение/медиану segment_mid = segment_summ / len(flat_segment) #посчитать нормально среднее значение/медиану
for ind in range(1, len(flat_segment) - 1): for ind in range(1, len(flat_segment) - 1):
if flat_segment[ind + 1] > segment_mid and flat_segment[ind - 1] < segment_mid: if flat_segment[ind + 1] > segment_mid and flat_segment[ind - 1] < segment_mid:
flat_mid_index = ind # найти пересечение средней и графика, получить его индекс flat_mid_index = ind # найти пересечение средней и графика, получить его индекс
segment_mid_index = flat_mid_index - 5 segment_mid_index = flat_mid_index - 5
labeled_drop = data[segment_mid_index - 120 : segment_mid_index + 120] labeled_drop = data[segment_mid_index - 120 : segment_mid_index + 120]
labeled_min = min(labeled_drop) labeled_min = min(labeled_drop)
for value in labeled_drop: # обрезаем for value in labeled_drop: # обрезаем
value = value - labeled_min value = value - labeled_min
labeled_max = max(labeled_drop) labeled_max = max(labeled_drop)
for value in labeled_drop: # нормируем for value in labeled_drop: # нормируем
value = value / labeled_max value = value / labeled_max
convolve = scipy.signal.fftconvolve(labeled_drop, labeled_drop) convolve = scipy.signal.fftconvolve(labeled_drop, labeled_drop)
convolve_list.append(max(convolve)) # сворачиваем паттерн convolve_list.append(max(convolve)) # сворачиваем паттерн
# плюс надо впихнуть сюда логистическую сигмоиду и поиск альфы # плюс надо впихнуть сюда логистическую сигмоиду и поиск альфы
if len(confidences) > 0: if len(confidences) > 0:
self.confidence = min(confidences) self.confidence = min(confidences)
else: else:
self.confidence = 1.5 self.confidence = 1.5
if len(convolve_list) > 0: if len(convolve_list) > 0:
self.convolve_max = max(convolve_list) self.convolve_max = max(convolve_list)
else: else:
self.convolve_max = 120 # макс метрика свертки равна отступу(120), вау! self.convolve_max = 120 # макс метрика свертки равна отступу(120), вау!
def logistic_sigmoid(x1, x2, alpha, height): def logistic_sigmoid(x1, x2, alpha, height):
distribution = [] distribution = []
for i in range(x, y): for i in range(x, y):
F = 1 * height / (1 + math.exp(-i * alpha)) F = 1 * height / (1 + math.exp(-i * alpha))
distribution.append(F) distribution.append(F)
return distribution return distribution
async def predict(self, dataframe): async def predict(self, dataframe):
data = dataframe['value'] data = dataframe['value']
result = self.__predict(data) result = self.__predict(data)
result.sort() result.sort()
if len(self.segments) > 0: if len(self.segments) > 0:
result = [segment for segment in result if not is_intersect(segment, self.segments)] result = [segment for segment in result if not is_intersect(segment, self.segments)]
return result return result
def __predict(self, data): def __predict(self, data):
window_size = 24 window_size = 24
all_max_flatten_data = data.rolling(window=window_size).mean() all_max_flatten_data = data.rolling(window=window_size).mean()
extrema_list = [] extrema_list = []
# добавить все пересечения экспоненты со сглаженным графиком # добавить все пересечения экспоненты со сглаженным графиком
# #
for i in exponential_smoothing(data + self.confidence, 0.02): for i in exponential_smoothing(data + self.confidence, 0.02):
extrema_list.append(i) extrema_list.append(i)
segments = [] segments = []
for i in all_mins: for i in all_mins:
if all_max_flatten_data[i] > extrema_list[i]: if all_max_flatten_data[i] > extrema_list[i]:
segments.append(i - window_size) segments.append(i - window_size)
return [(x - 1, x + 1) for x in self.__filter_prediction(segments, all_max_flatten_data)] return [(x - 1, x + 1) for x in self.__filter_prediction(segments, all_max_flatten_data)]
def __filter_prediction(self, segments, all_max_flatten_data): def __filter_prediction(self, segments, all_max_flatten_data):
delete_list = [] delete_list = []
variance_error = int(0.004 * len(all_max_flatten_data)) variance_error = int(0.004 * len(all_max_flatten_data))
if variance_error > 200: if variance_error > 200:
variance_error = 200 variance_error = 200
for i in range(1, len(segments)): for i in range(1, len(segments)):
if segments[i] < segments[i - 1] + variance_error: if segments[i] < segments[i - 1] + variance_error:
delete_list.append(segments[i]) delete_list.append(segments[i])
for item in delete_list: for item in delete_list:
segments.remove(item) segments.remove(item)
# изменить секонд делит лист, сделать для свертки с сигмоидой # изменить секонд делит лист, сделать для свертки с сигмоидой
delete_list = [] delete_list = []
pattern_data = all_max_flatten_data[segments[0] - 120 : segments[0] + 120] pattern_data = all_max_flatten_data[segments[0] - 120 : segments[0] + 120]
for segment in segments: for segment in segments:
convol_data = all_max_flatten_data[segment - 120 : segment + 120] convol_data = all_max_flatten_data[segment - 120 : segment + 120]
conv = scipy.signal.fftconvolve(pattern_data, convol_data) conv = scipy.signal.fftconvolve(pattern_data, convol_data)
if max(conv) > self.convolve_max * 1.1 or max(conv) < self.convolve_max * 0.9: if max(conv) > self.convolve_max * 1.1 or max(conv) < self.convolve_max * 0.9:
delete_list.append(segment) delete_list.append(segment)
for item in delete_list: for item in delete_list:
segments.remove(item) segments.remove(item)
return segments return segments
def save(self, model_filename): def save(self, model_filename):
with open(model_filename, 'wb') as file: with open(model_filename, 'wb') as file:
pickle.dump((self.confidence, self.convolve_max), file) pickle.dump((self.confidence, self.convolve_max), file)
def load(self, model_filename): def load(self, model_filename):
try: try:
with open(model_filename, 'rb') as file: with open(model_filename, 'rb') as file:
(self.confidence, self.convolve_max) = pickle.load(file) (self.confidence, self.convolve_max) = pickle.load(file)
except: except:
pass pass

128
analytics/server.py
Unescape View File

@ -1,64 +1,64 @@
import config import config
import json import json
import logging import logging
import sys import sys
import asyncio import asyncio
import services import services
from analytic_unit_worker import AnalyticUnitWorker from analytic_unit_worker import AnalyticUnitWorker
root = logging.getLogger() root = logging.getLogger()
logger = logging.getLogger('SERVER') logger = logging.getLogger('SERVER')
worker = None worker = None
server_service = None server_service = None
data_service = None data_service = None
root.setLevel(logging.DEBUG) root.setLevel(logging.DEBUG)
ch = logging.StreamHandler(sys.stdout) ch = logging.StreamHandler(sys.stdout)
ch.setLevel(logging.DEBUG) ch.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] %(message)s") formatter = logging.Formatter("%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] %(message)s")
ch.setFormatter(formatter) ch.setFormatter(formatter)
root.addHandler(ch) root.addHandler(ch)
async def handle_task(text): async def handle_task(text):
try: try:
task = json.loads(text) task = json.loads(text)
logger.info("Command is OK") logger.info("Command is OK")
await server_service.send_message(json.dumps({ await server_service.send_message(json.dumps({
'_taskId': task['_taskId'], '_taskId': task['_taskId'],
'task': task['type'], 'task': task['type'],
'analyticUnitId': task['analyticUnitId'], 'analyticUnitId': task['analyticUnitId'],
'status': "in progress" 'status': "in progress"
})) }))
res = await worker.do_task(task) res = await worker.do_task(task)
res['_taskId'] = task['_taskId'] res['_taskId'] = task['_taskId']
await server_service.send_message(json.dumps(res)) await server_service.send_message(json.dumps(res))
except Exception as e: except Exception as e:
logger.error("Exception: '%s'" % str(e)) logger.error("Exception: '%s'" % str(e))
def init_services(): def init_services():
logger.info("Starting services...") logger.info("Starting services...")
logger.info("Server...") logger.info("Server...")
server_service = services.ServerService(handle_task) server_service = services.ServerService(handle_task)
logger.info("Ok") logger.info("Ok")
logger.info("Data service...") logger.info("Data service...")
data_service = services.DataService(server_service) data_service = services.DataService(server_service)
logger.info("Ok") logger.info("Ok")
return server_service, data_service return server_service, data_service
if __name__ == "__main__": if __name__ == "__main__":
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
logger.info("Starting worker...") logger.info("Starting worker...")
worker = AnalyticUnitWorker() worker = AnalyticUnitWorker()
logger.info("Ok") logger.info("Ok")
server_service, data_service = init_services() server_service, data_service = init_services()
loop.run_until_complete(server_service.handle_loop()) loop.run_until_complete(server_service.handle_loop())

4
analytics/services/__init__.py
Unescape View File

@ -1,2 +1,2 @@
from services.server_service import ServerService from services.server_service import ServerService
from services.data_service import DataService from services.data_service import DataService

16
analytics/services/data_service.py
Unescape View File

@ -1,9 +1,9 @@
class DataService: class DataService:
def __init__(self, server_service): def __init__(self, server_service):
self.server_service = server_service self.server_service = server_service
async def safe_file(filename, content): async def safe_file(filename, content):
pass pass
async def load_file(filename, content): async def load_file(filename, content):
pass pass

84
analytics/services/server_service.py
Unescape View File

@ -1,42 +1,42 @@
import config import config
import zmq import zmq
import zmq.asyncio import zmq.asyncio
import logging import logging
import asyncio import asyncio
logger = logging.getLogger('SERVER_SERVICE') logger = logging.getLogger('SERVER_SERVICE')
class ServerService: class ServerService:
def __init__(self, on_message_handler): def __init__(self, on_message_handler):
self.on_message_handler = on_message_handler self.on_message_handler = on_message_handler
logger.info("Binding to %s ..." % config.ZEROMQ_CONNECTION_STRING) logger.info("Binding to %s ..." % config.ZEROMQ_CONNECTION_STRING)
self.context = zmq.asyncio.Context() self.context = zmq.asyncio.Context()
self.socket = self.context.socket(zmq.PAIR) self.socket = self.context.socket(zmq.PAIR)
self.socket.bind(config.ZEROMQ_CONNECTION_STRING) self.socket.bind(config.ZEROMQ_CONNECTION_STRING)
async def handle_loop(self): async def handle_loop(self):
while True: while True:
received_bytes = await self.socket.recv() received_bytes = await self.socket.recv()
text = received_bytes.decode('utf-8') text = received_bytes.decode('utf-8')
if text == 'ping': if text == 'ping':
asyncio.ensure_future(self.__handle_ping()) asyncio.ensure_future(self.__handle_ping())
else: else:
asyncio.ensure_future(self.__handle_message(text)) asyncio.ensure_future(self.__handle_message(text))
async def send_message(self, string): async def send_message(self, string):
await self.socket.send_string(string) await self.socket.send_string(string)
async def __handle_ping(self): async def __handle_ping(self):
await self.socket.send(b'pong') await self.socket.send(b'pong')
async def __handle_message(self, text): async def __handle_message(self, text):
try: try:
asyncio.ensure_future(self.on_message_handler(text)) asyncio.ensure_future(self.on_message_handler(text))
except Exception as e: except Exception as e:
logger.error("Exception: '%s'" % str(e)) logger.error("Exception: '%s'" % str(e))

Write Preview
Loading…
Cancel
Save