Analysis-Services/ASJobGraphEvents/gantt/dgml.py

66 lines
1.6 KiB
Python
Raw Normal View History

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
2020-11-18 08:12:05 +08:00
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] = []
2020-11-18 08:12:05 +08:00
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