97 lines
2.1 KiB
Python
97 lines
2.1 KiB
Python
import json
|
|
from collections import defaultdict, Iterable
|
|
|
|
from log_analyzer import LogSettings
|
|
|
|
|
|
class Analyzer:
|
|
def __init__(self, settings: LogSettings):
|
|
self.settings = settings
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
raise NotImplementedError()
|
|
|
|
def result(self) -> Iterable:
|
|
raise NotImplementedError()
|
|
|
|
def name(self):
|
|
return self.__name__
|
|
|
|
|
|
class LocationAnalyzer(Analyzer):
|
|
"""
|
|
store spatial log entries
|
|
"""
|
|
entries = []
|
|
__name__ = "Location"
|
|
|
|
class Formats:
|
|
geojson = 0
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
|
|
def result(self) -> list:
|
|
return self.entries
|
|
|
|
def render(self, format: int = Formats.geojson):
|
|
if format is self.Formats.geojson:
|
|
return json.dumps([entry['location']['coordinates'] for entry in self.entries])
|
|
raise NotImplementedError()
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
if entry[self.settings.entry_type] in self.settings.spatials:
|
|
self.entries.append(entry)
|
|
return False
|
|
|
|
|
|
class LogEntryCountAnalyzer(Analyzer):
|
|
"""
|
|
count occurrences of log entry types
|
|
"""
|
|
__name__ = "LogEntryCount"
|
|
|
|
def result(self) -> defaultdict:
|
|
return self.store
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
self.store[entry[self.settings.entry_type]] += 1
|
|
return False
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
self.store = defaultdict(lambda: 0)
|
|
|
|
|
|
class LogEntrySequenceAnalyzer(Analyzer):
|
|
"""
|
|
store sequence of all log entry types
|
|
"""
|
|
__name__ = "LogEntrySequence"
|
|
|
|
def result(self) -> list:
|
|
return self.store
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
entry_type = entry[self.settings.entry_type]
|
|
self.store.append(entry_type)
|
|
return False
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
self.store = []
|
|
|
|
|
|
class ActionSequenceAnalyzer(LogEntrySequenceAnalyzer):
|
|
"""
|
|
find sequence of non-spatial log entry types
|
|
"""
|
|
__name__ = "ActionSequenceAnalyzer"
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
entry_type = entry[self.settings.entry_type]
|
|
if entry_type in self.settings.spatials:
|
|
return False
|
|
self.store.append(entry_type)
|
|
return False
|