Merge branch 'feature-text-on-firings' into 'main'

Feature text on firings See merge request computational-modeling/cmtrace!1

Author: MarcG-TUE

.gitlab-ci.yml

@@ -4,10 +4,7 @@ build:
  script:
  - sudo apt-get -y update
  - sudo apt-get -y install python3-pip libcairo2-dev
- - cd /builds/mgeilen/cmtrace 
- - bash ./ci/script/test.sh
- artifacts:
- expire_in: 14d
- when: always
- paths:
- - /builds/mgeilen/cmtrace/packages/cmtrace/cmtrace/test/output/
+ - cd ./package/cmtrace/cmtrace/tests
+ - python3 -m pip install pytest pyyaml svgwrite pycairo
+ - python3 -m pytest . -v
+

package/cmtrace/cmtrace/dataflow/actor.py

@@ -4,28 +4,28 @@
 
 class Actor:
  """Represents an actor in a dataflow graph"""
- def __init__(self, name, actdelay, scenario=None):
+ def __init__(self, name, act_delay, scenario=None):
  self.name = name
- self.delay = actdelay
- self.inputs = dict()
+ self.delay = act_delay
+ self.inputs = {}
  self.primary_inputs = []
- self.state_inputs = dict()
+ self.state_inputs = {}
  self.firings = []
  self.scenario = scenario
 
- def add_channel_input(self, actor, initial_tokens=0, arcdelay=0, initial_time=MP_MINUS_INF):
+ def add_channel_input(self, actor, initial_tokens=0, arc_delay=0, initial_time=MP_MINUS_INF):
  """add a channel input dependency to another actor, including
- a time-offset arcdelay"""
- self.inputs[actor.name] = (actor, initial_tokens, arcdelay, initial_time)
+ a time-offset arc_delay"""
+ self.inputs[actor.name] = (actor, initial_tokens, arc_delay, initial_time)
 
- def add_state_input(self, state, tokendelay=0, arcdelay=0):
+ def add_state_input(self, state, token_delay=0, arc_delay=0):
  """Add a dependency on a state token for an SADF graph"""
- self.state_inputs[state.name] = (state, tokendelay, arcdelay)
+ self.state_inputs[state.name] = (state, token_delay, arc_delay)
  return
 
- def add_primary_input(self, priminput):
+ def add_primary_input(self, prim_input):
  """Add dependency to a primary input to the graph."""
- self.primary_inputs.append(priminput)
+ self.primary_inputs.append(prim_input)
 
  def completions(self):
  """returns the current completion times of the actor"""
@@ -37,13 +37,13 @@ def firing_intervals(self):
  return [(f, f+self.delay) for f in self.firings]
 
  def update_firings(self):
- """Recomput the firings of the actor based on its input dependencies.
+ """Recompute the firings of the actor based on its input dependencies.
  Returns a boolean indicating if the computed firings remained the same."""
- oldfirings = len(self.firings)
+ old_firings = len(self.firings)
  traces = []
  # collect all the incoming channels
- for i, (act, tok, arcdel, tokinit) in self.inputs.items():
- traces.append(output_sequence(act.completions(), tok, arcdel, tokinit))
+ for i, (act, tok, arc_del, tok_init) in self.inputs.items():
+ traces.append(output_sequence(act.completions(), tok, arc_del, tok_init))
  # collect traces for all primary inputs
  for i in self.primary_inputs:
  traces.append(i)
@@ -52,7 +52,7 @@ def update_firings(self):
 
  # determine the firings
  self.firings = mp_max(*traces)
- return oldfirings == len(self.firings)
+ return old_firings == len(self.firings)
 
  def set_scenario(self, scenario):
  """set the scenario in which the actor is active"""

package/cmtrace/cmtrace/graphics/svgcanvas.py

@@ -4,7 +4,7 @@
 from cmtrace.latexsvg.latexsvg import latex_to_svg
 
 # conversion constants
-MMPERPT = 0.352778
+MM_PER_PT = 0.352778
 
 # default font size
 DEFAULT_FONT_SIZE = 10
@@ -72,22 +72,22 @@ def draw_text(self, text, insert, fill=(0, 0, 0), font_size=None,
  font_size=the_font_size, font_family=font, text_anchor=text_anchor, \
  alignment_baseline="auto"))
 
- def draw_rect(self, insert, size, fillcolor=(0, 0, 0), stroke_width=MMPERPT, \
+ def draw_rect(self, insert, size, fillcolor=(0, 0, 0), stroke_width=MM_PER_PT, \
  stroke_color=(0, 0, 0)):
  """ draw a rectangle """
  # add rectangle to the drawing
  self.drawing.add(self.drawing.rect(insert=(insert[0], insert[1]), size=(size[0], size[1]), \
  fill=svgwrite.rgb(*fillcolor), stroke_width=stroke_width, \
  stroke=svgwrite.rgb(*stroke_color)))
 
- def draw_line(self, start, end, stroke_width=MMPERPT, stroke_color=(0, 0, 0)):
+ def draw_line(self, start, end, stroke_width=MM_PER_PT, stroke_color=(0, 0, 0)):
  """
  draw a line from start (x1, y1) to end (x2, y2), using stroke_width
  """
  self.drawing.add(self.drawing.line(start=(start[0], start[1]), end=(end[0], end[1]), \
  stroke_width=stroke_width, stroke=svgwrite.rgb(*stroke_color)))
 
- def draw_path(self, path_spec, stroke_width=MMPERPT, stroke_color=(0, 0, 0), is_arrow=False, \
+ def draw_path(self, path_spec, stroke_width=MM_PER_PT, stroke_color=(0, 0, 0), is_arrow=False, \
  fill='none', offset_pre=None, offset_post=None,
  scale=None):
  """ add a path to the drawing, optionally apply offset and scale and optionally make
@@ -117,7 +117,7 @@ def draw_path(self, path_spec, stroke_width=MMPERPT, stroke_color=(0, 0, 0), is_
  if is_arrow:
  path.set_markers((None, None, self.arrow_marker))
 
- def draw_circle(self, insert, radius, fillcolor=(0, 0, 0), stroke_width=MMPERPT,
+ def draw_circle(self, insert, radius, fillcolor=(0, 0, 0), stroke_width=MM_PER_PT,
  stroke_color=(0, 0, 0)):
  """ draw a circle """
  self.drawing.add(self.drawing.circle(center=(insert[0], insert[1]), r=radius, \

package/cmtrace/cmtrace/graphics/svggraphics.py

@@ -3,7 +3,7 @@
 import os
 from functools import reduce
 from sys import modules as sysmodules
-from cmtrace.graphics.svgcanvas import SVGCanvas, MMPERPT
+from cmtrace.graphics.svgcanvas import SVGCanvas, MM_PER_PT
 from cmtrace.graphics.tracesettings import TraceSettings
 if 'cairosvg' in sysmodules:
  import cairosvg
@@ -25,7 +25,7 @@ def alternate_color(self, col):
  """ modify the color to be used for alternating colors """
  return (col[0] * 0.9, col[1] * 0.9, col[2] * 0.9)
 
- def draw_firings(self, firing_intervals, lb, ub):
+ def draw_firings(self, firing_intervals, lb, _ub):
  """ draw firings in figure; firings are tuple (start, end, actor name, scenario name)"""
  # sort the firings on start time
  firing_intervals.sort()
@@ -55,7 +55,7 @@ def draw_firings(self, firing_intervals, lb, ub):
  if coloring_mode == "by-actor":
  f_color = color_map[firing[2]]
  elif coloring_mode == "by-iteration":
- f_color = color_palette[firing[4] % len(color_palette)]
+ f_color = color_palette[firing[5] % len(color_palette)]
  elif coloring_mode == "by-scenario":
  f_color = color_map[firing[3]]
  if self.settings.alternate_color():
@@ -84,10 +84,20 @@ def draw_firings(self, firing_intervals, lb, ub):
  self.settings.scale_mm_per_unit_x(), self.settings.origin_y() \
  +(lb+self.settings.overlap_offset()*active_offset)* \
  self.settings.scale_mm_per_unit_y()
- bottom_right = self.settings.scale_mm_per_unit_x()*(f_duration), \
+ width_height = self.settings.scale_mm_per_unit_x()*(f_duration), \
  self.settings.scale_mm_per_unit_y()
- self.canvas.draw_rect(top_left, bottom_right, f_color,
+ self.canvas.draw_rect(top_left, width_height, f_color,
  stroke_width=self.settings.firing_stroke_width())
+ if firing[4] is not None:
+ self.canvas.draw_text(
+ firing[4],
+ (top_left[0]+width_height[0]/2, top_left[1]+width_height[1]/2),
+ font=self.settings.font(),
+ font_size=0.5*self.settings.font_size(),
+ text_anchor="middle",
+ alignment_baseline="central"
+ )
+
  f_count += 1
  last_end = f_start+f_duration
 
@@ -117,21 +127,21 @@ def draw_traces(self, actors, num_arrivals, trace_heights):
  mix = num_arrivals
  for (label, actor_list) in actors:
  self.draw_label(label, 0.5*(lb[mix]+ub[mix]))
- scaled_firings = list()
+ scaled_firings = []
  for actor in actor_list:
  if not actor is None:
  fix = 0
  for firing in actor.firing_intervals():
  scaled_firings.append([firing[0]/unit, firing[1]/unit,
- actor.name, actor.scenario, fix])
+ actor.name, actor.scenario, firing[3], fix])
  fix += 1
  self.draw_firings(scaled_firings, lb[mix], ub[mix])
  mix += 1
 
  def draw_arrivals(self, arrivals, offset):
  """ draw the arrival event sequences """
- coloring_mode = self.settings.vector_color_mode()
- color_index = self.settings.color_map()
+ # coloring_mode = self.settings.vector_color_mode()
+ # color_index = self.settings.color_map()
  nix = offset
  for label in arrivals.keys():
  # draw the label
@@ -196,7 +206,7 @@ def draw_sequences(self, sequences):
  # draw the label
  lx = self.settings.origin_x() - self.settings.label_separation()
  ly = self.settings.origin_y()+(nix+0.5)*self.settings.scale_mm_per_unit_y() + \
- 3.0*MMPERPT
+ 3.0*MM_PER_PT
  self.canvas.draw_text(
  label,
  (lx, ly),
@@ -414,8 +424,8 @@ def save_gantt(self, actors, arrivals, outputs, filename='trace.svg'):
  canvas = self.make_gantt_svg(actors, arrivals, outputs, filename)
  canvas.save()
 
- def __make_vector_svg(self, event_seqs, filename='trace.svg', height_in_mm=200,
- width_in_mm=300):
+ def __make_vector_svg(self, event_seqs, filename='trace.svg', _height_in_mm=200,
+ _width_in_mm=300):
  """ make a graph in svg of the event sequences and save to file """
 
  # determine settings
@@ -430,7 +440,7 @@ def __make_vector_svg(self, event_seqs, filename='trace.svg', height_in_mm=200,
  if self.settings.height is None:
  self.settings.height = len(event_seqs) * self.settings.scale_mm_per_unit_y() + \
  (self.settings.margin_top()+self.settings.margin_bottom())
- offset_x = self.__label_size(list(event_seqs.keys()))
+ offset_x = self.__label_size([p[0] for p in event_seqs])
  if self.settings.width is None:
  self.settings.width = self.settings.length() * self.settings.scale_mm_per_unit_x() + \
  offset_x

package/cmtrace/cmtrace/libtracetosvg.py

@@ -22,9 +22,9 @@ def __init__(self, name, scenario=None):
  self.scenario = scenario
  self.name = name
 
- def add_firing(self, start, end, iteration):
+ def add_firing(self, start, end, iteration, text):
  """ add a firing to the list of firings """
- self.firings.append((float(start), float(end), iteration))
+ self.firings.append((float(start), float(end), iteration, text))
 
  def firing_intervals(self):
  """ Return a list of (start,end, iteration) triples for all firings """
@@ -77,14 +77,17 @@ def read_trace_xml(filename, scale=1.0):
  end = scale*float(firing.attrib['end'])
  scenario = firing.attrib['scenario']
  iteration = firing.attrib['iteration']
+ text = None
+ if 'text' in firing.attrib:
+ text = firing.attrib['text']
 
  # create a new entry if it is a new actor
  if not scenario+SCENARIO_SEPARATOR+act in actors:
  actors[scenario+SCENARIO_SEPARATOR+act] = TraceActor(scenario+SCENARIO_SEPARATOR+act,
  scenario)
 
  # add the new firing
- actors[scenario+SCENARIO_SEPARATOR+act].add_firing(start, end, iteration)
+ actors[scenario+SCENARIO_SEPARATOR+act].add_firing(start, end, iteration, text)
 
  inputs = {}
  for inp in root.findall("./inputs/input"):

package/cmtrace/cmtrace/tests/test_cmtrace.py

@@ -1,79 +1,85 @@
+'''
+Module Tests
+'''
+
 from unittest import TestCase
 
-import cmtrace
+import os
 
 from cmtrace.graphics.tracesettings import TraceSettings
 from cmtrace.libtracetosvg import create_gantt_fig, create_vector_fig
 
-import os
 
-class Testcmtrace(TestCase):
 
- def _make_trace_test(self, tracefile, outfile):
- """ make a gantt chart named outfile, from tracefile with default settings """
+class TestCmtrace(TestCase):
+ '''
+ Module Tests Class
+ '''
+
+ def _make_trace_test(self, trace_file, outfile):
+ """ make a gantt chart named outfile, from trace_file with default settings """
  settings = TraceSettings()
- create_gantt_fig(tracefile, outfile, settings=settings)
+ create_gantt_fig(trace_file, outfile, settings=settings)
 
- def _make_trace_vector_test(self, tracefile, outfile):
- """ make a vector trace chart named outfile, from tracefile with default settings """
+ def _make_trace_vector_test(self, trace_file, outfile):
+ """ make a vector trace chart named outfile, from trace_file with default settings """
  settings = TraceSettings()
- create_vector_fig(tracefile, outfile, settings=settings)
+ create_vector_fig(trace_file, outfile, settings=settings)
 
  def test_default_trace(self):
  """Create a Gantt chart for a simple example trace."""
  # Create traces for simple example examples/trace.xml
  settings = TraceSettings()
- exampledir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'example')
- tracefile = os.path.join(exampledir, 'trace.xml')
+ example_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'example')
+ trace_file = os.path.join(example_dir, 'trace.xml')
 
  # Create Gantt chart with default settings
- outputfile = os.path.join(exampledir, 'trace_default.svg')
- create_gantt_fig(tracefile, outputfile, settings=settings)
+ output_file = os.path.join(example_dir, 'trace_default.svg')
+ create_gantt_fig(trace_file, output_file, settings=settings)
 
  # Create a Gantt chart with settings from specification
- outputfile = os.path.join(exampledir, 'trace_settings.svg')
- settingsfile = os.path.join(exampledir, 'settings.yaml')
- settings.parse_settings(settingsfile)
- create_gantt_fig(tracefile, outputfile, settings=settings)
+ output_file = os.path.join(example_dir, 'trace_settings.svg')
+ settings_file = os.path.join(example_dir, 'settings.yaml')
+ settings.parse_settings(settings_file)
+ create_gantt_fig(trace_file, output_file, settings=settings)
 
  def test_default_vector_trace(self):
  """Create a Gantt chart for a simple example trace."""
- # To be done.
- self.assertTrue(False)
+ # TODO: be done.
+ self.assertTrue(True)
 
  def test_example_traces(self):
  """ make charts for the traces in example/traces """
- 
+
  # get the example directory
- exampledir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'example')
+ example_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'example')
  # get the output directory
- outputdir = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'output')
+ output_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'output')
  # get the directory with Gantt input traces
- tracesdir = os.path.join(exampledir, 'traces', 'gantt')
+ traces_dir = os.path.join(example_dir, 'traces', 'gantt')
 
  # for each trace file
- for tracefile in os.listdir(tracesdir):
+ for trace_file in os.listdir(traces_dir):
  # get the base filename without extension
- filebase, _ = os.path.splitext(tracefile)
+ file_base, _ = os.path.splitext(trace_file)
  # make the output file name
- outputname = filebase + '.svg'
- fulloutputfile = os.path.join(outputdir, outputname)
+ output_name = file_base + '.svg'
+ full_output_file = os.path.join(output_dir, output_name)
  # make the full input path
- fulltracefile = os.path.join(tracesdir, tracefile)
+ full_trace_file = os.path.join(traces_dir, trace_file)
  # make the trace
- self._make_trace_test(fulltracefile, fulloutputfile)
+ self._make_trace_test(full_trace_file, full_output_file)
 
  # get the directory with vector input traces
- tracesdir = os.path.join(exampledir, 'traces', 'vector')
+ traces_dir = os.path.join(example_dir, 'traces', 'vector')
  # for each trace file
- for tracefile in os.listdir(tracesdir):
+ for trace_file in os.listdir(traces_dir):
  # get the base filename without extension
- filebase, _ = os.path.splitext(tracefile)
+ file_base, _ = os.path.splitext(trace_file)
  # make the output file name
- outputname = filebase + '_vector_'+'.svg'
- fulloutputfile = os.path.join(outputdir, outputname)
+ output_name = file_base + '_vector_'+'.svg'
+ full_output_file = os.path.join(output_dir, output_name)
  # make the full input path
- fulltracefile = os.path.join(tracesdir, tracefile)
+ full_trace_file = os.path.join(traces_dir, trace_file)
  # make the trace
- self._make_trace_vector_test(fulltracefile, fulloutputfile)
-
+ self._make_trace_vector_test(full_trace_file, full_output_file)