231 lines
7.4 KiB
Python
231 lines
7.4 KiB
Python
import logging
|
|
from collections import defaultdict, namedtuple, OrderedDict
|
|
from types import SimpleNamespace
|
|
from typing import List, NamedTuple
|
|
|
|
from util import json_path, combinate
|
|
from util.download import download_board
|
|
from . import Result, LogSettings, Analyzer, ResultStore
|
|
from .default import CategorizerStub, Store
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BoardDurationAnalyzer(Analyzer):
|
|
"""
|
|
calculate display duration of boards
|
|
"""
|
|
__name__ = "BoardDuration"
|
|
|
|
def render(self) -> str:
|
|
return "\n".join(["{}\t{}".format(entry["active"], entry["id"]) for entry in self.result().get()])
|
|
|
|
def result(self, store: ResultStore) -> None:
|
|
result = []
|
|
last_timestamp = None
|
|
last_board = None
|
|
for board in self.store:
|
|
board_id, timestamp = board["id"], board["timestamp"]
|
|
|
|
if not last_timestamp is None:
|
|
result.append(self.save_entry(last_board, last_timestamp, (timestamp - last_timestamp)/1000))
|
|
last_timestamp = timestamp
|
|
last_board = board_id
|
|
# TODO: last board?
|
|
store.add(Result(type(self), result))
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
entry_type = entry[self.settings.type_field]
|
|
if entry_type in self.settings.boards:
|
|
self.store.append(self.save_entry(entry["board_id"], entry["timestamp"])) # TODO: make configurable
|
|
return False
|
|
|
|
def save_entry(self, board_id: str, timestamp: int, active: int = None):
|
|
entry = {"id": board_id, "timestamp": timestamp}
|
|
if not active is None:
|
|
entry["active"] = active
|
|
return entry
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
self.store = []
|
|
self.last = {}
|
|
|
|
|
|
class SimulationRoundsAnalyzer(Analyzer):
|
|
__name__ = "SimuRounds"
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
self.store = defaultdict(lambda: -1) # TODO verify
|
|
|
|
def result(self, store: ResultStore) -> None:
|
|
store.add(Result(type(self), dict(self.store)))
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
entry_type = entry[self.settings.type_field]
|
|
if entry_type in self.settings.custom['simulation_rounds']:
|
|
if entry["answers"][self.settings.type_field] in self.settings.custom["simu_data"]:
|
|
simu_id = entry['answers']["@id"]
|
|
self.store[simu_id] += 1
|
|
return False
|
|
|
|
|
|
class ActivationSequenceAnalyzer(Analyzer):
|
|
__name__ = "ActivationSequence"
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
self.store = []
|
|
|
|
def result(self, store: ResultStore) -> None:
|
|
store.add(Result(type(self), self.store))
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
if entry[self.settings.type_field] in self.settings.sequences['start']:
|
|
if entry['cache']:
|
|
self.store.append(entry['cache']['@id'])
|
|
else:
|
|
logger.error("null cache")
|
|
return False
|
|
|
|
|
|
class BiogamesCategorizer(CategorizerStub):
|
|
__name__ = "BiogamesCategorizer"
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
if self.key is "default":
|
|
if entry[self.settings.type_field] in self.settings.custom['instance_start']:
|
|
self.key = entry[self.settings.custom['instance_id']]
|
|
return False
|
|
|
|
|
|
class ActivityMapper(Analyzer):
|
|
__name__ = "ActivityMapper"
|
|
|
|
def __init__(self, settings: LogSettings) -> None:
|
|
super().__init__(settings)
|
|
self.store: List[self.State] = []
|
|
self.instance_config_id: str = None
|
|
self.filters = SimpleNamespace()
|
|
self.filters.start = lambda entry: combinate(self.settings.custom["sequences2"]["start"], entry)
|
|
self.filters.end = lambda entry: combinate(self.settings.custom["sequences2"]["end"], entry)
|
|
|
|
self.State: NamedTuple = namedtuple("State", ["sequence", "events", "track", "timestamp"])
|
|
|
|
def result(self, store: ResultStore) -> None:
|
|
instance_config_id = self.instance_config_id
|
|
for active_segment in self.store: # active_segment → sequence or None (None → map active)
|
|
seq_data_url = "/game2/editor/config/{config_id}/sequence/{sequence_id}/".format(
|
|
config_id=instance_config_id,
|
|
sequence_id=active_segment.sequence,
|
|
)
|
|
source = self.settings.source
|
|
seq_data = source._get(seq_data_url)
|
|
if not seq_data.ok:
|
|
logger.error("HTTP ERROR:", seq_data)
|
|
seq_data = {}
|
|
else:
|
|
seq_data = seq_data.json()
|
|
# TODO: use sequence names
|
|
logger.warning(seq_data)
|
|
for event in active_segment.events:
|
|
if event[self.settings.type_field] in self.settings.boards:
|
|
sequence_id = active_segment.sequence
|
|
board_id = event["board_id"]
|
|
local_file = download_board(board_id, instance_config_id, sequence_id, source)
|
|
if local_file is not None:
|
|
event["image"] = local_file[16:]
|
|
store.add(Result(type(self), {"instance": instance_config_id, "store": [x._asdict() for x in self.store]}))
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
if self.instance_config_id is None:
|
|
if entry[self.settings.type_field] in self.settings.custom['instance_start']:
|
|
self.instance_config_id = json_path(entry, self.settings.custom['instance_config_id'])
|
|
if self.filters.start(entry):
|
|
self.store.append(
|
|
self.State(
|
|
sequence=json_path(entry, json_path(self.settings.custom, "sequences2.id_field")),
|
|
events=[],
|
|
track=[],
|
|
timestamp=entry['timestamp']))
|
|
elif self.filters.end(entry) or not self.store:
|
|
self.store.append(self.State(sequence=None, events=[], track=[], timestamp=entry['timestamp']))
|
|
|
|
if entry[self.settings.type_field] in self.settings.spatials:
|
|
self.store[-1].track.append(
|
|
{
|
|
'timestamp': entry['timestamp'],
|
|
'coordinates': json_path(entry, "location.coordinates"),
|
|
'accuracy': entry['accuracy']
|
|
}
|
|
)
|
|
else:
|
|
self.store[-1].events.append(entry)
|
|
return False
|
|
|
|
|
|
class BiogamesStore(Store):
|
|
__name__ = "BiogamesStore"
|
|
|
|
def result(self, store: ResultStore) -> None:
|
|
result = OrderedDict()
|
|
for event in self.store:
|
|
if event[self.settings.type_field] in self.settings.boards:
|
|
sequence_id = json_path(event, json_path(self.settings.custom, "sequences2.id_field"))
|
|
board_id = event["board_id"]
|
|
local_file = download_board(
|
|
board_id=board_id,
|
|
instance_config_id=json_path(self.store[0], self.settings.custom["instance_config_id"]),
|
|
sequence_id=sequence_id,
|
|
source=self.settings.source)
|
|
if local_file is not None:
|
|
event["image"] = local_file[16:]
|
|
result[event["timestamp"]] = event
|
|
|
|
store.add(Result(type(self), result))
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
self.store.append(entry)
|
|
return False
|
|
|
|
|
|
class InstanceConfig(Analyzer):
|
|
__name__ = "InstanceConfig"
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
self.store = {}
|
|
|
|
def process(self, entry: dict):
|
|
if entry[self.settings.type_field] in self.settings.custom["instance_start"]:
|
|
print(entry)
|
|
self.store["instance_id"] = json_path(entry, self.settings.custom["instance_config_id"])
|
|
|
|
def result(self, store: ResultStore):
|
|
store.add(Result(type(self), dict(self.store)))
|
|
|
|
|
|
class SimulationOrderAnalyzer(Analyzer):
|
|
__name__ = "SimuOrder"
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
self.store = defaultdict(lambda: -1) # TODO verify
|
|
self.order = []
|
|
|
|
def result(self, store: ResultStore) -> None:
|
|
store.add(Result(type(self), [self.store[sim] for sim in self.order]))
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
entry_type = entry[self.settings.type_field]
|
|
if entry_type in self.settings.custom['simulation_rounds']:
|
|
if entry["answers"][self.settings.type_field] in self.settings.custom["simu_data"]:
|
|
simu_id = entry['answers']["@id"]
|
|
self.store[simu_id] += 1
|
|
if not simu_id in self.order:
|
|
self.order.append(simu_id)
|
|
return False |