Skip to content
Snippets Groups Projects
Commit 2677b763 authored by Oscar Gustafsson's avatar Oscar Gustafsson :bicyclist:
Browse files

Fix #163 and add documentation

parent 4ea4ce71
No related branches found
No related tags found
No related merge requests found
Pipeline #88801 passed
......@@ -11,6 +11,8 @@ from typing import Dict, List, Optional, Tuple, cast
import matplotlib.pyplot as plt
import numpy as np
from matplotlib.axes import Axes
from matplotlib.figure import Figure
from matplotlib.lines import Line2D
from matplotlib.patches import PathPatch, Polygon
from matplotlib.path import Path
......@@ -40,7 +42,20 @@ _SIGNAL_COLOR = tuple(c / 255 for c in SIGNAL_COLOR)
class Schedule:
"""Schedule of an SFG with scheduled Operations."""
"""
Schedule of an SFG with scheduled Operations.
Parameters
----------
sfg : SFG
The signal flow graph to schedule.
schedule_time : int, optional
The schedule time. If not provided, it will be determined by the scheduling algorithm.
cyclic : bool, default: False
If the schedule is cyclic.
scheduling_alg : {'ASAP'}, optional
The scheduling algorithm to use. Currently, only "ASAP" is supported.
"""
_sfg: SFG
_start_times: Dict[GraphID, int]
......@@ -82,7 +97,7 @@ class Schedule:
def start_time_of_operation(self, graph_id: GraphID) -> int:
"""
Get the start time of the operation with the specified by *graph_id*.
Return the start time of the operation with the specified by *graph_id*.
"""
if graph_id not in self._start_times:
raise ValueError(
......@@ -91,7 +106,7 @@ class Schedule:
return self._start_times[graph_id]
def get_max_end_time(self) -> int:
"""Returns the current maximum end time among all operations."""
"""Return the current maximum end time among all operations."""
max_end_time = 0
for graph_id, op_start_time in self._start_times.items():
op = cast(Operation, self._sfg.find_by_id(graph_id))
......@@ -103,6 +118,16 @@ class Schedule:
return max_end_time
def forward_slack(self, graph_id: GraphID) -> int:
"""
Parameters
----------
graph_id
Returns
-------
The number of time steps the operation with *graph_id* can ba moved forward in time.
"""
if graph_id not in self._start_times:
raise ValueError(
f"No operation with graph_id {graph_id} in schedule"
......@@ -137,6 +162,16 @@ class Schedule:
return ret
def backward_slack(self, graph_id: GraphID) -> int:
"""
Parameters
----------
graph_id
Returns
-------
The number of time steps the operation with *graph_id* can ba moved backward in time.
"""
if graph_id not in self._start_times:
raise ValueError(
f"No operation with graph_id {graph_id} in schedule"
......@@ -479,9 +514,11 @@ class Schedule:
return operation_gap + y_location * (operation_height + operation_gap)
def _plot_schedule(self, ax, operation_gap=None):
"""Draw the schedule."""
line_cache = []
def _draw_arrow(start, end, name="", laps=0):
"""Draw an arrow from *start* to *end*."""
if end[0] < start[0] or laps > 0: # Wrap around
if start not in line_cache:
line = Line2D(
......@@ -564,6 +601,7 @@ class Schedule:
def _draw_offset_arrow(
start, end, start_offset, end_offset, name="", laps=0
):
"""Draw an arrow from *start* to *end*, but with an offset."""
_draw_arrow(
[start[0] + start_offset[0], start[1] + start_offset[1]],
[end[0] + end_offset[0], end[1] + end_offset[1]],
......@@ -576,7 +614,7 @@ class Schedule:
ax.set_axisbelow(True)
ax.grid()
for graph_id, op_start_time in self._start_times.items():
ypos = -self._get_y_position(graph_id, operation_gap=operation_gap)
ypos = self._get_y_position(graph_id, operation_gap=operation_gap)
op = self._sfg.find_by_id(graph_id)
# Rewrite to make better use of NumPy
latency_coords, execution_time_coords = op.get_plot_coordinates()
......@@ -602,7 +640,7 @@ class Schedule:
for graph_id, op_start_time in self._start_times.items():
op = self._sfg.find_by_id(graph_id)
_, out_coords = op.get_io_coordinates()
source_ypos = -self._get_y_position(
source_ypos = self._get_y_position(
graph_id, operation_gap=operation_gap
)
......@@ -610,7 +648,7 @@ class Schedule:
for output_signal in output_port.signals:
dest_op = output_signal.destination.operation
dest_start_time = self._start_times[dest_op.graph_id]
dest_ypos = -self._get_y_position(
dest_ypos = self._get_y_position(
dest_op.graph_id, operation_gap=operation_gap
)
(
......@@ -633,35 +671,72 @@ class Schedule:
# Get operation with maximum position
max_pos_graph_id = max(self._y_locations, key=self._y_locations.get)
yposmin = -self._get_y_position(
max_pos_graph_id, operation_gap=operation_gap
) - (OPERATION_GAP if operation_gap is None else operation_gap)
ax.axis([-1, self._schedule_time + 1, yposmin, 1])
yposmax = (
self._get_y_position(max_pos_graph_id, operation_gap=operation_gap)
+ 1
+ (OPERATION_GAP if operation_gap is None else operation_gap)
)
ax.axis([-1, self._schedule_time + 1, yposmax, 0]) # Inverted y-axis
ax.xaxis.set_major_locator(MaxNLocator(integer=True))
ax.add_line(
Line2D([0, 0], [yposmin, 1], linestyle="--", color="black")
Line2D([0, 0], [0, yposmax], linestyle="--", color="black")
)
ax.add_line(
Line2D(
[self._schedule_time, self._schedule_time],
[yposmin, 1],
[0, yposmax],
linestyle="--",
color="black",
)
)
def _reset_y_locations(self):
"""Reset all the y-locations in the schedule to None"""
self._y_locations = self._y_locations = defaultdict(lambda: None)
def plot_schedule(self, operation_gap=None) -> None:
def plot_in_axes(self, ax: Axes, operation_gap: float = None) -> None:
"""
Plot the schedule in a :class:`matplotlib.axes.Axes` or subclass.
Parameters
----------
ax : matplotlib.axes.Axes
The :class:`matplotlib.axes.Axes` to plot in.
operation_gap : float, optional
The vertical distance between operations in the schedule. The height of the operation is always 1.
"""
def plot(self, operation_gap: float = None) -> None:
"""
Plot the schedule. Will display based on the current Matplotlib backend.
Parameters
----------
operation_gap : float, optional
The vertical distance between operations in the schedule. The height of the operation is always 1.
"""
self._get_figure(operation_gap=operation_gap).show()
def _get_figure(self, operation_gap=None) -> None:
def _get_figure(self, operation_gap: float = None) -> Figure:
"""
Create a Figure and an Axes and plot schedule in the Axes.
Parameters
----------
operation_gap : float, optional
The vertical distance between operations in the schedule. The height of the operation is always 1.
Returns
-------
The Matplotlib Figure.
"""
fig, ax = plt.subplots()
self._plot_schedule(ax, operation_gap=operation_gap)
return fig
def _repr_svg_(self):
"""Generate an SVG of the schedule. This is automatically displayed in e.g. Jupyter Qt console.
"""
fig, ax = plt.subplots()
self._plot_schedule(ax)
f = io.StringIO()
......
......@@ -170,7 +170,7 @@ class MainWindow(QMainWindow, Ui_MainWindow):
# TODO: remove
if self.schedule is None:
return
self.schedule.plot_schedule()
self.schedule.plot()
if self._graph is not None:
print(f"filtersChildEvents(): {self._graph.filtersChildEvents()}")
# self._printButtonPressed('callback_pushButton()')
......
......@@ -78,10 +78,30 @@ class GraphIDGenerator:
class SFG(AbstractOperation):
"""
Signal flow graph.
Construct an SFG given its inputs and outputs.
Contains a set of connected operations, forming a new operation.
Used as a base for simulation, scheduling, etc.
Parameters
----------
inputs
outputs
input_signals
output_signals
id_number_offset
name
input_sources
Inputs/outputs may be specified using either Input/Output operations
directly with the inputs/outputs parameters, or using signals with the
input_signals/output_signals parameters. If signals are used, the
corresponding Input/Output operations will be created automatically.
The id_number_offset parameter specifies what number graph IDs will be
offset by for each new graph component type. IDs start at 1 by default,
so the default offset of 0 will result in IDs like "c1", "c2", etc.
while an offset of 3 will result in "c4", "c5", etc.
"""
_components_by_id: Dict[GraphID, GraphComponent]
......@@ -110,20 +130,6 @@ class SFG(AbstractOperation):
Sequence[Optional[SignalSourceProvider]]
] = None,
):
"""
Construct an SFG given its inputs and outputs.
Inputs/outputs may be specified using either Input/Output operations
directly with the inputs/outputs parameters, or using signals with the
input_signals/output_signals parameters. If signals are used, the
corresponding Input/Output operations will be created automatically.
The id_number_offset parameter specifies what number graph IDs will be
offset by for each new graph component type. IDs start at 1 by default,
so the default offset of 0 will result in IDs like "c1", "c2", etc.
while an offset of 3 will result in "c4", "c5", etc.
"""
input_signal_count = 0 if input_signals is None else len(input_signals)
input_operation_count = 0 if inputs is None else len(inputs)
output_signal_count = (
......@@ -311,7 +317,7 @@ class SFG(AbstractOperation):
)
def __str__(self) -> str:
"""Get a string representation of this SFG."""
"""Return a string representation of this SFG."""
string_io = StringIO()
string_io.write(super().__str__() + "\n")
string_io.write("Internal Operations:\n")
......@@ -329,7 +335,7 @@ class SFG(AbstractOperation):
self, *src: Optional[SignalSourceProvider], name: Name = Name("")
) -> "SFG":
"""
Get a new independent SFG instance that is identical to this SFG
Return a new independent SFG instance that is identical to this SFG
except without any of its external connections.
"""
return SFG(
......@@ -359,6 +365,19 @@ class SFG(AbstractOperation):
bits_override: Optional[int] = None,
truncate: bool = True,
) -> Number:
"""
Parameters
----------
index
input_values
results
delays
prefix
bits_override
truncate
"""
if index < 0 or index >= self.output_count:
raise IndexError(
"Output index out of range (expected"
......@@ -476,6 +495,18 @@ class SFG(AbstractOperation):
return self
def inputs_required_for_output(self, output_index: int) -> Iterable[int]:
"""
Return which inputs that the output depends on.
Parameters
----------
output_index : int
The output index.
Returns
-------
A list of inputs that are required to compute the output with the given *output_index*.
"""
if output_index < 0 or output_index >= self.output_count:
raise IndexError(
"Output index out of range (expected"
......
......@@ -10,6 +10,12 @@ from typing import Callable, Sequence
class SignalGenerator:
"""
Base class for signal generators.
Handles operator overloading and defined the ``__call__`` method that should be overridden.
"""
def __call__(self, time: int) -> complex:
raise NotImplementedError
......
......@@ -42,4 +42,4 @@ sfg.set_execution_time_of_type(Addition.type_name(), 1)
# Create schedule
schedule = Schedule(sfg, cyclic=True)
schedule.plot_schedule()
schedule.plot()
......@@ -53,4 +53,4 @@ sfg.set_execution_time_of_type(Subtraction.type_name(), 1)
# %%
# Generate schedule
schedule = Schedule(sfg, cyclic=True)
schedule.plot_schedule()
schedule.plot()
......@@ -10,8 +10,8 @@
inkscape:version="1.2.2 (b0a8486541, 2022-12-01)"
sodipodi:docname="logo.svg"
inkscape:export-filename="icon_logo.png"
inkscape:export-xdpi="5.3303123"
inkscape:export-ydpi="5.3303123"
inkscape:export-xdpi="5.1594224"
inkscape:export-ydpi="5.1594224"
xmlns:inkscape="http://www.inkscape.org/namespaces/inkscape"
xmlns:sodipodi="http://sodipodi.sourceforge.net/DTD/sodipodi-0.dtd"
xmlns="http://www.w3.org/2000/svg"
......@@ -28,7 +28,7 @@
inkscape:document-units="mm"
showgrid="false"
inkscape:zoom="0.4204826"
inkscape:cx="1055.9295"
inkscape:cx="511.31723"
inkscape:cy="709.89857"
inkscape:window-width="1200"
inkscape:window-height="1896"
......
test/baseline/test__get_figure_no_execution_times.png

25.6 KiB | W: | H:

test/baseline/test__get_figure_no_execution_times.png

25.8 KiB | W: | H:

test/baseline/test__get_figure_no_execution_times.png
test/baseline/test__get_figure_no_execution_times.png
test/baseline/test__get_figure_no_execution_times.png
test/baseline/test__get_figure_no_execution_times.png
  • 2-up
  • Swipe
  • Onion skin
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment