# Copyright 2023 (c) Anna Schumaker. """Our tree model for selecting an xfstests run to view.""" import bisect import datetime import typing from gi.repository import GObject from gi.repository import Gio from gi.repository import Gtk from .. import sqlite class XfstestsRun(GObject.GObject): """A single XfstessRun.""" timestamp = GObject.Property(type=GObject.TYPE_PYOBJECT) runid = GObject.Property(type=int) ftime = GObject.Property(type=str) def __init__(self, runid: int, timestamp: datetime.datetime, ftime: str = "%T") -> None: """Initialize an XfstestsRun instance.""" super().__init__(timestamp=timestamp, runid=runid, ftime=ftime) def __gt__(self, rhs: typing.Self) -> bool: """Compare the timestamps an runids of two XfstestRun objects.""" return (self.timestamp, self.runid) > (rhs.timestamp, rhs.runid) def __lt__(self, rhs: typing.Self) -> bool: """Compare the timestamps and runids of two XfstestsRun objects.""" return (self.timestamp, self.runid) < (rhs.timestamp, rhs.runid) def __str__(self) -> str: """Get a string representation of this run.""" return f"#{self.runid}: {self.timestamp.strftime(self.ftime)}" class DeviceRunsList(GObject.GObject, Gio.ListModel): """Contains a single test device with multiple runs.""" name = GObject.Property(type=str) n_items = GObject.Property(type=int) def __init__(self, name: str) -> None: """Initialize our DeviceRunsList.""" super().__init__(name=name) self.__runs = [] def __lt__(self, rhs: typing.Self) -> bool: """Compare two DeviceRunsList objects.""" return self.name < rhs.name def __str__(self) -> str: """Get the name of this device.""" return self.name def do_get_item_type(self) -> GObject.GType: """Get the type of objects in the list.""" return XfstestsRun.__gtype__ def do_get_n_items(self) -> int: """Get the number of runs added to this device.""" return self.n_items def do_get_item(self, n: int) -> XfstestsRun: """Get the run at the nth position.""" return self.__runs[n] def add_run(self, runid: int, timestamp: datetime.datetime, ftime: str = "%T") -> None: """Add a run to the Device's list.""" bisect.insort(self.__runs, XfstestsRun(runid, timestamp, ftime)) self.n_items = len(self.__runs) def get_earliest_run(self) -> XfstestsRun | None: """Get the earliest XfstestsRun that we know about.""" return self.__runs[0] if self.n_items > 0 else None class DateDeviceList(GObject.GObject, Gio.ListModel): """A list containing test devices used on a specific day.""" date = GObject.Property(type=GObject.TYPE_PYOBJECT) n_items = GObject.Property(type=int) treemodel = GObject.Property(type=Gtk.TreeListModel) def __init__(self, sql: sqlite.Connection, date: datetime.date) -> None: """Initialize our DateDeviceList.""" super().__init__(date=date) devices = {} for row in sql("""SELECT DISTINCT runid, device, timestamp FROM tagged_runs WHERE timestamp >= ? AND timestamp < ?""", *self.get_range(date)).fetchall(): if (dev := devices.get(row['device'])) is None: devices[row['device']] = dev = DeviceRunsList(row['device']) ts = datetime.datetime.fromisoformat(row['timestamp']) dev.add_run(row['runid'], ts) self.__items = sorted(devices.values()) self.n_items = len(self.__items) self.treemodel = Gtk.TreeListModel.new(root=self, passthrough=False, autoexpand=True, create_func=self.__create_func) def __create_func(self, item: GObject.GObject) -> Gio.ListModel: if isinstance(item, DeviceRunsList): return item return None def do_get_item_type(self) -> GObject.GType: """Get the type of objects in the list.""" return DeviceRunsList.__gtype__ def do_get_n_items(self) -> int: """Get the number of items in the list.""" return self.n_items def do_get_item(self, n: int) -> DeviceRunsList: """Get a specific item in the list.""" return self.__items[n] def get_range(self, date: datetime.date) -> tuple[datetime.datetime, datetime.datetime]: """Get the minimum and maximum timestamps for the date.""" min = datetime.datetime.combine(date, datetime.time()) max = min + datetime.timedelta(days=1) return (min, max) class TagDeviceList(GObject.GObject, Gio.ListModel): """Contains a single tag with multiple devices and runs.""" name = GObject.Property(type=str) n_items = GObject.Property(type=int) def __init__(self, name: str) -> None: """Initialize our TagDeviceList.""" super().__init__(name=name) self.__devices = [] def __lt__(self, rhs: typing.Self) -> bool: """Compare two TagDeviceList objects.""" lhs_run = self.get_earliest_run() rhs_run = rhs.get_earliest_run() if lhs_run.runid == rhs_run.runid: return self.name > rhs.name return lhs_run > rhs_run def __str__(self) -> str: """Get the name of this tag.""" return self.name def do_get_item_type(self) -> GObject.GType: """Get the type of objects in the list.""" return DeviceRunsList.__gtype__ def do_get_n_items(self) -> int: """Get the number of devices added to this tag.""" return self.n_items def do_get_item(self, n: int) -> DeviceRunsList: """Get the device at the nth position.""" return self.__devices[n] def add_run(self, runid: int, device: str, timestamp: datetime.datetime) -> None: """Add a run to the Tag's list.""" pos = bisect.bisect_left(self.__devices, device, key=str) if pos < len(self.__devices) and self.__devices[pos].name == device: dev = self.__devices[pos] else: dev = DeviceRunsList(device) self.__devices.insert(pos, dev) dev.add_run(runid, timestamp, "%c") self.n_items = len(self.__devices) def get_earliest_run(self) -> XfstestsRun | None: """Get the earliest run added to the tag.""" runs = [d.get_earliest_run() for d in self.__devices] return min(runs, default=None)