import json import logging from typing import List import analyzers from analyzers import get_renderer, Analyzer, render, Store from analyzers.analyzer import ResultStore from analyzers.settings import LogSettings, load_settings from loaders import LOADERS logging.basicConfig(format='%(levelname)s %(name)s:%(message)s', level=logging.DEBUG) log: logging.Logger = logging.getLogger(__name__) requests_log = logging.getLogger('requests') requests_log.setLevel(logging.WARN) def process_log(log_id: str, settings: LogSettings) -> List[Analyzer]: logfile: str = "data/inst_{id}.{format}".format(id=log_id, format=settings.log_format) loader = LOADERS[settings.log_format]() try: loader.load(logfile) except BaseException as e: raise RuntimeError(e) analyzers: List[Analyzer] = [] log.debug("build analyzers") for analyzer in settings.analyzers: analyzers.append(analyzer(settings)) log.debug("process entries") for entry in loader.get_entry(): for analyzer in analyzers: if analyzer.process(entry): break return analyzers if __name__ == '__main__': settings: LogSettings = load_settings("biogames2.json") log_ids: List[str] = [ "20d4244719404ffab0ca386c76e4b112", "56d9b64144ab44e7b90bf766f3be32e3", "dc2cdc28ca074715b905e4aa5badff10", "e32b16998440475b994ab46d481d3e0c", ] log_ids: List[str] = [ #"34fecf49dbaca3401d745fb467", # "44ea194de594cd8d63ac0314be", # "57c444470dbf88605433ca935c", # "78e0c545b594e82edfad55bd7f", # "91abfd4b31a5562b1c66be37d9", "597b704fe9ace475316c345903", "e01a684aa29dff9ddd9705edf8", "fbf9d64ae0bdad0de7efa3eec6", # "fe1331481f85560681f86827ec", "fe1331481f85560681f86827ec"] #"fec57041458e6cef98652df625", ] store: ResultStore = ResultStore() for log_id in log_ids: for analysis in process_log(log_id, settings): log.info("* Result for " + analysis.name()) # print(analysis.result()) # print(analysis.render()) analysis.result(store) if False: for r in get_renderer(analyzers.LocomotionActionAnalyzer): r().render(store.get_all()) if False: render(analyzers.LocationAnalyzer, store.get_all()) #print(json.dumps(store.serializable(), indent=1)) if False: render(analyzers.ActivityMapper, store.get_all()) render(analyzers.ProgressAnalyzer, store.get_all()) if False: from analyzers.postprocessing import graph g = graph.Cache(settings) g.run(store) if True: render(analyzers.SimulationOrderAnalyzer, store.get_all()) # for analyzers in analyzers: # if analyzers.name() in ["LogEntryCount", "ActionSequenceAnalyzer"]: # print(json.dumps(analyzers.result(), indent=2)) # for analyzers in analyzers: # if analyzers.name() in ["BoardDuration"]: # print(json.dumps(analyzers.result(), indent=2)) # print(analyzers.render()) # coords = analyzers[1].render() # with open("test.js", "w") as out: # out.write("coords = "+coords)