Skip to content

erp5
erp5
Commit 9880c119 authored by Jean-Paul Smets's avatar Jean-Paul Smets
Browse files
Options

Some more ideas where BPM is heading to 

git-svn-id: https://svn.erp5.org/repos/public/erp5/sandbox/amount_generator@35585 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent e31f2aef
Show whitespace changes
Inline Side-by-side
Showing with 611 additions and 202 deletions
product/ERP5/Document/BusinessPath.py
View file @ 9880c119
......@@ -34,42 +34,12 @@ from AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet, interfaces
from Products.ERP5.Document.Path import Path
from Products.ERP5.Document.Predicate import Predicate
from Products.ERP5.ExplanationCache import _getExplanationCache
import zope.interface
from zLOG import LOG
class ExplanationCache:
"""ExplanationCache provides a central access to
all parameters and values which are needed to process
an explanation. It is based on the idea that a value is calculated
once and once only, as a way to accelerate performance of algorithms
related to an explanation.
'explanation_uid': self._getExplanationUidList(explanation) # XXX-JPS why do we need explanation_uid ? and why a list
'simulation_path': simulation_path,
explanation_uid = self._getExplanationUidList(explanation) # A hint value to reduce the size of the tree
simulation_path = '/erp5/p.../%' # A list of path
"""
def __init__(self, explanation):
"""
"""
self.explanation = explanation
def getRootExplanationUidList(self):
"""
"""
def getSimulationPathPatternList(self):
"""
"""
def _getExplanationCache(explanation):
# XXX-JPS Cache this in a transaction variable or better
return ExplanationCache(explanation)
class BusinessPath(Path, Predicate):
"""
The BusinessPath class embeds all information related to
......@@ -175,6 +145,12 @@ class BusinessPath(Path, Predicate):
#'destination_transport'
)
# Helper Methods
def _getExplanationRelatedSimulationMovementValueList(self, explanation):
explanation_cache = _getExplanationCache(explanation)
return explanation_cache.getBusinessPathRelatedSimulationMovementValueList(self)
# XXX-JPS UNkonwn ?
security.declareProtected(Permissions.AccessContentsInformation,
'getArrowCategoryDict')
def getArrowCategoryDict(self, context=None, **kw): # XXX-JPS do we need it in API ?
......@@ -272,6 +248,22 @@ class BusinessPath(Path, Predicate):
method = getattr(movement, method_id) # We wish to raise if it does not exist
return method()
def getCompletionDate(self, explanation):
"""Returns the date of completion of the movemnet
based on paremeters of the business path. This complete date can be
the start date, the stop date, the date of a given workflow transition
on the explaining delivery, etc.
XXX - DOC
movement -- a Simulation Movement
"""
date_list = []
for movement in self._getExplanationRelatedSimulationMovementValueList(explanation):
date_list.append(self.getMovementCompletionDate(movement))
return max(date_list)
security.declareProtected(Permissions.AccessContentsInformation,
'getExpectedQuantity')
def getExpectedQuantity(self, amount):
......@@ -363,6 +355,15 @@ class BusinessPath(Path, Predicate):
return False
return True
def isDelivered(self, explanation):
"""XXX
"""
for simulation_movement in self._getExplanationRelatedSimulationMovementValueList(
explanation):
if not simulation_movement.getDelivery():
return False
return True
def build(self, explanation):
"""Builds all related movements in the simulation using the builders
defined on the Business Path.
......
product/ERP5/Document/BusinessProcess.py
View file @ 9880c119
......@@ -32,6 +32,7 @@ from AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet, interfaces
from Products.ERP5Type.XMLObject import XMLObject
from Products.ERP5.Document.Path import Path
from Products.ERP5.ExplanationCache import _getExplanationCache, _getBusinessPathClosure
import zope.interface
......@@ -43,6 +44,8 @@ class BusinessProcess(Path, XMLObject):
TODO:
- finish interface implementation
RENAME:
getPathValueList -> getBusinessPathValueList
"""
meta_type = 'ERP5 Business Process'
portal_type = 'Business Process'
......@@ -66,19 +69,23 @@ class BusinessProcess(Path, XMLObject):
zope.interface.implements(interfaces.IBusinessProcess,
interfaces.IArrowBase)
# Access to path and states of the business process
security.declareProtected(Permissions.AccessContentsInformation, 'getPathValueList')
def getPathValueList(self, trade_phase=None, context=None, **kw):
"""
Returns all Path of the current BusinessProcess which
# IBusinessPathProcess implementation
security.declareProtected(Permissions.AccessContentsInformation, 'getBusinessPathValueList')
def getBusinessPathValueList(self, trade_phase=None, context=None,
predecessor=None, successor=None, **kw):
"""Returns all Path of the current BusinessProcess which
are matching the given trade_phase and the optional context.
trade_phase -- a single trade phase category or a list of
trade phases
trade_phase -- filter by trade phase
context -- a context to test each Business Path on
and filter out Business Path which do not match
context -- the context to search matching predicates for
predecessor -- filter by trade state predecessor
**kw -- same parameters as for searchValues / contentValues
successor -- filter by trade state successor
**kw -- same arguments as those passed to searchValues / contentValues
"""
if trade_phase is None:
trade_phase = set()
......@@ -87,14 +94,22 @@ class BusinessProcess(Path, XMLObject):
else:
trade_phase = set(trade_phase)
result = []
if kw.get('portal_type', None) is None:
kw['portal_type'] = self.getPortalBusinessPathTypeList()
if kw.get('sort_on', None) is None:
kw['sort_on'] = 'int_index'
original_business_path_list = self.objectValues(**kw), # Why Object Values ??? XXX-JPS
if len(trade_phase) == 0:
return result
# Separate the selection of business paths into twp steps
return original_business_path_list # If not trade_phase is specified, return all Business Path
# Separate the selection of business paths into two steps
# for easier debugging.
# First, collect business paths which can be applicable to a given context.
business_path_list = []
for business_path in self.objectValues(portal_type='Business Path',
sort_on='int_index'):
for business_path in original_business_path_list:
if predecessor is not None and business_path.getPredecessor() != predecessor:
break # Filter our business path which predecessor does not match
if successor is not None and business_path.getSuccessor() != successor:
break # Filter our business path which predecessor does not match
if trade_phase.intersection(business_path.getTradePhaseList()):
business_path_list.append(business_path)
# Then, filter business paths by Predicate API.
......@@ -106,174 +121,430 @@ class BusinessProcess(Path, XMLObject):
result.append(business_path)
return result
security.declareProtected(Permissions.AccessContentsInformation, 'getStateValueList')
def getStateValueList(self, *args, **kw):
"""
Returns all states of the business process. The method
**kw parameters follows the API of searchValues / contentValues
"""
# Naive implementation to redo XXX
kw['portal_type'] = "Business Path"
return [x for x in [y.getSuccessorValue() for y in \
self.contentValues(*args, **kw)] if x is not None]
def isBusinessPathCompleted(self, explanation, business_path):
"""Returns True if given Business Path document
is completed in the context of provided explanation.
# Access to path and states of the business process
def isCompleted(self, explanation):
"""
True if all states are completed
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
business_path -- a Business Path document
"""
for state in self.getStateValueList():
if not state.isCompleted(explanation):
# Return False if Business Path is not completed
if not business_path.isCompleted(explanation):
return False
predecessor_state = business_path.getPredecessor()
if not predecessor_state:
# This is a root business path, no predecessor
# so no need to do any recursion
return True
if self.isTradeStateCompleted(explanation, predecessor_state):
# If predecessor state is globally completed for the
# given explanation, return True
# Please note that this is a specific case for a Business Process
# built using asUnionBusinessProcess. In such business process
# a business path may be completed even if its predecessor state
# is not
return True
# Build the closure business process which only includes those business
# path wich are directly related to the current business path but DO NOT
# narrow down the explanation else we might narrow down so much that
# it becomes an empty set
closure_process = _getBusinessPathClosure(explanation, business_path)
return closure_process.isTradeStateCompleted(explanation, predecessor_state)
def isBuildable(self, explanation):
"""
True if all any path is buildable
"""
return len(self.getBuildablePathValueList(explanation)) != 0
def isBusinessPathPartiallyCompleted(self, explanation, business_path):
"""Returns True if given Business Path document
is partially completed in the context of provided explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
def getBuildablePathValueList(self, explanation):
business_path -- a Business Path document
"""
Returns the list of Business Path which are ready to
be built
# Return False if Business Path is not partially completed
if not business_path.isPartiallyCompleted(explanation):
return False
predecessor_state = business_path.getPredecessor()
if not predecessor_state:
# This is a root business path, no predecessor
# so no need to do any recursion
return True
if self.isTradeStatePartiallyCompleted(explanation, predecessor_state):
# If predecessor state is globally partially completed for the
# given explanation, return True
# Please note that this is a specific case for a Business Process
# built using asUnionBusinessProcess. In such business process
# a business path may be partially completed even if its predecessor
# state is not
return True
# Build the closure business process which only includes those business
# path wich are directly related to the current business path but DO NOT
# narrow down the explanation else we might narrow down so much that
# it becomes an empty set
closure_process = _getBusinessPathClosure(explanation, business_path)
return closure_process.isTradeStatePartiallyCompleted(explanation,
predecessor_state)
def getExpectedBusinessPathCompletionDate(explanation, business_path,
delay_mode=None):
"""Returns the expected completion date of given Business Path document
in the context of provided explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
business_path -- a Business Path document
delay_mode -- optional value to specify calculation mode ('min', 'max')
if no value specified use average delay
"""
return filter(lambda x:x.isBuildable(explanation),
self.objectValues(portal_type='Business Path'))
closure_process = _getBusinessPathClosure(explanation, business_path)
# XXX use explanatoin cache to optimize
predecessor = business_path.getPredecessor()
if closure_process.isTradeStateRootState(predecessor):
return business_path.getCompletionDate(explanation)
reference_date = closure_process.getExpectedTradeStateCompletionDate(explanation, predecessor)
return reference_date + wait_time + lead_time
def getExpectedBusinessPathStartAndStopDate(explanation, business_path,
delay_mode=None):
"""Returns the expected start and stop dates of given Business Path
document in the context of provided explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
business_path -- a Business Path document
def getCompletedStateValueList(self, explanation):
delay_mode -- optional value to specify calculation mode ('min', 'max')
if no value specified use average delay
"""
Returns the list of Business States which are finished
closure_process = _getBusinessPathClosure(explanation, business_path)
# XXX use explanatoin cache to optimize
trade_date = self.getTradeDate()
if trade_date is not None:
reference_date = closure_process.gettExpectedTradePhaseCompletionDate(explanation, trade_date)
else:
predecessor = business_path.getPredecessor() # XXX-JPS they should all have
reference_date = closure_process.getExpectedTradeStateCompletionDate(explanation, predecessor)
start_date = reference_date + wait_time
stop_date = start_date + lead_time # XXX-JPS use appropriate values
return start_date, stop_date
# IBuildableBusinessPathProcess implementation
def getBuildableBusinessPathValueList(self, explanation):
"""Returns the list of Business Path which are buildable
by taking into account trade state dependencies between
Business Path.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
return filter(lambda x:x.isCompleted(explanation), self.getStateValueList())
result = []
for business_path in self.getBusinessPathValueList():
if self.isBusinessPathBuildable(explanation, business_path):
result.append(business_path)
return result
def getPartiallyBuildableBusinessPathValueList(self, explanation):
"""Returns the list of Business Path which are partially buildable
by taking into account trade state dependencies between
Business Path.
def getPartiallyCompletedStateValueList(self, explanation):
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
Returns the list of Business States which are finished
result = []
for business_path in self.getBusinessPathValueList():
if self.isBusinessPathPartiallyBuildable(explanation, business_path):
result.append(business_path)
return result
def isBusinessPathBuildable(self, explanation, business_path):
"""Returns True if any of the related Simulation Movement
is buildable and if the predecessor trade state is completed.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
business_path -- a Business Path document
"""
return filter(lambda x:x.isPartiallyCompleted(explanation), self.getStateValueList())
# If everything is delivered, no need to build
if business_path.isDelivered(explanation):
return False
# We must take the closure cause only way to combine business process
closure_process = _getBusinessPathClosure(explanation, business_path)
predecessor = business_path.getPredecessor()
return closure_process.isTradeStateCompleted(predecessor)
def getLatestCompletedStateValue(self, explanation):
def isBusinessPathPartiallyBuildable(self, explanation, business_path):
"""Returns True if any of the related Simulation Movement
is buildable and if the predecessor trade state is partially completed.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
business_path -- a Business Path document
"""
Returns the most advanced completed state
# If everything is delivered, no need to build
if business_path.isDelivered(explanation):
return False
# We must take the closure cause only way to combine business process
closure_process = _getBusinessPathClosure(explanation, business_path)
predecessor = business_path.getPredecessor()
return closure_process.isTradeStatePartiallyCompleted(predecessor)
# ITradeStateProcess implementation
def getTradeStateList():
"""Returns list of all trade_state of this Business Process
by looking at successor and predecessor values of contained
Business Path.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
result = set()
for business_path in self.getBusinessPathValueList():
result.add(business_path.getPredecessor())
result.add(business_path.getSuccessor())
return result
def getSuccessorTradeStateList(explanation, trade_state):
"""Returns the list of successor states in the
context of given explanation. This list is built by looking
at all successor of business path involved in given explanation
and which predecessor is the given trade_phase.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
trade_state -- a Trade State category
"""
for state in self.getCompletedStateValueList(explanation):
for path in state.getPredecessorRelatedValueList():
if not path.isCompleted(explanation):
return state
return None
result = set()
for business_path in self.getBusinessPathValueList():
if business_path.getPredecessor() == trade_state:
result.add(business_path.getSuccessor())
return result
def getPredecessorTradeStateList(explanation, trade_state):
"""Returns the list of predecessor states in the
context of given explanation. This list is built by looking
at all predecessor of business path involved in given explanation
and which sucessor is the given trade_phase.
def getLatestPartiallyCompletedStateValue(self, explanation):
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
trade_state -- a Trade State category
"""
Returns the most advanced completed state
result = set()
for business_path in self.getBusinessPathValueList():
if business_path.getSuccessor() == trade_state:
result.add(business_path.getPredecessor())
return result
def getCompletedTradeStateList(explanation):
"""Returns the list of Trade States which are completed
in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
for state in self.getCompletedStateValueList(explanation):
for path in state.getPredecessorRelatedValueList():
if not path.isPartiallyCompleted(explanation):
return state
return None
return filter(lambda x:self.isTradeStateCompleted(explanation, x), self.getTradeStateList())
def getLatestCompletedStateValueList(self, explanation):
def getPartiallyCompletedTradeStateList(explanation):
"""Returns the list of Trade States which are partially
completed in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
Returns the most advanced completed state
return filter(lambda x:self.isTradeStatePartiallyCompleted(explanation, x), self.getTradeStateList())
def getLatestCompletedTradeStateList(explanation):
"""Returns the list of completed trade states which predecessor
states are completed and for which no successor state
is completed in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
result = []
for state in self.getCompletedStateValueList(explanation):
for path in state.getPredecessorRelatedValueList():
if not path.isCompleted(explanation):
result.append(state)
result = set()
for state in self.getCompletedTradeStateValue(explanation):
for business_path in state.getPredecessorRelatedValueList():
if not self.isBusinessPathCompleted(explanation, business_path):
result.add(state)
return result
def getLatestPartiallyCompletedStateValueList(self, explanation):
"""
Returns the most advanced completed state
def getLatestPartiallyCompletedTradeStateList(explanation):
"""Returns the list of completed trade states which predecessor
states are completed and for which no successor state
is partially completed in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
result = []
for state in self.getCompletedStateValueList(explanation):
for path in state.getPredecessorRelatedValueList():
if not path.isPartiallyCompleted(explanation):
result.append(state)
result = set()
for state in self.getCompletedTradeStateValue(explanation):
for business_path in state.getPredecessorRelatedValueList():
if not self.isBusinessPathPartiallyCompleted(explanation, business_path):
result.add(state)
return result
def build(self, explanation_relative_url):
def isTradeStateCompleted(explanation, trade_state):
"""Returns True if all predecessor trade states are
completed and if no successor trade state is completed
in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
trade_state -- a Trade State category
"""
Build whatever is buildable
for business_path in self.getBusinessPathValueList(successor=trade_state):
if not closure_process.isBusinessPathCompleted(explanation, business_path):
return False
return True
def isTradeStatePartiallyCompleted(explanation, trade_state):
"""Returns True if all predecessor trade states are
completed and if no successor trade state is partially completed
in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
trade_state -- a Trade State category
"""
explanation = self.restrictedTraverse(explanation_relative_url)
for path in self.getBuildablePathValueList(explanation):
path.build(explanation)
for business_path in self.getBusinessPathValueList(successor=trade_state):
if not self.isBusinessPathPartiallyCompleted(explanation, business_path):
return False
return True
def getExpectedTradeStateCompletionDate(explanation, trade_state,
delay_mode=None):
"""Returns the date at which the give trade state is expected
to be completed in the context of given explanation.
def isStartDateReferential(self): # XXX - not in interface
return self.getReferentialDate() == 'start_date'
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
def isStopDateReferential(self): # XXX - not in interface
return self.getReferentialDate() == 'stop_date'
trade_state -- a Trade State category
def getTradePhaseList(self):
delay_mode -- optional value to specify calculation mode ('min', 'max')
if no value specified use average delay
"""
Returns all trade_phase of this business process
date_list = []
for business_path in self.getBusinessPathValueList(successor=trade_state):
date_list.append(self.getExpectedBusinessPathCompletionDate(explanation, business_path))
return max(date_list) # XXX-JPS provide -infinite for...
# ITradePhaseProcess implementation
def getTradePhaseList():
"""Returns list of all trade_phase of this Business Process
by looking at trade_phase values of contained Business Path.
"""
path_list = self.objectValues(portal_type=self.getPortalBusinessPathTypeList())
return filter(None, [path.getTradePhase()
for path in path_list])
result = set()
for business_path in self.getBusinessPathValueList():
result.union(business_path.getTradePhaseList())
return result
def getRootExplanationPathValue(self):
def getCompletedTradePhaseList(explanation):
"""Returns the list of Trade Phases which are completed
in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
Returns a root path of this business process
return filter(lambda x:self.isTradePhaseCompleted(explanation, x), self.getTradePhaseList())
def getPartiallyCompletedTradePhaseList(explanation):
"""Returns the list of Trade Phases which are partially completed
in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
path_list = self.objectValues(portal_type=self.getPortalBusinessPathTypeList())
path_list = filter(lambda x: x.isDeliverable(), path_list)
return filter(lambda x:self.isTradePhasePartiallyCompleted(explanation, x), self.getTradePhaseList())
if len(path_list) > 1:
raise Exception, "this business process has multi root paths"
def isTradePhaseCompleted(explanation, trade_phase):
"""Returns True all business path with given trade_phase
applicable to given explanation are completed.
if len(path_list) == 1:
return path_list[0]
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
def getHeadPathValueList(self, trade_phase_list=None):
trade_phase -- a Trade Phase category
"""
Returns a list of head path(s) of this business process
for business_path in self.getBusinessPathValueList(trade_phase=trade_phase):
if not self.isBusinessPathCompleted(explanation, business_path):
return False
return True
trade_phase_list -- used to filtering, means that discovering
a list of head path with the trade_phase_list
"""
head_path_list = list()
for state in self.getStateValueList():
if len(state.getSuccessorRelatedValueList()) == 0:
head_path_list += state.getPredecessorRelatedValueList()
def isTradePhasePartiallyCompleted(explanation, trade_phase):
"""Returns True at least one business path with given trade_phase
applicable to given explanation is partially completed
or completed.
if trade_phase_list is not None:
_set = set(trade_phase_list)
_list = list()
# start to discover a head path with the trade_phase_list from head path(s) of whole
for path in head_path_list:
_list += self._getHeadPathValueList(path, _set)
head_path_list = map(lambda t: t[0], filter(lambda t: t != (None, None), _list))
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
return head_path_list
trade_phase -- a Trade Phase category
"""
for business_path in self.getBusinessPathValueList(trade_phase=trade_phase):
if not self.isBusinessPathPartiallyCompleted(explanation, business_path):
return False
return True
def _getHeadPathValueList(self, path, trade_phase_set):
# if the path has target trade_phase, it is a head path.
_set = set(path.getTradePhaseList())
if _set & trade_phase_set:
return [(path, _set & trade_phase_set)]
def getExpectedTradePhaseCompletionDate(explanation, trade_phase,
delay_mode=None):
"""Returns the date at which the give trade phase is expected
to be completed in the context of given explanation, taking
into account the graph of date constraints defined by business path
and business states.
node = path.getSuccessorValue()
if node is None:
return [(None, None)]
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
_list = list()
for next_path in node.getPredecessorRelatedValueList():
_list += self._getHeadPathValueList(next_path, trade_phase_set)
return _list
trade_phase -- a Trade Phase category
def getRemainingTradePhaseList(self, explanation, trade_state, trade_phase_list=None):
delay_mode -- optional value to specify calculation mode ('min', 'max')
if no value specified use average delay
"""
Returns the list of remaining trade phase for this
state based on the explanation.
date_list = []
for business_path in self.getBusinessPathValueList(trade_phase=trade_phase):
date_list.append(self.getExpectedTradePhaseCompletionDate(explanation, business_path, delay_mode=delay_mode))
return max(date_list)
trade_phase_list -- if provide, the result is filtered by it after collected
def getRemainingTradePhaseList(business_path, trade_phase_list=None):
"""Returns the list of remaining trade phases which to be achieved
as part of a business process. This list is calculated by analysing
the graph of business path and trade states, starting from a given
business path. The result if filtered by a list of trade phases. This
method is useful mostly for production and MRP to manage a distributed
supply and production chain.
business_path -- a Business Path document
trade_phase_list -- if provided, the result is filtered by it after
being collected - ???? useful ? XXX-JPS ?
NOTE: explanation is not involved here because we consider here that
self is the result of asUnionBusinessProcess and thus only contains
applicable Business Path to a given simulation subtree. Since the list
of remaining trade phases does not depend on exact values in the
simulation, we did not include the explanation. However, this makes the
API less uniform.
"""
# XXXX REVIEWJPS
remaining_trade_phase_list = []
for path in [x for x in self.objectValues(portal_type="Business Path") \
if x.getPredecessorValue() == trade_state]:
......@@ -297,38 +568,54 @@ class BusinessProcess(Path, XMLObject):
return remaining_trade_phase_list
def isStatePartiallyCompleted(self, explanation, trade_state):
"""
If all path which reach this state are partially completed
then this state is completed
# IBusinessProcess global API
def isCompleted(explanation):
"""Returns True is all applicable Trade States and Trade Phases
are completed in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
for path in [x for x in self.objectValues(portal_type="Business Path") \
if x.getSuccessorValue() == trade_state]:
if not path.isPartiallyCompleted(explanation):
for state in self.getTradeStateList():
if not state.isTradeStateCompleted(explanation):
return False
return True
def getExpectedStateCompletionDate(self, explanation, trade_state, *args, **kwargs):
def getExpectedCompletionDate(explanation, delay_mode=None):
"""Returns the expected date at which all applicable Trade States and
Trade Phases are completed in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
Returns the expected completion date for this
state based on the explanation.
explanation -- the document
def isBuildable(explanation):
"""Returns True is one Business Path of this Business Process
is buildable in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
# Should be re-calculated?
# XXX : what is the purpose of the two following lines ? comment it until there is
# good answer
if 'predecessor_date' in kwargs:
del kwargs['predecessor_date']
successor_list = [x for x in self.objectValues(portal_type="Business Path") \
if x.getSuccessorValue() == trade_state]
date_list = self._getExpectedDateList(explanation,
successor_list,
self._getExpectedCompletionDate,
*args,
**kwargs)
if len(date_list) > 0:
return min(date_list)
return len(self.getBuildableBusinessPathValueList(explanation)) != 0
def isPartiallyBuildable(explanation):
"""Returns True is one Business Path of this Business Process
is partially buildable in the context of given explanation.
explanation -- an Order, Order Line, Delivery or Delivery Line which
implicitely defines a simulation subtree
"""
return len(self.getPartiallyBuildableBusinessPathValueList(explanation)) != 0
def build(self, explanation):
"""
Build whatever is buildable
"""
for business_path in self.getBuildableBusinessPathValueList(explanation):
business_path.build(explanation)
# GARBAGE - XXXXXXXXXXXXXXXXXXXXXXX
def getExpectedStateBeginningDate(self, explanation, trade_state, *args, **kwargs):
"""
......@@ -381,6 +668,75 @@ class BusinessProcess(Path, XMLObject):
return expected_date_list
def getRootExplanationPathValue(self): # XXX-JPS not in API
"""
Returns a root path of this business process
"""
path_list = self.objectValues(portal_type=self.getPortalBusinessPathTypeList())
path_list = filter(lambda x: x.isDeliverable(), path_list)
if len(path_list) > 1:
raise Exception, "this business process has multi root paths"
if len(path_list) == 1:
return path_list[0]
def getHeadPathValueList(self, trade_phase_list=None): # XXX-JPS not in API
"""
Returns a list of head path(s) of this business process
trade_phase_list -- used to filtering, means that discovering
a list of head path with the trade_phase_list
"""
head_path_list = list()
for state in self.getStateValueList():
if len(state.getSuccessorRelatedValueList()) == 0:
head_path_list += state.getPredecessorRelatedValueList()
if trade_phase_list is not None:
_set = set(trade_phase_list)
_list = list()
# start to discover a head path with the trade_phase_list from head path(s) of whole
for path in head_path_list:
_list += self._getHeadPathValueList(path, _set)
head_path_list = map(lambda t: t[0], filter(lambda t: t != (None, None), _list))
return head_path_list
def _getHeadPathValueList(self, path, trade_phase_set):
# if the path has target trade_phase, it is a head path.
_set = set(path.getTradePhaseList())
if _set & trade_phase_set:
return [(path, _set & trade_phase_set)]
node = path.getSuccessorValue()
if node is None:
return [(None, None)]
_list = list()
for next_path in node.getPredecessorRelatedValueList():
_list += self._getHeadPathValueList(next_path, trade_phase_set)
return _list
def _getExpectedCompletionDate(self, path, *args, **kwargs):
return path.getExpectedStopDate(*args, **kwargs)
# From Business Path
def _getExplanationUidList(self, explanation):
"""
Helper method to fetch really explanation related movements
""" # XXX-JPS this seems to be doing too much - why do you need many "explanations"
explanation_uid_list = [explanation.getUid()]
# XXX: getCausalityRelatedValueList is oversimplification, assumes
# that documents are sequenced like simulation movements, which
# is wrong
if getattr(explanation, "getCausalityValueList", None) is None:
return explanation_uid_list
for found_explanation in explanation.getCausalityRelatedValueList( # XXX-JPS this also seems exagerated, and breaking the APIs
portal_type=self.getPortalDeliveryTypeList()) + \
explanation.getCausalityValueList(): # Wrong if an item
explanation_uid_list.extend(self._getExplanationUidList(
found_explanation))
return explanation_uid_list
product/ERP5/ExplanationCache.py
View file @ 9880c119
......@@ -51,12 +51,12 @@ class ExplanationCache:
"""
# Define share properties
self.explanation = explanation
self.portal_catalog = getToolByName(explanation 'portal_catalog')
self.portal_catalog = getToolByName(explanation, 'portal_catalog')
self.simulation_movement_cache = {} # Simulation Movement Cache
self.explanation_uid_cache = []
self.explanation_path_pattern_cache = []
def getDeliveryMovementList(self):
def _getDeliveryMovementList(self):
"""Returns self is explanation is a delivery line
of the list of explanation delivery lines if explanation
is a delivery
......@@ -80,7 +80,7 @@ class ExplanationCache:
return self.explanation_uid_cache
result = set()
# For each delivery movement
for movement in self.getDeliveryMovementList():
for movement in self._getDeliveryMovementList():
# For each simulation movement
for simulation_movement in movement.getDeliveryRelatedValueList():
result.add(simulation_movement.getExplanationUid()) # XXX-JPS use new API later
......@@ -117,7 +117,7 @@ class ExplanationCache:
local_path_dict[simulation_movement_id] = simulation_movement
# For each delivery movement
for movement in self.getDeliveryMovementList():
for movement in self._getDeliveryMovementList():
# For each simulation movement
for simulation_movement in movement.getDeliveryRelatedValueList():
updatePathDict(simulation_movement)
......@@ -133,7 +133,7 @@ class ExplanationCache:
else:
browsePathDict('%s/%s' % (prefix, key), value) # Recursing with string append is slow XXX-JPS
browsePathDict('', path_dict)
browsePathDict('/', path_dict)
self.explanation_path_pattern_cache = result
return result
......@@ -143,7 +143,7 @@ class ExplanationCache:
"""
return self.getSimulationMovementList(causality_uid=business_path.getUid())
def getSimulationMovementList(self, **kw)
def getSimulationMovementList(self, **kw):
"""Search Simulation Movements related to our explanation.
Cache result so that the second time we saarch for the same
list we need not involve the catalog again.
......@@ -157,9 +157,48 @@ class ExplanationCache:
**kw)
return self.simulation_movement_cache[kw_tuple]
def geBusinessPathClosure(business_path):
"""Creates a Business Process by filtering out all Business Path
in self which are not related to a simulation movement
which is either or parent or a child of explanation simulations movements
caused by 'business_path'
cache keys: business_path (simple) then path_set
XXX-JPS PSEUDO CODE
"""
new_business_process = BusinessProcess()
accepted_path = []
explanation_cache = _getExplanationCache(explanation)
path_list = explanation_cache.getSimulationPathPatternList()
path_list = map(lambda x:x[0:-1], path_list) # Remove trailing %
path_set = set()
for simulation_movement in business_path.\
_getExplanationRelatedSimulationMovementValueList(explanation):
simulation_path = simulation_movement.getPath()
for path in path_list:
if simulation_path.startswith(path):
path_set.add(path) # This selection path is part of explanation
for business_path in self.getBusinessPathValueList():
if business_path.hasMovementsIn(explanation, path_set):
accepted_path.append(business_path)
new_business_process.addValueList(business_path)
return new_business_process
def _getExplanationCache(explanation):
if explanation.isinstance(ExplanationCache):
return explanation
# Return cached value if any
tv = getTransactionalVariable(explanation)
if tv.get('explanation_cache', None) is None:
tv['explanation_cache'] = ExplanationCache(explanation)
return tv.get('explanation_cache')
def _getBusinessPathClosure(explanation, business_path):
"""Returns a
"""
explanation_cache = _getExplanationCache(explanation)
return explanation_cache.getBusinessPathClosure(business_path)
product/ERP5/interfaces/business_path.py
View file @ 9880c119
......@@ -59,6 +59,15 @@ class IBusinessPath(Interface):
movement -- a Simulation Movement
"""
def getCompletionDate(explanation):
"""Returns the date of completion of the movemnet
based on paremeters of the business path. This completion date can be
the start date, the stop date, the date of a given workflow transition
on the explaining delivery, etc.
XXXXXXXXXXXXXXXx
"""
def getExpectedQuantity(amount):
"""Returns the new quantity for the provided amount taking
into account the efficiency or the quantity defined on the business path.
......@@ -116,6 +125,10 @@ class IBusinessPath(Interface):
not yet completed (ex. in 'delivered' state).
"""
def isDelivered(explanation):
"""XXX
"""
def build(explanation):
"""Builds all related movements in the simulation using the builders
defined on the Business Path
......
product/ERP5/interfaces/business_process.py
View file @ 9880c119
......@@ -231,7 +231,7 @@ class ITradeStateProcess(Interface):
implicitely defines a simulation subtree
"""
def getLatestPartiallyCompletedTradeState(explanation):
def getLatestPartiallyCompletedTradeStateList(explanation):
"""Returns the list of completed trade states which predecessor
states are completed and for which no successor state
is partially completed in the context of given explanation.
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7