<commit_before>import time
from collections import defaultdict
from datetime import datetime

HEARTBEAT_EXPIRE = 120 # Heartbeats must be at most 2 minutes apart.


class MonitorState(object):

    def __init__(self):
        self.hearts = {}
        self.tasks = {}
        self.task_events = defaultdict(lambda: [])
        self.workers = defaultdict(lambda: [])

    def tasks_by_type(self):
        t = defaultdict(lambda: [])
        for id, events in self.task_events.items():
            try:
                task_type = self.tasks[id]["name"]
            except KeyError:
                pass
            else:
                t[task_type].append(id)
        return t

    def get_task_info(self, task_id):
        task_info = dict(self.tasks[task_id])

        task_events = []
        for event in self.task_events[task_id]:
            if event["state"] in ("task-failed", "task-retried"):
                task_info["exception"] = event["exception"]
                task_info["traceback"] = event["traceback"]
            elif event["state"] == "task-succeeded":
                task_info["result"] = event["result"]
            task_events.append({event["state"]: event["when"]})
        task_info["events"] = task_events

        return task_info

    def receive_task_event(self, event):
        event["state"] = event.pop("type")
        event["when"] = self.timestamp_to_isoformat(event["timestamp"])
        self.task_events[event["uuid"]].append(event)

    def timestamp_to_isoformat(self, timestamp):
        return datetime.fromtimestamp(timestamp).isoformat()

    def receive_heartbeat(self, event):
        self.hearts[event["hostname"]] = event["timestamp"]

    def receive_task_received(self, event):
        task_info = dict(event)
        event = dict(event)
        task_info.pop("type")
        event["state"] = event.pop("type")
        event["when"] = self.timestamp_to_isoformat(event["timestamp"])
        self.tasks[task_info["uuid"]] = task_info
        self.task_events[event["uuid"]].append(event)

    def list_workers(self):
        alive_workers = []
        for hostname, events in self.workers.items():
            if events[-1]["state"] == "worker-online":
                alive_workers.append({hostname: events[-1]["when"]})
        return alive_workers

    def list_worker_tasks(self, hostname):
        alive_workers = self.list_workers()
        tasks_for_worker = defaultdict(lambda: [])
        for hostname, when in alive_workers.items():
            for task_id, task_info in self.tasks:
                if task_info["hostname"] == hostname:
                    tasks_for_worker[hostname].append(task_id)
        return tasks_for_worker

    def receive_worker_event(self, event):
        event["state"] = event.pop("type")
        event["when"] = self.timestamp_to_isoformat(event["timestamp"])
        self.workers[event["hostname"]].append(event)

    def worker_is_alive(self, hostname):
        last_worker_event = self.workers[hostname][-1]
        if last_worker_event and last_worker_event == "worker-online":
            time_of_last_heartbeat = self.hearts[hostname]<commit_msg>Fixed issue with worker tasks not being displayed<commit_after>import time
from collections import defaultdict
from datetime import datetime

HEARTBEAT_EXPIRE = 120 # Heartbeats must be at most 2 minutes apart.


class MonitorState(object):

    def __init__(self):
        self.hearts = {}
        self.tasks = {}
        self.task_events = defaultdict(lambda: [])
        self.workers = defaultdict(lambda: [])

    def tasks_by_type(self):
        t = defaultdict(lambda: [])
        for id, events in self.task_events.items():
            try:
                task_type = self.tasks[id]["name"]
            except KeyError:
                pass
            else:
                t[task_type].append(id)
        return t

    def get_task_info(self, task_id):
        task_info = dict(self.tasks[task_id])

        task_events = []
        for event in self.task_events[task_id]:
            if event["state"] in ("task-failed", "task-retried"):
                task_info["exception"] = event["exception"]
                task_info["traceback"] = event["traceback"]
            elif event["state"] == "task-succeeded":
                task_info["result"] = event["result"]
            task_events.append({event["state"]: event["when"]})
        task_info["events"] = task_events

        return task_info

    def receive_task_event(self, event):
        event["state"] = event.pop("type")
        event["when"] = self.timestamp_to_isoformat(event["timestamp"])
        self.task_events[event["uuid"]].append(event)

    def timestamp_to_isoformat(self, timestamp):
        return datetime.fromtimestamp(timestamp).isoformat()

    def receive_heartbeat(self, event):
        self.hearts[event["hostname"]] = event["timestamp"]

    def receive_task_received(self, event):
        task_info = dict(event)
        event = dict(event)
        task_info.pop("type")
        event["state"] = event.pop("type")
        event["when"] = self.timestamp_to_isoformat(event["timestamp"])
        self.tasks[task_info["uuid"]] = task_info
        self.task_events[event["uuid"]].append(event)

    def list_workers(self):
        alive_workers = []
        for hostname, events in self.workers.items():
            if events[-1]["state"] == "worker-online":
                alive_workers.append({hostname: events[-1]["when"]})
        return alive_workers

    def list_worker_tasks(self, hostname):
        alive_workers = self.list_workers()
        tasks_for_worker = defaultdict(lambda: [])
        for worker in alive_workers:
            try:
                hostname, when = worker.items()[0]
            except IndexError:
                continue
            for task_id, task_info in self.tasks.items():
                if task_info["hostname"] == hostname:
                    tasks_for_worker[hostname].append(task_id)
        return tasks_for_worker

    def receive_worker_event(self, event):
        event["state"] = event.pop("type")
        event["when"] = self.timestamp_to_isoformat(event["timestamp"])
        self.workers[event["hostname"]].append(event)

    def worker_is_alive(self, hostname):
        last_worker_event = self.workers[hostname][-1]
        if last_worker_event and last_worker_event == "worker-online":
            time_of_last_heartbeat = self.hearts[hostname]
{'fraction_non_alphanumeric': 0.12492231199502797, 'fraction_numerical': 0.0023306401491609697, 'mean_word_length': 2.592075892857143, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7353636', 'n_tokens_mistral': 1866, 'n_tokens_neox': 1825, 'n_words': 352}