7288074417

Committed 21 Dec 2023 12:23PM UTC coverage: 95.206% (+0.05%) from 95.156%

Build # 7288074417

Build Type

push

github

Committed by

christophevg

Commit Message

exposed custom class and boundary on conditional Item

Run Details

6 of 6 new or added lines in 2 files covered. (100.0%)

5 existing lines in 1 file now uncovered.

1569 of 1648 relevant lines covered (95.21%)

7.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.88

/bpmn_tools/builder/process.py

"""

Process builder builds single-process, forward branching-flow-like structures:


                           | -> [ Task ] ------------------- |
  S -> [ Task ] -> < Branch >                               < > -> [ Task ] -> S
                           | -> < Branch > ----------- < > - | 
                                        | -> [ Task ] - |   
""" 

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Type

from bpmn_tools import flow

from bpmn_tools.collaboration import Collaboration, Participant
from bpmn_tools.diagrams      import Diagram, Plane
from bpmn_tools.notation      import Definitions
from bpmn_tools.colors        import Red, Green, Orange, Blue

from bpmn_tools.util import slugify

PADDING               =  10
FLOW_SPACE            =  20
MODEL_OFFSET_X        = 150
MODEL_OFFSET_Y        =  80
DEFAULT_BRANCH_HEIGHT =  75
STANDARD_TASK_HEIGHT  =  80

def NoColor(bpmn):
  return bpmn

class Color(Enum):
  NONE   : NoColor
  RED    : Red
  GREEN  : Green
  ORANGE : Orange
  BLUE   : Blue

@dataclass
class Step():
  children: List["Step"] = field(default_factory=list)
  label   : str   = None
  color   : Color = NoColor
  
  @property
  def height(self):
    raise NotImplementedError

  @property
  def width(self):
    raise NotImplementedError

  def render(self, x=0, y=0):
    raise NotImplementedError

  def shape(self, cls, **kwargs):
    return self.color(cls(**kwargs))

  def connect(self, source, target, label=None):
    if source == target:
      raise ValueError("connect: {source} == {target}")
    return flow.Flow(id=f"flow_{source.id}_{target.id}", source=source, target=target, name=label)

@dataclass
class Task(Step):
  name    : str = ""
  cls     : Type[flow.Task] = flow.Task
  args    : Dict = field(default_factory=dict)
  boundary: Type[flow.EventDefinition] = None

  tasks = 0 # class

  @classmethod
  def reset(cls):
    cls.tasks = 0
    
  def __post_init__(self):
    self.args["name"] = self.name
    self.element = self.shape(self.cls, id=f"task_{self.tasks}", **self.args)
    self.__class__.tasks += 1
  
  @property
  def height(self):
    return self.element.height + PADDING * 2

  @property
  def width(self):
    return self.element.width + PADDING * 2

  def render(self, x=0, y=0):
    self.element.x = x + PADDING
    self.element.y = y + PADDING
    shapes = [ self.element ]
    if self.boundary:
      shapes.append(flow.BoundaryEvent(
        id=f"boundary-event-{self.element.id}",
        definition=self.boundary(id=f"{slugify(self.boundary.__name__)}-{self.element.id}"),
        on=self.element
      ))
    return (shapes, [])

  @property
  def root(self):
    return self.element

  @property
  def tail(self):
    return self.element

@dataclass
class Process(Step):
  name   : str  = ""
  starts : bool = False
  ends   : bool = False
  _root  = None
  _tail  = None

  def __post_init__(self):
    if self.starts:
      self._root = self.shape(flow.Start, id="start")
    if self.ends:
      self._tail = self.shape(flow.End, id="end")

  def add(self, step):
    self.children.append(step)
    return self

  def extend(self, steps):
    for step in steps:
      self.add(step)
    return self

  @property
  def height(self):
    return max([ step.height for step in self.children ])

  @property
  def width(self):
    total_width = sum([ step.width for step in self.children ])
    total_width += FLOW_SPACE * (len(self.children) - 1)
    if self.starts:
      total_width += self.root.width + FLOW_SPACE
    if self.ends:
      total_width += self.tail.width + FLOW_SPACE
    return total_width 

  def render(self, x=None, y=None):
    # if we're called without positioning, we're adding diagram boilerplate
    wrap = x is None or y is None
    if wrap:
      x = 50 + MODEL_OFFSET_X # header of participant
      y = 0  + MODEL_OFFSET_Y
    
    shapes = []
    flows  = []
    prev = None

    # add optional start event
    if self.starts:
      self.root.x = x
      x += self.root.width
      self.root.y = y + int((STANDARD_TASK_HEIGHT + PADDING * 2)/2) - (self.root.height/2)
      shapes.append(self.root)
      x += FLOW_SPACE

    # add steps
    for step in self.children:
      more_shapes, more_flows = step.render(x=x, y=y)
      shapes.extend(more_shapes)
      flows.extend(more_flows)
      x += step.width + FLOW_SPACE
      if prev:
        flows.append(self.connect(source=prev.tail, target=step.root))
      elif self.starts:
        flows.append(self.connect(source=self.root, target=step.root))
      prev = step

    # add optional end event
    if self.ends:
      self.tail.x = x
      self.tail.y = self.root.y
      shapes.append(self.tail)
      flows.append(self.connect(source=shapes[-2], target=self.tail))

    # optinally wrap it all
    if wrap:
      process = flow.Process(id="process").extend(shapes).extend(flows)
      participant = Participant(self.name, process, id="participant")
      # participant shapes the lane
      participant.x = MODEL_OFFSET_X
      participant.y = MODEL_OFFSET_Y
      participant.width  = self.width + 50 + PADDING
      participant.height = self.height
      collaboration = Collaboration(id="collaboration").append(participant)
      return Definitions(id="definitions").extend([
        process,
        collaboration,
        Diagram(
          id="diagram",
          plane=Plane(id="plane", element=collaboration)
        )
      ])
    return (shapes, flows)

  @property
  def root(self):
    if self._root:
      return self._root
    return self.children[0].root

  @property
  def tail(self):
    if self._tail:
      return self._tail
    return self.children[-1].tail

class BranchKind(Enum):
  XOR = flow.ExclusiveGateway
  OR  = flow.InclusiveGateway
  AND = flow.ParallelGateway

# utility wrapper for an even more fluid API ;-)
def If(condition, step):
  return (step, condition)

@dataclass
class Branch(Step):
  default  : str = None
  kind     : BranchKind = BranchKind.XOR

  gws = 0

  def __post_init__(self):
    global gws
    self.root = self.shape(self.kind.value, id=f"gateway_start_{self.gws}", name=self.label)
    self.tail = self.shape(self.kind.value, id=f"gateway_end_{self.gws}")
    self.__class__.gws += 1
    # ensure all children to ensure they are tuples
    self.children = list(map(lambda c: c if type(c) is tuple else (c,None), self.children))

  @classmethod
  def reset(cls):
    cls.gws = 0

  def add(self, branch, condition=None):
    self.children.append((branch, condition))
    return self

  def extend(self, steps):
    for step in steps:
      self.add(step)
    return self

  @property
  def height(self):
    total_height = sum([ branch.height for branch, _ in self.children ])
    if self.default:
      total_height += DEFAULT_BRANCH_HEIGHT
    return total_height

  @property
  def width(self):
    total_width = max([ branch.width for branch, _ in self.children ])
    total_width += self.root.width + self.tail.width
    total_width += FLOW_SPACE * 2
    return total_width

  def render(self, x=0, y=0):
    shapes = []
    flows  = []
    self.root.x = x
    self.root.y = y + int((STANDARD_TASK_HEIGHT + PADDING * 2)/2) - (self.root.height/2)
    shapes.append(self.root)
    shapes.append(self.tail)
    x += FLOW_SPACE
    if self.default:
      flows.append(self.connect(source=self.root, target=self.tail))
      y += DEFAULT_BRANCH_HEIGHT
    self.tail.x = x + self.width - self.tail.width - FLOW_SPACE
    self.tail.y = self.root.y
    x += self.root.width
    for branch, condition in self.children:
      more_shapes, more_flows = branch.render(x=x, y=y)
      shapes.extend(more_shapes)
      flows.extend(more_flows)
      y += branch.height
      flows.append(self.connect(source=self.root,   target=branch.root, label=condition))
      flows.append(self.connect(source=branch.tail, target=self.tail))
    return shapes, flows

1	"""	4✔
2
3	Process builder builds single-process, forward branching-flow-like structures:
4
5
6	\| -> [ Task ] ------------------- \|
7	S -> [ Task ] -> < Branch > < > -> [ Task ] -> S
8	\| -> < Branch > ----------- < > - \|
9	\| -> [ Task ] - \|
10	"""
11
12	from dataclasses import dataclass, field	8✔
13	from enum import Enum	8✔
14	from typing import List, Dict, Type	8✔
15
16	from bpmn_tools import flow	8✔
17
18	from bpmn_tools.collaboration import Collaboration, Participant	8✔
19	from bpmn_tools.diagrams import Diagram, Plane	8✔
20	from bpmn_tools.notation import Definitions	8✔
21	from bpmn_tools.colors import Red, Green, Orange, Blue	8✔
22
23	from bpmn_tools.util import slugify	8✔
24
25	PADDING = 10	8✔
26	FLOW_SPACE = 20	8✔
27	MODEL_OFFSET_X = 150	8✔
28	MODEL_OFFSET_Y = 80	8✔
29	DEFAULT_BRANCH_HEIGHT = 75	8✔
30	STANDARD_TASK_HEIGHT = 80	8✔
31
32	def NoColor(bpmn):	8✔
33	return bpmn	8✔
34
35	class Color(Enum):	8✔
36	NONE : NoColor	8✔
37	RED : Red	8✔
38	GREEN : Green	8✔
39	ORANGE : Orange	8✔
40	BLUE : Blue	8✔
41
42	@dataclass	8✔
43	class Step():	8✔
44	children: List["Step"] = field(default_factory=list)	8✔
45	label : str = None	8✔
46	color : Color = NoColor	8✔
47
48	@property	8✔
49	def height(self):	8✔
UNCOV 50	raise NotImplementedError	×
51
52	@property	8✔
53	def width(self):	8✔
UNCOV 54	raise NotImplementedError	×
55
56	def render(self, x=0, y=0):	8✔
UNCOV 57	raise NotImplementedError	×
58
59	def shape(self, cls, **kwargs):	8✔
60	return self.color(cls(**kwargs))	8✔
61
62	def connect(self, source, target, label=None):	8✔
63	if source == target:	8✔
UNCOV 64	raise ValueError("connect: {source} == {target}")	×
65	return flow.Flow(id=f"flow_{source.id}_{target.id}", source=source, target=target, name=label)	8✔
66
67	@dataclass	8✔
68	class Task(Step):	8✔
69	name : str = ""	8✔
70	cls : Type[flow.Task] = flow.Task	8✔
71	args : Dict = field(default_factory=dict)	8✔
72	boundary: Type[flow.EventDefinition] = None	8✔
73
74	tasks = 0 # class	8✔
75
76	@classmethod	8✔
77	def reset(cls):	8✔
78	cls.tasks = 0	8✔
79
80	def __post_init__(self):	8✔
81	self.args["name"] = self.name	8✔
82	self.element = self.shape(self.cls, id=f"task_{self.tasks}", **self.args)	8✔
83	self.__class__.tasks += 1	8✔
84
85	@property	8✔
86	def height(self):	8✔
87	return self.element.height + PADDING * 2	8✔
88
89	@property	8✔
90	def width(self):	8✔
91	return self.element.width + PADDING * 2	8✔
92
93	def render(self, x=0, y=0):	8✔
94	self.element.x = x + PADDING	8✔
95	self.element.y = y + PADDING	8✔
96	shapes = [ self.element ]	8✔
97	if self.boundary:	8✔
98	shapes.append(flow.BoundaryEvent(	8✔
99	id=f"boundary-event-{self.element.id}",	4✔
100	definition=self.boundary(id=f"{slugify(self.boundary.__name__)}-{self.element.id}"),	4✔
101	on=self.element	4✔
102	))
103	return (shapes, [])	8✔
104
105	@property	8✔
106	def root(self):	8✔
107	return self.element	8✔
108
109	@property	8✔
110	def tail(self):	8✔
111	return self.element	8✔
112
113	@dataclass	8✔
114	class Process(Step):	8✔
115	name : str = ""	8✔
116	starts : bool = False	8✔
117	ends : bool = False	8✔
118	_root = None	8✔
119	_tail = None	8✔
120
121	def __post_init__(self):	8✔
122	if self.starts:	8✔
123	self._root = self.shape(flow.Start, id="start")	8✔
124	if self.ends:	8✔
125	self._tail = self.shape(flow.End, id="end")	8✔
126
127	def add(self, step):	8✔
128	self.children.append(step)	8✔
129	return self	8✔
130
131	def extend(self, steps):	8✔
132	for step in steps:	8✔
133	self.add(step)	8✔
134	return self	8✔
135
136	@property	8✔
137	def height(self):	8✔
138	return max([ step.height for step in self.children ])	8✔
139
140	@property	8✔
141	def width(self):	8✔
142	total_width = sum([ step.width for step in self.children ])	8✔
143	total_width += FLOW_SPACE * (len(self.children) - 1)	8✔
144	if self.starts:	8✔
145	total_width += self.root.width + FLOW_SPACE	8✔
146	if self.ends:	8✔
147	total_width += self.tail.width + FLOW_SPACE	8✔
148	return total_width	8✔
149
150	def render(self, x=None, y=None):	8✔
151	# if we're called without positioning, we're adding diagram boilerplate
152	wrap = x is None or y is None	8✔
153	if wrap:	8✔
154	x = 50 + MODEL_OFFSET_X # header of participant	8✔
155	y = 0 + MODEL_OFFSET_Y	8✔
156
157	shapes = []	8✔
158	flows = []	8✔
159	prev = None	8✔
160
161	# add optional start event
162	if self.starts:	8✔
163	self.root.x = x	8✔
164	x += self.root.width	8✔
165	self.root.y = y + int((STANDARD_TASK_HEIGHT + PADDING * 2)/2) - (self.root.height/2)	8✔
166	shapes.append(self.root)	8✔
167	x += FLOW_SPACE	8✔
168
169	# add steps
170	for step in self.children:	8✔
171	more_shapes, more_flows = step.render(x=x, y=y)	8✔
172	shapes.extend(more_shapes)	8✔
173	flows.extend(more_flows)	8✔
174	x += step.width + FLOW_SPACE	8✔
175	if prev:	8✔
176	flows.append(self.connect(source=prev.tail, target=step.root))	8✔
177	elif self.starts:	8✔
178	flows.append(self.connect(source=self.root, target=step.root))	8✔
179	prev = step	8✔
180
181	# add optional end event
182	if self.ends:	8✔
183	self.tail.x = x	8✔
184	self.tail.y = self.root.y	8✔
185	shapes.append(self.tail)	8✔
186	flows.append(self.connect(source=shapes[-2], target=self.tail))	8✔
187
188	# optinally wrap it all
189	if wrap:	8✔
190	process = flow.Process(id="process").extend(shapes).extend(flows)	8✔
191	participant = Participant(self.name, process, id="participant")	8✔
192	# participant shapes the lane
193	participant.x = MODEL_OFFSET_X	8✔
194	participant.y = MODEL_OFFSET_Y	8✔
195	participant.width = self.width + 50 + PADDING	8✔
196	participant.height = self.height	8✔
197	collaboration = Collaboration(id="collaboration").append(participant)	8✔
198	return Definitions(id="definitions").extend([	8✔
199	process,	4✔
200	collaboration,	4✔
201	Diagram(	4✔
202	id="diagram",	4✔
203	plane=Plane(id="plane", element=collaboration)	4✔
204	)
205	])
206	return (shapes, flows)	8✔
207
208	@property	8✔
209	def root(self):	8✔
210	if self._root:	8✔
211	return self._root	8✔
212	return self.children[0].root	8✔
213
214	@property	8✔
215	def tail(self):	8✔
216	if self._tail:	8✔
217	return self._tail	8✔
218	return self.children[-1].tail	8✔
219
220	class BranchKind(Enum):	8✔
221	XOR = flow.ExclusiveGateway	8✔
222	OR = flow.InclusiveGateway	8✔
223	AND = flow.ParallelGateway	8✔
224
225	# utility wrapper for an even more fluid API ;-)
226	def If(condition, step):	8✔
227	return (step, condition)	8✔
228
229	@dataclass	8✔
230	class Branch(Step):	8✔
231	default : str = None	8✔
232	kind : BranchKind = BranchKind.XOR	8✔
233
234	gws = 0	8✔
235
236	def __post_init__(self):	8✔
237	global gws
238	self.root = self.shape(self.kind.value, id=f"gateway_start_{self.gws}", name=self.label)	8✔
239	self.tail = self.shape(self.kind.value, id=f"gateway_end_{self.gws}")	8✔
240	self.__class__.gws += 1	8✔
241	# ensure all children to ensure they are tuples
242	self.children = list(map(lambda c: c if type(c) is tuple else (c,None), self.children))	8✔
243
244	@classmethod	8✔
245	def reset(cls):	8✔
246	cls.gws = 0	8✔
247
248	def add(self, branch, condition=None):	8✔
249	self.children.append((branch, condition))	8✔
250	return self	8✔
251
252	def extend(self, steps):	8✔
253	for step in steps:	×
254	self.add(step)	×
UNCOV 255	return self	×
256
257	@property	8✔
258	def height(self):	8✔
259	total_height = sum([ branch.height for branch, _ in self.children ])	8✔
260	if self.default:	8✔
261	total_height += DEFAULT_BRANCH_HEIGHT	8✔
262	return total_height	8✔
263
264	@property	8✔
265	def width(self):	8✔
266	total_width = max([ branch.width for branch, _ in self.children ])	8✔
267	total_width += self.root.width + self.tail.width	8✔
268	total_width += FLOW_SPACE * 2	8✔
269	return total_width	8✔
270
271	def render(self, x=0, y=0):	8✔
272	shapes = []	8✔
273	flows = []	8✔
274	self.root.x = x	8✔
275	self.root.y = y + int((STANDARD_TASK_HEIGHT + PADDING * 2)/2) - (self.root.height/2)	8✔
276	shapes.append(self.root)	8✔
277	shapes.append(self.tail)	8✔
278	x += FLOW_SPACE	8✔
279	if self.default:	8✔
280	flows.append(self.connect(source=self.root, target=self.tail))	8✔
281	y += DEFAULT_BRANCH_HEIGHT	8✔
282	self.tail.x = x + self.width - self.tail.width - FLOW_SPACE	8✔
283	self.tail.y = self.root.y	8✔
284	x += self.root.width	8✔
285	for branch, condition in self.children:	8✔
286	more_shapes, more_flows = branch.render(x=x, y=y)	8✔
287	shapes.extend(more_shapes)	8✔
288	flows.extend(more_flows)	8✔
289	y += branch.height	8✔
290	flows.append(self.connect(source=self.root, target=branch.root, label=condition))	8✔
291	flows.append(self.connect(source=branch.tail, target=self.tail))	8✔
292	return shapes, flows	8✔

christophevg / bpmn-tools / 7288074417

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous