scheduler_item.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""B-ASIC Scheduler-gui Graphics Graph Item Module.

Contains the scheduler-gui SchedulerItem class for drawing and
maintain a component in a graph.
"""
from collections import defaultdict
from math import floor
from pprint import pprint
from typing import Dict, List, Optional, Set, cast

# QGraphics and QPainter imports
from qtpy.QtWidgets import QGraphicsItem, QGraphicsItemGroup

# B-ASIC
from b_asic._preferences import OPERATION_GAP
from b_asic.operation import Operation
from b_asic.port import InputPort
from b_asic.schedule import Schedule
from b_asic.scheduler_gui.axes_item import AxesItem
from b_asic.scheduler_gui.operation_item import OperationItem
from b_asic.scheduler_gui.scheduler_event import SchedulerEvent
from b_asic.scheduler_gui.signal_item import SignalItem


class SchedulerItem(SchedulerEvent, QGraphicsItemGroup):  # PySide2 / PyQt5
    """
    A class to represent a graph in a QGraphicsScene. This class is a
    subclass of QGraphicsItemGroup and contains the objects, axes from
    AxesItem, as well as components from OperationItem. It
    also inherits from SchedulerEvent, which acts as a filter for events
    to OperationItem objects.
    """

    _axes: Optional[AxesItem]
    _components: List[OperationItem]
    _x_axis_indent: float
    _event_items: List[QGraphicsItem]
    _signal_dict: Dict[OperationItem, Set[SignalItem]]

    def __init__(
        self, schedule: Schedule, parent: Optional[QGraphicsItem] = None
    ):
        """Constructs a SchedulerItem. 'parent' is passed to QGraphicsItemGroup's constructor.
        """
        # QGraphicsItemGroup.__init__(self, self)
        # SchedulerEvent.__init__(self)
        super().__init__(parent=parent)
        # if isinstance(parent, QGraphicsItem):
        #     super().__init__(parent=parent)
        # else:
        #     super().__init__(parent=self)
        self._schedule = schedule
        self._axes = None
        self._components = []
        self._x_axis_indent = 0.2
        self._event_items = []
        self._signal_dict = defaultdict(set)
        self._make_graph()

    def clear(self) -> None:
        """Sets all children's parent to 'None' and delete the children objects.
        """
        self._event_items = []
        for item in self.childItems():
            item.setParentItem(None)
            del item

    def is_component_valid_pos(self, item: OperationItem, pos: float) -> bool:
        """Takes in a component position and returns true if the component's new
        position is valid, false otherwise."""
        # TODO: implement
        assert self.schedule is not None, "No schedule installed."
        end_time = item.end_time
        new_start_time = floor(pos) - floor(self._x_axis_indent)
        slacks = self.schedule.slacks(item.op_id)
        op_start_time = self.schedule.start_time_of_operation(item.op_id)
        if not -slacks[0] <= new_start_time - op_start_time <= slacks[1]:
            # Cannot move due to dependencies
            return False
        if self.schedule.cyclic:
            if new_start_time < -1:
                # Moving one position before left edge => wrap
                return False
            if new_start_time > self.schedule.schedule_time + 1:
                # Moving one position after schedule_time => wrap
                return False
        else:
            if pos < 0:
                return False
            if new_start_time + end_time > self.schedule.schedule_time:
                return False

        return True

    def _redraw_lines(self, item: OperationItem):
        """Update lines connected to *item*."""
        for signal in self._signal_dict[item]:
            signal.update_path()

    def set_item_active(self, item: OperationItem):
        item.set_active()
        for signal in self._signal_dict[item]:
            signal.set_active()

    def set_item_inactive(self, item: OperationItem):
        item.set_inactive()
        for signal in self._signal_dict[item]:
            signal.set_inactive()

    def set_new_starttime(self, item: OperationItem) -> None:
        """Set new starttime for *item*."""
        pos = item.x()
        op_start_time = self.schedule.start_time_of_operation(item.op_id)
        new_start_time = floor(pos) - floor(self._x_axis_indent)
        move_time = new_start_time - op_start_time
        if move_time:
            self.schedule.move_operation(item.op_id, move_time)

    def is_valid_delta_time(self, delta_time: int) -> bool:
        """Takes in a delta time and returns true if the new schedule time is
        valid, false otherwise."""
        # TODO: implement
        # item = self.scene().mouseGrabberItem()
        assert self.schedule is not None, "No schedule installed."
        return (
            self.schedule.schedule_time + delta_time
            >= self.schedule.get_max_end_time()
        )

    def set_schedule_time(self, delta_time: int) -> None:
        """Set the schedule time and redraw the graph."""
        if self._axes is None:
            raise RuntimeError("No AxesItem!")
        assert self.schedule is not None, "No schedule installed."
        self.schedule.set_schedule_time(
            self.schedule.schedule_time + delta_time
        )
        self._axes.set_width(self._axes.width + delta_time)
        # Redraw all lines
        for signals in self._signal_dict.values():
            for signal in signals:
                signal.update_path()

    @property
    def schedule(self) -> Schedule:
        """The schedule."""
        return self._schedule

    @property
    def axes(self) -> Optional[AxesItem]:
        return self._axes

    @property
    def components(self) -> List[OperationItem]:
        return self._components

    @property
    def event_items(self) -> List[QGraphicsItem]:
        """Return a list of objects that receives events."""
        return self._event_items

    def _make_graph(self) -> None:
        """Makes a new graph out of the stored attributes."""
        # build components
        _components_dict = {}
        # print('Start times:')
        for op_id, op_start_time in self.schedule.start_times.items():
            operation = cast(Operation, self.schedule.sfg.find_by_id(op_id))
            component = OperationItem(operation, parent=self)
            component.setPos(
                self._x_axis_indent + op_start_time,
                self.schedule._get_y_position(op_id),
            )
            self._components.append(component)
            _components_dict[operation] = component
            self._event_items += component.event_items

        # build axes
        schedule_time = self.schedule.schedule_time
        max_pos_op_id = max(
            self.schedule._y_locations, key=self.schedule._y_locations.get
        )
        yposmin = self.schedule._get_y_position(max_pos_op_id)

        self._axes = AxesItem(schedule_time, yposmin + 0.5)
        self._axes.setPos(0, yposmin + 1 + OPERATION_GAP)
        self._event_items += self._axes.event_items
        # self._axes.width = schedule_time

        # add axes and components
        self.addToGroup(self._axes)
        for component in self._components:
            self.addToGroup(component)
        # self.addToGroup(self._components)

        # add signals
        for component in self._components:
            for output_port in component.operation.outputs:
                for signal in output_port.signals:
                    destination = cast(InputPort, signal.destination)
                    dest_component = _components_dict[destination.operation]
                    gui_signal = SignalItem(
                        component, dest_component, signal, parent=self
                    )
                    self.addToGroup(gui_signal)
                    self._signal_dict[component].add(gui_signal)
                    self._signal_dict[dest_component].add(gui_signal)


pprint(SchedulerItem.__mro__)