51 lines
1.1 KiB
Python
51 lines
1.1 KiB
Python
import json
|
|
from collections import defaultdict
|
|
|
|
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) -> object:
|
|
raise NotImplementedError()
|
|
|
|
|
|
class LocationAnalyzer(Analyzer):
|
|
entries = []
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
|
|
def result(self) -> object:
|
|
return self.entries
|
|
|
|
def render(self, format="geojson"):
|
|
if format is "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):
|
|
def result(self) -> object:
|
|
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)
|
|
|
|
|