339 lines
11 KiB
Python
339 lines
11 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, get_board_data
|
|
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))
|
|
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 TypedBoardDuration(Analyzer):
|
|
__name__ = "BoardDuration"
|
|
|
|
def result(self, store: ResultStore) -> None:
|
|
pass
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
entry_type = entry[self.settings.type_field]
|
|
if entry_type in self.settings.boards:
|
|
pass
|
|
|
|
def add_board(self, entry):
|
|
board_data = get_board_data(self.settings.source, )
|
|
|
|
def add_location(self, entry):
|
|
self.track['coordinates'].append(json_path(entry, self.settings.custom['coordinates']))
|
|
|
|
def add_track(self, **props):
|
|
self.track['properties'] = props
|
|
self.tracks.append(self.track)
|
|
self.track = dict(self.template)
|
|
|
|
def __init__(self, settings: LogSettings):
|
|
super().__init__(settings)
|
|
self.last_board = {}
|
|
self.tracks = []
|
|
self.template = {"type": "LineString", "coordinates": [], "properties": {}}
|
|
self.track = dict(self.template)
|
|
|
|
|
|
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): # TODO: refactor
|
|
__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"
|
|
classes = {
|
|
"sequence.simulation.": "simu",
|
|
"sequence.question.": "question",
|
|
"error": "error"
|
|
}
|
|
|
|
def __init__(self, settings: LogSettings) -> None:
|
|
super().__init__(settings)
|
|
self.store: List[self.State] = []
|
|
self.timeline = []
|
|
self.last_board = {}
|
|
self.last_board_type = "other"
|
|
self.last_coordinate = None
|
|
self.tracks = []
|
|
self.track = None
|
|
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_old(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).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 result(self, store: ResultStore) -> None:
|
|
|
|
store.add(Result(type(self), {"instance": self.instance_config_id, "track": self.tracks, "boards": self.timeline}))
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
if self.track is None:
|
|
self.track = self.new_track(entry['timestamp'])
|
|
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'])
|
|
|
|
self.update_board_type(entry)
|
|
if entry[self.settings.type_field] in self.settings.spatials:
|
|
self.add_location(entry)
|
|
elif entry[self.settings.type_field] in self.settings.boards:
|
|
board_data = get_board_data(self.settings.source, self.instance_config_id, entry["sequence_id"],
|
|
entry["board_id"])
|
|
entry["extra_data"] = board_data
|
|
entry['coordinate'] = self.new_coordinate()
|
|
self.timeline.append(entry)
|
|
return False
|
|
|
|
def update_board_type(self, entry):
|
|
type = self.classify_entry(entry)
|
|
if not type == self.last_board_type:
|
|
self.add_track(activity_type=self.last_board_type,end_timestamp=entry['timestamp'])
|
|
self.last_board_type = type
|
|
|
|
def classify_entry(self, entry):
|
|
entry_type = entry[self.settings.type_field]
|
|
if self.filters.end(entry):
|
|
return "map"
|
|
if not entry_type in self.settings.boards:
|
|
return self.last_board_type
|
|
board_data = get_board_data(self.settings.source, self.instance_config_id, entry["sequence_id"],
|
|
entry["board_id"])
|
|
for pattern in self.classes:
|
|
if pattern in board_data['class']:
|
|
return self.classes[pattern]
|
|
if board_data['has_video']:
|
|
return "video"
|
|
elif board_data['has_audio']:
|
|
return "audio"
|
|
elif board_data['has_image']:
|
|
return "image"
|
|
return "other"
|
|
|
|
def new_coordinate(self):
|
|
return {"type": "Point", "coordinates": self.last_coordinate}
|
|
|
|
def add_location(self, entry):
|
|
coordinates = json_path(entry, self.settings.custom['coordinates'])
|
|
self.track['coordinates'].append(coordinates)
|
|
self.last_coordinate = coordinates
|
|
|
|
def add_track(self, **props):
|
|
self.track['properties'].update(props)
|
|
self.tracks.append(self.track)
|
|
self.track = self.new_track(props['end_timestamp'])
|
|
|
|
def new_track(self, timestamp):
|
|
return {"type": "LineString", "coordinates": [], "properties": {'start_timestamp': timestamp}}
|
|
|
|
|
|
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
|
|
|
|
|
|
class SimulationCategorizer(CategorizerStub): # TODO: refactor categorizer
|
|
__name__ = "SimulationCategorizer"# TODO: rename -.- (InstanceConfigIDCategorizer)
|
|
|
|
def process(self, entry: dict) -> bool:
|
|
if self.key is "default":
|
|
if entry[self.settings.type_field] in self.settings.custom['instance_start']:
|
|
try:
|
|
self.key = json_path(entry, self.settings.custom['instance_config_id'])
|
|
except KeyError as e:
|
|
print(entry)
|
|
raise e
|
|
return False
|
|
|
|
|
|
class SimulationFlagsAnalyzer(Analyzer):
|
|
__name__ = "SimuFlags"
|
|
|
|
def __init__(self, settings: LogSettings) -> None:
|
|
super().__init__(settings)
|
|
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"]:
|
|
self.store.append(entry)
|
|
return False
|
|
|
|
def result(self, store: ResultStore, name=None) -> None:
|
|
store.add(Result(type(self), self.store, name=name))
|
|
|
|
|