8938991339

Committed 03 May 2024 12:30PM UTC coverage: 98.432% (+0.2%) from 98.217%

Build # 8938991339

Build Type

push

github

Committed by

aleneum

Commit Message

use coverage only for mypy job and update setup.py tags

Run Details

5149 of 5231 relevant lines covered (98.43%)

0.98 hits per line

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

97.7

/transitions/extensions/diagrams_base.py

"""
    transitions.extensions.diagrams_base
    ------------------------------------

    The class BaseGraph implements the common ground for Graphviz backends.
"""

import copy
import abc
import logging
import six

_LOGGER = logging.getLogger(__name__)
_LOGGER.addHandler(logging.NullHandler())


@six.add_metaclass(abc.ABCMeta)
class BaseGraph(object):
    """Provides the common foundation for graphs generated either with pygraphviz or graphviz. This abstract class
    should not be instantiated directly. Use .(py)graphviz.(Nested)Graph instead.
    Attributes:
        machine (GraphMachine): The associated GraphMachine
        fsm_graph (object): The AGraph-like object that holds the graphviz information
    """

    def __init__(self, machine):
        self.machine = machine
        self.fsm_graph = None
        self.generate()

    @abc.abstractmethod
    def generate(self):
        """Triggers the generation of a graph."""

    @abc.abstractmethod
    def set_previous_transition(self, src, dst):
        """Sets the styling of an edge to 'previous'
        Args:
            src (str): Name of the source state
            dst (str): Name of the destination
        """

    @abc.abstractmethod
    def reset_styling(self):
        """Resets the styling of the currently generated graph."""

    @abc.abstractmethod
    def set_node_style(self, state, style):
        """Sets the style of nodes associated with a model state
        Args:
            state (str, Enum or list): Name of the state(s) or Enum(s)
            style (str): Name of the style
        """

    @abc.abstractmethod
    def get_graph(self, title=None, roi_state=None):
        """Returns a graph object.
        Args:
            title (str): Title of the generated graph
            roi_state (State): If not None, the returned graph will only contain edges and states connected to it.
        Returns:
             A graph instance with a `draw` that allows to render the graph.
        """

    def _convert_state_attributes(self, state):
        label = state.get("label", state["name"])
        if self.machine.show_state_attributes:
            if "tags" in state:
                label += " [" + ", ".join(state["tags"]) + "]"
            if "on_enter" in state:
                label += r"\l- enter:\l  + " + r"\l  + ".join(state["on_enter"])
            if "on_exit" in state:
                label += r"\l- exit:\l  + " + r"\l  + ".join(state["on_exit"])
            if "timeout" in state:
                label += r'\l- timeout(' + state['timeout'] + 's) -> (' + ', '.join(state['on_timeout']) + ')'
        # end each label with a left-aligned newline
        return label + r"\l"

    def _get_state_names(self, state):
        if isinstance(state, (list, tuple, set)):
            for res in state:
                for inner in self._get_state_names(res):
                    yield inner
        else:
            yield self.machine.state_cls.separator.join(self.machine._get_enum_path(state))\
                if hasattr(state, "name") else state

    def _transition_label(self, tran):
        edge_label = tran.get("label", tran["trigger"])
        if "dest" not in tran:
            edge_label += " [internal]"
        if self.machine.show_conditions and any(prop in tran for prop in ["conditions", "unless"]):
            edge_label = "{edge_label} [{conditions}]".format(
                edge_label=edge_label,
                conditions=" & ".join(
                    tran.get("conditions", []) + ["!" + u for u in tran.get("unless", [])]
                ),
            )
        return edge_label

    def _get_global_name(self, path):
        if path:
            state = path.pop(0)
            with self.machine(state):
                return self._get_global_name(path)
        else:
            return self.machine.get_global_name()

    def _flatten(self, *lists):
        return (e for a in lists for e in
                (self._flatten(*a)
                 if isinstance(a, (tuple, list))
                 else (a.name if hasattr(a, 'name') else a,)))

    def _get_elements(self):
        states = []
        transitions = []
        try:
            markup = self.machine.get_markup_config()
            queue = [([], markup)]

            while queue:
                prefix, scope = queue.pop(0)
                for transition in scope.get("transitions", []):
                    if prefix:
                        tran = copy.copy(transition)
                        tran["source"] = self.machine.state_cls.separator.join(
                            prefix + [tran["source"]]
                        )
                        if "dest" in tran:  # don't do this for internal transitions
                            tran["dest"] = self.machine.state_cls.separator.join(
                                prefix + [tran["dest"]]
                            )
                    else:
                        tran = transition
                    transitions.append(tran)
                for state in scope.get("children", []) + scope.get("states", []):
                    if not prefix:
                        sta = state
                        states.append(sta)

                    ini = state.get("initial", [])
                    if not isinstance(ini, list):
                        ini = ini.name if hasattr(ini, "name") else ini
                        tran = dict(
                            trigger="",
                            source=self.machine.state_cls.separator.join(prefix + [state["name"]]),
                            dest=self.machine.state_cls.separator.join(
                                prefix + [state["name"], ini]
                            ),
                        )
                        transitions.append(tran)
                    if state.get("children", []):
                        queue.append((prefix + [state["name"]], state))
        except KeyError:
            _LOGGER.error("Graph creation incomplete!")
        return states, transitions

1	"""	1✔
2	transitions.extensions.diagrams_base
3	------------------------------------
4
5	The class BaseGraph implements the common ground for Graphviz backends.
6	"""
7
8	import copy	1✔
9	import abc	1✔
10	import logging	1✔
11	import six	1✔
12
13	_LOGGER = logging.getLogger(__name__)	1✔
14	_LOGGER.addHandler(logging.NullHandler())	1✔
15
16
17	@six.add_metaclass(abc.ABCMeta)	1✔
18	class BaseGraph(object):	1✔
19	"""Provides the common foundation for graphs generated either with pygraphviz or graphviz. This abstract class	1✔
20	should not be instantiated directly. Use .(py)graphviz.(Nested)Graph instead.
21	Attributes:
22	machine (GraphMachine): The associated GraphMachine
23	fsm_graph (object): The AGraph-like object that holds the graphviz information
24	"""
25
26	def __init__(self, machine):	1✔
27	self.machine = machine	1✔
28	self.fsm_graph = None	1✔
29	self.generate()	1✔
30
31	@abc.abstractmethod	1✔
32	def generate(self):	1✔
33	"""Triggers the generation of a graph."""
34
35	@abc.abstractmethod	1✔
36	def set_previous_transition(self, src, dst):	1✔
37	"""Sets the styling of an edge to 'previous'
38	Args:
39	src (str): Name of the source state
40	dst (str): Name of the destination
41	"""
42
43	@abc.abstractmethod	1✔
44	def reset_styling(self):	1✔
45	"""Resets the styling of the currently generated graph."""
46
47	@abc.abstractmethod	1✔
48	def set_node_style(self, state, style):	1✔
49	"""Sets the style of nodes associated with a model state
50	Args:
51	state (str, Enum or list): Name of the state(s) or Enum(s)
52	style (str): Name of the style
53	"""
54
55	@abc.abstractmethod	1✔
56	def get_graph(self, title=None, roi_state=None):	1✔
57	"""Returns a graph object.
58	Args:
59	title (str): Title of the generated graph
60	roi_state (State): If not None, the returned graph will only contain edges and states connected to it.
61	Returns:
62	A graph instance with a `draw` that allows to render the graph.
63	"""
64
65	def _convert_state_attributes(self, state):	1✔
66	label = state.get("label", state["name"])	1✔
67	if self.machine.show_state_attributes:	1✔
68	if "tags" in state:	1✔
69	label += " [" + ", ".join(state["tags"]) + "]"	1✔
70	if "on_enter" in state:	1✔
71	label += r"\l- enter:\l + " + r"\l + ".join(state["on_enter"])	1✔
72	if "on_exit" in state:	1✔
73	label += r"\l- exit:\l + " + r"\l + ".join(state["on_exit"])	1✔
74	if "timeout" in state:	1✔
75	label += r'\l- timeout(' + state['timeout'] + 's) -> (' + ', '.join(state['on_timeout']) + ')'	1✔
76	# end each label with a left-aligned newline
77	return label + r"\l"	1✔
78
79	def _get_state_names(self, state):	1✔
80	if isinstance(state, (list, tuple, set)):	1✔
81	for res in state:	1✔
82	for inner in self._get_state_names(res):	1✔
83	yield inner	1✔
84	else:
85	yield self.machine.state_cls.separator.join(self.machine._get_enum_path(state))\	1✔
86	if hasattr(state, "name") else state
87
88	def _transition_label(self, tran):	1✔
89	edge_label = tran.get("label", tran["trigger"])	1✔
90	if "dest" not in tran:	1✔
91	edge_label += " [internal]"	1✔
92	if self.machine.show_conditions and any(prop in tran for prop in ["conditions", "unless"]):	1✔
93	edge_label = "{edge_label} [{conditions}]".format(	1✔
94	edge_label=edge_label,
95	conditions=" & ".join(
96	tran.get("conditions", []) + ["!" + u for u in tran.get("unless", [])]
97	),
98	)
99	return edge_label	1✔
100
101	def _get_global_name(self, path):	1✔
102	if path:	1✔
103	state = path.pop(0)	1✔
104	with self.machine(state):	1✔
105	return self._get_global_name(path)	1✔
106	else:
107	return self.machine.get_global_name()	1✔
108
109	def _flatten(self, *lists):	1✔
110	return (e for a in lists for e in	1✔
111	(self._flatten(*a)
112	if isinstance(a, (tuple, list))
113	else (a.name if hasattr(a, 'name') else a,)))
114
115	def _get_elements(self):	1✔
116	states = []	1✔
117	transitions = []	1✔
118	try:	1✔
119	markup = self.machine.get_markup_config()	1✔
120	queue = [([], markup)]	1✔
121
122	while queue:	1✔
123	prefix, scope = queue.pop(0)	1✔
124	for transition in scope.get("transitions", []):	1✔
125	if prefix:	1✔
126	tran = copy.copy(transition)	1✔
127	tran["source"] = self.machine.state_cls.separator.join(	1✔
128	prefix + [tran["source"]]
129	)
130	if "dest" in tran: # don't do this for internal transitions	1✔
131	tran["dest"] = self.machine.state_cls.separator.join(	1✔
132	prefix + [tran["dest"]]
133	)
134	else:
135	tran = transition	1✔
136	transitions.append(tran)	1✔
137	for state in scope.get("children", []) + scope.get("states", []):	1✔
138	if not prefix:	1✔
139	sta = state	1✔
140	states.append(sta)	1✔
141
142	ini = state.get("initial", [])	1✔
143	if not isinstance(ini, list):	1✔
144	ini = ini.name if hasattr(ini, "name") else ini	1✔
145	tran = dict(	1✔
146	trigger="",
147	source=self.machine.state_cls.separator.join(prefix + [state["name"]]),
148	dest=self.machine.state_cls.separator.join(
149	prefix + [state["name"], ini]
150	),
151	)
152	transitions.append(tran)	1✔
153	if state.get("children", []):	1✔
154	queue.append((prefix + [state["name"]], state))	1✔
155	except KeyError:	×
156	_LOGGER.error("Graph creation incomplete!")	×
157	return states, transitions	1✔

pytransitions / transitions / 8938991339

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