############################################################################## # # Copyright (c) 2009 Nexedi SA and Contributors. All Rights Reserved. # Jean-Paul Smets-Solanes <jp@nexedi.com> # Yusuke Muraoka <yusuke@nexedi.com> # # WARNING: This program as such is intended to be used by professional # programmers who take the whole responsability of assessing all potential # consequences resulting from its eventual inadequacies and bugs # End users who are looking for a ready-to-use solution with commercial # garantees and support are strongly adviced to contract a Free Software # Service Company # # This program is Free Software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. # ############################################################################## from Globals import InitializeClass, PersistentMapping from AccessControl import ClassSecurityInfo from Products.CMFCore.PortalFolder import ContentFilter from Products.ERP5Type import Permissions, PropertySheet, Constraint, Interface from Products.ERP5.Document.Path import Path from Products.ERP5.Document.BusinessProcess import BackTrack import zope.interface class BusinessPath(Path): """ The BusinessPath class embeds all information related to lead times and parties involved at a give phase of a business process. BusinessPath are also used as helper to build buildable movements. Here is the typical code of an alarm: Approach 1: explanation per explanation builder = portal_deliveries.default_order_builder for path in builder.getSpecialiseRelatedValueList() # or wharever category for explanation in portal_catalog(buildable=1, portal_type='Order'): path.build(explanation) Pros: easy explanation based approach Cons: buildable column added in delivery table reexpand of a finished order might generate remaining buildable Approach 2: isBuildable is indexed for SimulationMovements isBuildable() method is added to SimulationMovement Pros: global select is possible Cons: reindex of simulation is required slow indexing Approach 3: isBuildable is invoked during build "after select" process builder = portal_deliveries.default_order_builder for path in builder.getSpecialiseRelatedValueList() # or wharever category builder.build(causality_uid=path.getUid(),) # Select movemenents Pros: global select is possible Cons: global select retrieves long lists slow build Method 3 is best """ meta_type = 'ERP5 Business Path' portal_type = 'Business Path' isPredicate = 1 # Declarative security security = ClassSecurityInfo() security.declareObjectProtected(Permissions.AccessContentsInformation) # Declarative properties property_sheets = ( PropertySheet.Base , PropertySheet.XMLObject , PropertySheet.CategoryCore , PropertySheet.DublinCore , PropertySheet.Folder , PropertySheet.Comment , PropertySheet.Arrow , PropertySheet.Chain , PropertySheet.BusinessPath ) # Declarative interfaces zope.interface.implements(Interface.ICategoryAccessProvider, Interface.IArrow) # IBusinessPath Interface security.declareProtected(Permissions.AccessContentsInformation, 'getSourceBaseCategoryList') def getSourceBaseCategoryList(self): """ Returns all categories which are used to define the source of this Arrow """ # Naive implementation - we must use category groups instead return ('source', 'source_section', 'source_payment', 'source_project', ) security.declareProtected(Permissions.AccessContentsInformation, 'getDestinationBaseCategoryList') def getDestinationBaseCategoryList(self): """ Returns all categories which are used to define the destination of this Arrow """ # Naive implementation - we must use category groups instead return ('destination', 'destination_section', 'destination_payment', 'destination_project', ) # ICategoryAccessProvider overriden methods def _getCategoryMembershipList(self, category, **kw): """ Overriden in order to take into account dynamic arrow categories """ context = kw.get('context') result = Path._getCategoryMembershipList(self, category, **kw) if context is not None: dynamic_category_list = self._getDynamicCategoryList(context) dynamic_category_list= self._filterCategoryList(dynamic_category_list, category, **kw) # TODO: static categories should have priority over dynamic categories result = dynamic_category_list + result return result def _getAcquiredCategoryMembershipList(self, category, **kw): """ Overriden in order to take into account dynamic arrow categories """ context = kw.pop('context', None) result = Path._getAcquiredCategoryMembershipList(self, category, **kw) if context is not None: dynamic_category_list = self._getDynamicCategoryList(context) dynamic_category_list= self._filterCategoryList(dynamic_category_list, category, **kw) # TODO: static categories should have priority over dynamic categories result = dynamic_category_list + result return result def _filterCategoryList(self, category_list, category, spec=(), filter=None, portal_type=(), base=0, keep_default=1, checked_permission=None): """ XXX - implementation missing TBD - look at CategoryTool._buildFilter for inspiration """ return category_list # Dynamic context based categories def _getDynamicCategoryList(self, context): return self._getDynamicSourceCategoryList(context) \ + self._getDynamicDestinationCategoryList(context) def _getDynamicSourceCategoryList(self, context): method_id = self.getSourceMethodId() if method_id: method = getattr(self, method_id) return method(context) return [] def _getDynamicDestinationCategoryList(self, context): method_id = self.getDestinationMethodId() if method_id: method = getattr(self, method_id) return method(context) return [] # Core API def isBuildable(self, explanation): """ """ if self.isCompleted(explanation): return False # No need to build what was already built if self.isFrozen(explanation): return False # No need to build what is frozen predecessor = self.getPredecessorValue() if predecessor is None: return True # No predecessor, let's build if predecessor.isCompleted(explanation): return True return False def isPartiallyBuildable(self, explanation): """ Not sure if this will exist some day XXX """ def _getRelatedSimulationMovementList(self, explanation): """ """ return self.getCausalityRelatedValueList(portal_type='Simulation Movement', explanation_uid=explanation.getUid()) def isCompleted(self, explanation): """ Looks at all simulation related movements and checks the simulation_state of the delivery """ acceptable_state_list = self.getCompletedStateList() for movement in self._getRelatedSimulationMovementList(explanation): if movement.getSimulationState() not in acceptable_state_list: return False return True def isPartiallyCompleted(self, explanation): """ Looks at all simulation related movements and checks the simulation_state of the delivery """ acceptable_state_list = self.getCompletedStateList() for movement in self._getRelatedSimulationMovementList(explanation): if movement.getSimulationState() in acceptable_state_list: return True return False def isFrozen(self, explanation): """ Looks at all simulation related movements and checks if frozen """ movement_list = self._getRelatedSimulationMovementList(explanation) if len(movement_list) == 0: return False # Nothing to be considered as Frozen for movement in movement_list: if not movement.isFrozen(): return False return True def build(self, explanation): """ Build """ builder_list = self.getBuilderList() # Missing method for builder in builder_list: builder.build(causality_uid=self.getUid()) # This is one way of doing builder.build(movement_relative_url_list= self._getRelatedSimulationMovementList(explanation)) # Another way # Date calculation def getExpectedStartDate(self, explanation, predecessor_date=None, *args, **kwargs): """ Returns the expected start date for this path based on the explanation. predecessor_date -- if provided, computes the date base on the date value provided """ return self._getExpectedDate(explanation, self._getRootExplanationExpectedStartDate, self._getPredecessorExpectedStartDate, self._getSuccessorExpectedStartDate, predecessor_date=predecessor_date, *args, **kwargs) def _getRootExplanationExpectedStartDate(self, explanation, *args, **kwargs): if self.getParentValue().isStartDateReferential(): return explanation.getStartDate() else: return self.getExpectedStopDate(explanation, *args, **kwargs)\ - self.getLeadTime() def _getPredecessorExpectedStartDate(self, explanation, predecessor_date=None, *args, **kwargs): if predecessor_date is None: node = self.getPredecessorValue() if node is not None: predecessor_date = node.getExpectedCompletionDate(explanation, *args, **kwargs) if predecessor_date is not None: return predecessor_date + self.getWaitTime() def _getSuccessorExpectedStartDate(self, explanation, *args, **kwargs): node = self.getSuccessorValue() if node is not None: return node.getExpectedBeginningDate(explanation, *args, **kwargs)\ - self.getLeadTime() def getExpectedStopDate(self, explanation, predecessor_date=None, *args, **kwargs): """ Returns the expected stop date for this path based on the explanation. predecessor_date -- if provided, computes the date base on the date value provided """ return self._getExpectedDate(explanation, self._getRootExplanationExpectedStopDate, self._getPredecessorExpectedStopDate, self._getSuccessorExpectedStopDate, predecessor_date=predecessor_date, *args, **kwargs) def _getRootExplanationExpectedStopDate(self, explanation, *args, **kwargs): if self.getParentValue().isStopDateReferential(): return explanation.getStopDate() else: return self.getExpectedStartDate(explanation, *args, **kwargs)\ + self.getLeadTime() def _getPredecessorExpectedStopDate(self, explanation, *args, **kwargs): node = self.getPredecessorValue() if node is not None: return node.getExpectedCompletionDate(explanation, *args, **kwargs)\ + self.getWaitTime() + self.getLeadTime() def _getSuccessorExpectedStopDate(self, explanation, *args, **kwargs): node = self.getSuccessorValue() if node is not None: return node.getExpectedBeginningDate(explanation, *args, **kwargs) def _getExpectedDate(self, explanation, root_explanation_method, predecessor_method, successor_method, visited=None, *args, **kwargs): """ Returns the expected stop date for this path based on the explanation. root_explanation_method -- used when the path is root explanation predecessor_method --- used to get expected date of side of predecessor successor_method --- used to get expected date of side of successor visited -- only used to prevent infinite recursion internally """ if visited is None: visited = [] # mark the path as visited if self not in visited: visited.append(self) if self.isDeliverable(): return root_explanation_method( explanation, visited=visited, *args, **kwargs) predecessor_expected_date = None try: predecessor_expected_date = predecessor_method( explanation, visited=visited, *args, **kwargs) except BackTrack: pass successor_expected_date = None try: successor_expected_date = successor_method( explanation, visited=visited, *args, **kwargs) except BackTrack: pass if successor_expected_date is not None or \ predecessor_expected_date is not None: # return minimum expected date but it is not None if successor_expected_date is None: return predecessor_expected_date elif predecessor_expected_date is None: return successor_expected_date else: if predecessor_expected_date < successor_expected_date: return predecessor_expected_date else: return successor_expected_date