1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 """
22 A workflow is defined by a set of states that a translation unit can be in and
23 the (allowed) transitions between these states. A state is defined by a range
24 between -128 and 127, indicating its level of "completeness". The range is
25 closed at the beginning and open at the end. That is, if a workflow contains
26 states A, B and C where A < B < C, a unit with state number n is in state A if
27 A <= n < B, state B if B <= n < C or state C if C <= n < MAX.
28
29 A value of 0 is typically the "empty" or "new" state with negative values
30 reserved for states like "obsolete" or "do not use".
31
32 Format specific workflows should be defined in such a way that the numeric
33 state values correspond to similar states. For example state 0 should be
34 "untranslated" in PO and "new" or "empty" in XLIFF, state 100 should be
35 "translated" in PO and "final" in XLIFF. This allows formats to implicitly
36 define similar states.
37 """
38
39
51
52
54
55 - def __init__(self, name, enter_action=None, leave_action=None):
56 self.name = name
57 self.enter_action = enter_action
58 self.leave_action = leave_action
59
62
64 return '<State "%s">' % (self.name)
65
67 if not self.enter_action or not callable(self.enter_action):
68 return
69 self.enter_action(obj)
70
72 if not self.leave_action or not callable(self.leave_action):
73 return
74 self.leave_action(obj)
75
76
78
82
84 return '<UnitState name=%s value=%d>' % (self.name, self.state_value)
85
88
89
92
93
96
97
100
101
106
107
113
114
116
117
119 self._current_state = None
120 self._edges = []
121 self._initial_state = None
122 self._states = []
123 self._workflow_obj = wf_obj
124
125
126
128 return list(self._edges)
129 edges = property(_get_edges)
130
132 return list(self._states)
133 states = property(_get_states)
134
135
136
137 - def add_edge(self, from_state, to_state):
138 if isinstance(from_state, basestring):
139 from_state = self.get_state_by_name(from_state)
140 if isinstance(to_state, basestring):
141 to_state = self.get_state_by_name(to_state)
142 for s in (from_state, to_state):
143 if s not in self.states:
144 raise StateNotInWorkflowError(s)
145 if (from_state, to_state) in self.edges:
146 return
147
148 self._edges.append((from_state, to_state))
149
151 if not isinstance(state, State):
152 raise InvalidStateObjectError(state)
153 if state in self.states:
154 raise ValueError('State already in workflow: %s' % (state))
155 self._states.append(state)
156 if self._initial_state is None:
157 self._initial_state = state
158
160 """Returns a list of states that can be transitioned from to the
161 current state."""
162 return [e[0] for e in self.edges if e[1] is self._current_state]
163
165 """Returns a list of states that can be transitioned to from the
166 current state."""
167 return [e[1] for e in self.edges if e[0] is self._current_state]
168
170 """Get the C{State} object for the given name."""
171 for s in self.states:
172 if s.name == state_name:
173 return s
174 else:
175 raise StateNotInWorkflowError(state_name)
176
178 """Set the current state. This is absolute and not subject to edge
179 constraints. The current state's C{leave} and the new state's
180 C{enter} method is still called. For edge transitions, see the
181 C{trans} method."""
182 if isinstance(state, basestring):
183 state = self.get_state_by_name(state)
184 if state not in self.states:
185 raise StateNotInWorkflowError(state)
186
187 if self._current_state:
188 self._current_state.leave(self._workflow_obj)
189 self._current_state = state
190 self._current_state.enter(self._workflow_obj)
191
201
202 - def reset(self, wf_obj, init_state=None):
203 """Reset the work flow to the initial state using the given object."""
204 self._workflow_obj = wf_obj
205 if init_state is not None:
206 if isinstance(init_state, basestring):
207 init_state = self.get_state_by_name(init_state)
208 if init_state not in self.states:
209 raise StateNotInWorkflowError()
210 self._initial_state = init_state
211 self._current_state = init_state
212 return
213 if self._initial_state is None:
214 raise NoInitialStateError()
215 self._current_state = None
216 self.set_current_state(self._initial_state)
217
218 - def trans(self, to_state=None):
219 """Transition to the given state. If no state is given, the first one
220 returned by C{get_to_states} is used."""
221 if self._current_state is None:
222 raise ValueError('No current state set')
223 if isinstance(to_state, basestring):
224 to_state = self.get_state_by_name(to_state)
225 if to_state is None:
226 to_state = self.get_to_states()
227 if not to_state:
228 raise TransitionError('No state to transition to')
229 to_state = to_state[0]
230 if to_state not in self.states:
231 raise StateNotInWorkflowError(to_state)
232 if (self._current_state, to_state) not in self.edges:
233 raise TransitionError('No edge between edges %s and %s' % (self._current_state, to_state))
234 self._current_state.leave(self._workflow_obj)
235 self._current_state = to_state
236 self._current_state.enter(self._workflow_obj)
237
238
240 wf = Workflow(unit)
241
242 state_info = unit.STATE.items()
243 state_info.sort(key=lambda x: x[0])
244
245 init_state, prev_state = None, None
246 for state_id, state_range in state_info:
247 if state_range[0] < 0:
248 continue
249 state_name = state_names[state_id]
250
251 state = UnitState(state_name, state_range[0])
252 wf.add_state(state)
253
254
255 if init_state is None and state_range[0] >= 0:
256 init_state = state
257
258 if prev_state:
259 wf.add_edge(prev_state, state_name)
260 prev_state = state_name
261
262 if init_state:
263 wf.set_initial_state(init_state)
264
265 return wf
266