84 lines
1.9 KiB
Python
84 lines
1.9 KiB
Python
import json
|
|
from collections import defaultdict
|
|
|
|
from . import Result, LogSettings, Analyzer
|
|
|
|
|
|
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) -> Result:
|
|
# return self.entries
|
|
return Result(type(self), 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.type_field] in self.settings.spatials:
|
|
self.entries.append(entry)
|
|
return False
|
|
|
|
|
|
class LogEntryCountAnalyzer(Analyzer):
|
|
"""
|
|
count occurrences of log entry types
|
|
"""
|
|
__name__ = "LogEntryCount"
|
|
|
|
def result(self) -> Result:
|
|
return Result(type(self), dict(self.store))
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
self.store[entry[self.settings.type_field]] += 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) -> Result:
|
|
return Result(type(self), self.store)
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
entry_type = entry[self.settings.type_field]
|
|
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.type_field]
|
|
if entry_type in self.settings.spatials:
|
|
return False
|
|
self.store.append(entry_type)
|
|
return False
|