Analysis-Services/ASJobGraphEvents/gantt/dgml.py

from dataclasses import dataclass
from typing import List, Set, Iterable, Optional, cast
from datetime import datetime
import xml.etree.ElementTree as ET

from structures import Job
from gantt_types import ThreadId

def get_jobEnd(elem: Job) -> datetime:
    return elem.end

def read_jobs(filename: str) -> List[Job]:
    jobs: List[Job] = []

    doc = ET.parse(filename)
    root = doc.getroot()

    try:
        nodes = [child for child in root if "nodes" in child.tag.lower()][0]
    except IndexError:
        return jobs

    for node in nodes:
        if job := parse_job_node(node):
            jobs.append(job)

    sortedJobs: List[Job] = []
    sortedJobs = sorted(jobs, key=get_jobEnd)
    return sortedJobs


def parse_iso(time: str) -> datetime:
    if time[-1].lower() == "z":
        time = time[:-1]

    return datetime.fromisoformat(time)


def parse_thread_id(s: str) -> ThreadId:
    return ThreadId(int(s))


def strip_newlines(s: str) -> str:

    return "".join([c for c in s if ord(c) > 32])


def parse_job_node(node: ET.Element) -> Optional[Job]:
    for attr, value in node.attrib.items():
        if attr == "StartedAt":
            start = parse_iso(value)
        if attr == "FinishedAt":
            end = parse_iso(value)
        if attr == "Label":
            name = value
        if attr == "Slot" or attr == "Thread":
            thread = value
        if attr == "Id":
            nodeId = value

    try:
        return Job(start, end, strip_newlines(name), parse_thread_id(thread), int(nodeId))
    except UnboundLocalError:
        # most likely doesn't include "Thread" or "Slot" attribute.
        return None