Skip to content

erp5
erp5
Commit 9b9c75c3 authored by Romain Courteaud's avatar Romain Courteaud
Browse files
Options

Rewrote the rule. 

Need to finish the date calculation.


git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@3421 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent 42534e1b
Show whitespace changes
Inline Side-by-side
Showing with 341 additions and 385 deletions
product/ERP5/Document/TransformationRule.py
View file @ 9b9c75c3
##############################################################################
#
# Copyright (c) 2002 Nexedi SARL and Contributors. All Rights Reserved.
# Copyright (c) 2002, 2005 Nexedi SARL and Contributors. All Rights Reserved.
# Jean-Paul Smets-Solanes <jp@nexedi.com>
# Romain Courteaud <romain@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
......@@ -32,6 +33,8 @@ from Products.CMFCore.utils import getToolByName
from Products.ERP5Type import Permissions, PropertySheet, Constraint, Interface
from Products.ERP5.Document.Rule import Rule
from Products.ERP5.Document.TransformationSourcingRule import\
TransformationSourcingRuleMixin
from zLOG import LOG
......@@ -40,91 +43,20 @@ class TransformationRule(Rule):
Order Rule object make sure an Order in the similation
is consistent with the real order
"""
# CMF Type Definition
meta_type = 'ERP5 Transformation Rule'
portal_type = 'Transformation Rule'
add_permission = Permissions.AddPortalContent
isPortalContent = 1
isRADContent = 1
# Declarative security
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.View)
# Default Properties
property_sheets = ( PropertySheet.Base
, PropertySheet.XMLObject
, PropertySheet.CategoryCore
, PropertySheet.DublinCore
)
# CMF Factory Type Information
factory_type_information = \
{ 'id' : portal_type
, 'meta_type' : meta_type
, 'description' : """\
An ERP5 Rule..."""
, 'icon' : 'rule_icon.gif'
, 'product' : 'ERP5'
, 'factory' : 'addTransformationRule'
, 'immediate_view' : 'rule_view'
, 'allow_discussion' : 1
, 'allowed_content_types': ()
, 'filter_content_types' : 1
, 'global_allow' : 1
, 'actions' :
( { 'id' : 'view'
, 'name' : 'View'
, 'category' : 'object_view'
, 'action' : 'rule_view'
, 'permissions' : (
Permissions.View, )
}
, { 'id' : 'list'
, 'name' : 'Object Contents'
, 'category' : 'object_action'
, 'action' : 'folder_contents'
, 'permissions' : (
Permissions.View, )
}
, { 'id' : 'print'
, 'name' : 'Print'
, 'category' : 'object_print'
, 'action' : 'rule_print'
, 'permissions' : (
Permissions.View, )
}
, { 'id' : 'metadata'
, 'name' : 'Metadata'
, 'category' : 'object_view'
, 'action' : 'metadata_edit'
, 'permissions' : (
Permissions.View, )
}
, { 'id' : 'translate'
, 'name' : 'Translate'
, 'category' : 'object_action'
, 'action' : 'translation_template_view'
, 'permissions' : (
Permissions.TranslateContent, )
}
)
}
security.declareProtected(Permissions.ModifyPortalContent, 'constructNewAppliedRule')
def constructNewAppliedRule(self, context, id=None):
"""
Creates a new applied rule which points to self
"""
my_applied_rule = Rule.constructNewAppliedRule(self, context, id=id)
resource = context.getDefaultResourceValue()
# Find my related transformation
transformation = resource.getDefaultResourceRelatedValue(portal_type = ('Transformation',))
if transformation is not None:
#LOG('Transformation Causality', 0, str(transformation.getId()))
my_applied_rule.setCausalityValue(transformation )
return my_applied_rule
# Class variable
simulation_movement_portal_type = "Simulation Movement"
security.declareProtected(Permissions.AccessContentsInformation, 'test')
def test(self, movement):
......@@ -134,188 +66,224 @@ An ERP5 Rule..."""
# Test if we must transform
# The test should actually be based on nodes, paths
# and capacities, which is not possible now
# so we just test if it is a "model" !
# and if it is being source from the workshop
#LOG('Test Transformation Rule', 0, '')
result = 1
# Only apply to Order applied rule
root_applied_rule = movement.getRootAppliedRule()
root_rule = root_applied_rule.getSpecialiseValue()
if root_rule is None:
return 0
if root_rule.getPortalType() != "Order Rule":
return 0
# Only apply to certain resources
resource = movement.getResourceValue()
if resource is None:
return 0
module = resource.aq_parent
if module.id == 'modele':
# This is the modele !
# We must also test the nodes
# if the source is a production node
source = movement.getSource()
if type(source) is type('a'):
if source.find('site/Piquage') >= 0 :
return 1
return 0
elif module.id == 'assortiment':
destination = movement.getDestination()
if type(destination) is type('a'):
if destination.find('site/Stock_PF/Gravelines') >= 0 :
source = movement.getSource()
if type(source) is type('a'):
if source.find('site/Stock_PF/Gravelines') >= 0 :
return 1
return 0
order = root_applied_rule.getCausalityValue()
root_movement = movement.getRootSimulationMovement()
# Test movement
if (root_rule is None) or\
(root_rule.getPortalType() != "Production Order Rule") or\
(order is None) or\
(movement.getResourceValue() is None) or\
(movement.getSourceValue() is None) or\
(movement.getResourceValue() != root_movement.getResourceValue()):
# We only produced what is asked on the Production Order
result = 0
else:
supply_chain = self.getSupplyChain(movement.getParent())
parent_supply_link = self.getCurrentSupplyLink(movement)
current_tranfo_link_list = supply_chain.\
getPreviousProductionSupplyLinkList(parent_supply_link)
length = len(current_tranfo_link_list)
if length == 0:
result = 0
elif length > 1:
result = 0
# XXX FIXME: implementation needed
raise "TransformationRuleError",\
"TransformationRule not able to use multiple SupplyLink."
return result
# Simulation workflow
security.declareProtected(Permissions.ModifyPortalContent, 'expand')
def expand(self, applied_rule, **kw):
"""
Expands the current movement downward.
-> new status -> expanded
An applied rule can be expanded only if its parent movement
is expanded.
"""
delivery_line_type = 'Simulation Movement'
# Get the order where we come from
my_transformation = applied_rule.getCausalityValue()
# No transformation defined
if my_transformation is None:
# Things stop here
applied_rule.diverge()
return
# Find production node
my_context_movement = applied_rule.getParent()
production_node = my_context_movement.getSource()
LOG('TransformationRule.expand my_context_movement.getPhysicalPath()',0,my_context_movement.getPhysicalPath())
LOG('TransformationRule.expand my_context_movement.getSource()',0,my_context_movement.getSource())
LOG('TransformationRule.expand my_context_movement.getTargetSource()',0,my_context_movement.getTargetSource())
production_section = my_context_movement.getSourceSection()
# Generate production and consumption lines
my_quantity = my_context_movement.getTargetQuantity()
#LOG('Transformation', 0, str(my_transformation))
# We used to call this with context = my_context_movement
# but it still has some issue which need to be fixed XXX As
# a temp solution, we use the dict based API, but it is not general enough
# and will causse errors on countinuous variations
# suspected bug cause is probably related to the use of REQUEST where it should not
# ie. we acquire some unwanted context
amount_list , grand_total_base_price, grand_total_source_base_price,\
grand_total_duration, grand_total_duration_france, grand_total_variated_base_price,\
grand_total_variated_source_base_price = my_transformation.getAggregatedAmountList(
REQUEST = {'categories': my_context_movement.getVariationCategoryList()} )
# Coramy Specific
# Create a line for the resource produced by the transformation
new_id = 'produced_resource'
produced_resource = applied_rule.get(new_id)
if produced_resource is None:
my_context_movement.portal_types.constructContent(
type_name = delivery_line_type,
container = applied_rule,
id = new_id,
) # quantity
lost_quantity = 0.0
parent_movement = applied_rule.getParent()
# Get production node and production section
production = parent_movement.getSource()
production_section = parent_movement.getSourceSection()
# Get the current supply link used to calculate consumed resource
# The current supply link is calculated from the parent AppliedRule.
supply_chain = self.getSupplyChain(parent_movement.getParent())
parent_supply_link = self.getCurrentSupplyLink(parent_movement)
current_supply_link_list = supply_chain.\
getPreviousProductionSupplyLinkList(parent_supply_link)
if len(current_supply_link_list) != 1:
# We shall no pass here.
# The test method returned a wrong value !
raise "TransformationRuleError",\
"Expand must not be called on %r" %\
applied_rule.getRelativeUrl()
else:
lost_quantity = produced_resource.getLostQuantity()
produced_resource = applied_rule[new_id]
produced_resource._edit(
start_date = my_context_movement.getStartDate(),
stop_date = my_context_movement.getStartDate(),
resource = my_context_movement.getResource(),
quantity = my_context_movement.getQuantity() + lost_quantity,
source_list = (),
source_section_list = (),
quantity_unit = my_context_movement.getQuantityUnit(),
destination_section = production_section,
destination = production_node,
deliverable = 1
)
# Mising quantity unit conversion for my_quantity !!!! XXXX
produced_resource.setVariationCategoryList(my_context_movement.getVariationCategoryList())
# Add lines
line_number = 0
acceptable_id_list = ['produced_resource']
# getRootAppliedRules is not defined
#production_order = self.getRootAppliedRule().getCausalityValue() # get the production order
production_order = applied_rule.getRootAppliedRule().getCausalityValue() # get the production order
filter_list = production_order.contentValues(filter={'portal_type': 'Amount Filter'})
for amount_line in amount_list:
# Apply each amount filter
for f in filter_list:
f.update(amount_line)
new_id = 'transformed_resource_%s' % line_number
transformed_resource = applied_rule.get(new_id)
if transformed_resource is None:
my_context_movement.portal_types.constructContent(
type_name = delivery_line_type,
container = applied_rule,
id = new_id,
) # quantity
transformed_resource = applied_rule[new_id]
#LOG("amount_line", 0, str(amount_line))
if amount_line['quantity'] != 0.0:
# Only create line if it is not 0.0
transformed_resource._edit(
start_date = my_context_movement.getStartDate(),
stop_date = my_context_movement.getStartDate(),
quantity = amount_line['quantity'] * my_quantity,
efficiency = amount_line['efficiency'],
resource_value = amount_line['resource'],
quantity_unit = amount_line['quantity_unit'],
source = production_node,
source_section = production_section,
destination_list = (),
deliverable = 1
)
LOG('TransformationRule.expand transformed_resource.getPhysicalPath()',0,transformed_resource.getPhysicalPath())
LOG('TransformationRule.expand transformed_resource.getTargetSource()',0,transformed_resource.getTargetSource())
LOG('TransformationRule.expand transformed_resource.getSource()',0,transformed_resource.getSource())
#LOG('RESOURCE', 0, str(amount_line['resource'].getRelativeUrl()))
#LOG('VC List', 0, str(amount_line['variation_category_list']))
#LOG('Quantity', 0, str(amount_line['quantity']))
#LOG('Co Quantity', 0, str(amount_line['converted_quantity']))
variation_category_list = amount_line['variation_category_list']
# Verify each category
category_list = []
for category in variation_category_list:
value = self.portal_simulation.resolveCategory(category)
if value is not None:
category_list += [category]
transformed_resource.setVariationCategoryList(category_list)
acceptable_id_list += [new_id]
LOG('TransformationRule.expand transformed_resource.getPhysicalPath()',0,transformed_resource.getPhysicalPath())
LOG('TransformationRule.expand transformed_resource.getTargetSource()',0,transformed_resource.getTargetSource())
LOG('TransformationRule.expand transformed_resource.getSource()',0,transformed_resource.getSource())
LOG('TransformationRule.expand transformed_resource.showDict()',0,transformed_resource.showDict())
line_number += 1
current_supply_link = current_supply_link_list[0]
# Generate produced movement
movement_dict = self._expandProducedResource(applied_rule,
production,
production_section,
current_supply_link)
# Generate consumed movement
consumed_mvt_dict = self._expandConsumedResource(applied_rule,
production,
production_section,
current_supply_link)
movement_dict.update(consumed_mvt_dict)
# Finally, build movement
self._buildMovementList(applied_rule, movement_dict)
# Expand each movement created
Rule.expand(self, applied_rule, **kw)
# Remove each movement not in the transformation
for movement in applied_rule.objectValues():
if movement.getId() not in acceptable_id_list:
movement.flushActivity(invoke=0)
applied_rule._delObject(movement.getId()) # XXXX Make sur this is not deleted if already in delivery
LOG('TransformationRule.expand movement.getPhysicalPath()',0,movement.getPhysicalPath())
LOG('TransformationRule.expand movement.getTargetSource()',0,movement.getTargetSource())
LOG('TransformationRule.expand movement.getSource()',0,movement.getSource())
LOG('TransformationRule.expand movement.getTargetSource',0,movement.getTargetSource)
LOG('TransformationRule.expand movement.showDict()',0,movement.showDict())
def _expandProducedResource(self, applied_rule, production,
production_section, current_supply_link):
"""
Produced resource.
Create a movement for the resource produced by the transformation.
Only one produced movement can be created.
"""
parent_movement = applied_rule.getParent()
produced_movement_dict = {
'pr': {
"resource": parent_movement.getResource(),
# XXX what is lost quantity ?
"quantity": parent_movement.getQuantity(),# + lost_quantity,
"quantity_unit": parent_movement.getQuantityUnit(),
"variation_category_list":\
parent_movement.getVariationCategoryList(),
"source_list": (),
"source_section_list": (),
"destination": production,
"destination_section": production_section,
"deliverable": 1,
# XXX FIXME date not implemented
# "start_date": parent_movement.getStartDate(),
# "stop_date": parent_movement.getStartDate(),
'causality_value': current_supply_link,
}
}
return produced_movement_dict
# Pass to base class
Rule.expand(self, applied_rule, **kw)
def _expandConsumedResource(self, applied_rule, production,
production_section, current_supply_link):
"""
Consumed resource.
Create a movement for each resource consumed by the transformation,
and for the previous variation of the produced resource.
"""
# Calculate all consumed resource
# Store each value in a dictionnary before created them.
# { movement_id: {property_name: property_value,} ,}
consumed_movement_dict = {}
parent_movement = applied_rule.getParent()
supply_chain = self.getSupplyChain(parent_movement.getParent())
# Consumed previous variation
previous_variation_dict = self._expandConsumedPreviousVariation(
applied_rule,
production,
production_section,
supply_chain,
current_supply_link)
consumed_movement_dict.update(previous_variation_dict)
# Consumed raw materials
raw_material_dict = self._expandConsumedRawMaterials(
applied_rule,
production,
production_section,
supply_chain,
current_supply_link)
consumed_movement_dict.update(raw_material_dict)
return consumed_movement_dict
def _expandConsumedPreviousVariation(self, applied_rule, production,
production_section, supply_chain,
current_supply_link):
"""
Create a movement for the previous variation of the produced resource.
"""
consumed_movement_dict = {}
parent_movement = applied_rule.getParent()
# First, calculate the previous variation of the previous resource
previous_ind_phase_list = supply_chain.\
getPreviousProductionIndustrialPhaseList(current_supply_link)
for ind_phase_value in previous_ind_phase_list:
ind_phase = ind_phase_value.getLogicalPath()
consumed_mvt_id = "%s_%s" % ("mr", ind_phase_value.getId())
consumed_movement_dict[consumed_mvt_id] = {
# XXX FIXME: Not yet implemented
# start_date = parent_movement.getStartDate(),
# stop_date = parent_movement.getStartDate(),
"resource": parent_movement.getResource(),
# XXX Is the quantity value correct ?
"quantity": parent_movement.getQuantity(),
"quantity_unit": parent_movement.getQuantityUnit(),
"destination_list": (),
"destination_section_list": (),
"source": production,
"source_section": production_section,
"deliverable": 1,
"variation_category_list": \
parent_movement.getVariationCategoryList(),
'causality_value': current_supply_link,
"industrial_phase": ind_phase}
return consumed_movement_dict
def _expandConsumedRawMaterials(self, applied_rule, production,
production_section, supply_chain,
current_supply_link):
"""
Create a movement for each resource consumed by the transformation,
"""
parent_movement = applied_rule.getParent()
# Calculate the context for getAggregatedAmountList
base_category_list = parent_movement.getVariationBaseCategoryList()
if "industrial_phase" in base_category_list:
# We do not want to get the industrial phase variation
base_category_list.remove("industrial_phase")
category_list = parent_movement.getVariationCategoryList(
base_category_list=base_category_list)
# Get the transformation to use
production_order = applied_rule.getRootAppliedRule().\
getCausalityValue()
transformation = production_order.getSpecialiseValue(
portal_type=self.getPortalTransformationTypeList())
# Generate the fake context
tmp_context = parent_movement.asContext(
context=parent_movement,
REQUEST={'categories':category_list})
# Calculate the industrial phase list
previous_ind_phase_list = supply_chain.\
getPreviousPackingListIndustrialPhaseList(current_supply_link)
ind_phase_id_list = [x.getId() for x in previous_ind_phase_list]
# Call getAggregatedAmountList
amount_list = transformation.getAggregatedAmountList(
tmp_context,
ind_phase_id_list=ind_phase_id_list)
# Add entries in the consumed_movement_dict
consumed_movement_dict = {}
for amount in amount_list:
consumed_mvt_id = "%s_%s" % ("cr", amount.getId())
consumed_movement_dict[consumed_mvt_id] = {
# XXX FIXME date not implemented
# "start_date": movement.getStartDate(),
# "stop_date": movement.getStartDate(),
"resource": amount.getResource(),
"variation_category_list":\
amount.getVariationCategoryList(),
"quantity": amount.getQuantity() * parent_movement.getQuantity(),
"quantity_unit": amount.getQuantityUnit(),
"destination_list": (),
"destination_section_list": (),
"source": production,
"source_section": production_section,
"deliverable": 1,
'causality_value': current_supply_link,
}
return consumed_movement_dict
security.declareProtected(Permissions.ModifyPortalContent, 'solve')
def solve(self, applied_rule, solution_list):
......@@ -361,3 +329,8 @@ An ERP5 Rule..."""
# Deliverability / orderability
def isDeliverable(self, m):
return 1
def isOrderable(self, m):
return 0
from Products.ERP5Type.Utils import monkeyPatch
monkeyPatch(TransformationSourcingRuleMixin, TransformationRule)
product/ERP5/Document/TransformationSourcingRule.py
View file @ 9b9c75c3
##############################################################################
#
# Copyright (c) 2002 Nexedi SARL and Contributors. All Rights Reserved.
# Copyright (c) 2002, 2005 Nexedi SARL and Contributors. All Rights Reserved.
# Jean-Paul Smets-Solanes <jp@nexedi.com>
# Romain Courteaud <romain@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
......@@ -26,6 +27,8 @@
#
##############################################################################
import ExtensionClass
from AccessControl import ClassSecurityInfo
from Acquisition import aq_base, aq_parent, aq_inner, aq_acquire
from Products.CMFCore.utils import getToolByName
......@@ -35,110 +38,107 @@ from Products.ERP5.Document.Rule import Rule
from zLOG import LOG
class TransformationSourcingRuleMixin(ExtensionClass.Base):
"""
Mixin class used by TransformationSourcingRule and TransformationRule
"""
# Declarative security
security = ClassSecurityInfo()
security.declareProtected(Permissions.View,
'getSupplyChain')
def getSupplyChain(self, applied_rule):
"""
Get the SupplyChain.
"""
# Get the SupplyChain to use
supply_chain_portal_type = "Supply Chain"
order = applied_rule.getRootAppliedRule().getCausalityValue()
supply_chain = order.getSpecialiseValue(
portal_type=supply_chain_portal_type)
if supply_chain is None:
raise "ProductionOrderError",\
"No SupplyChain defined on %s" % str(order)
else:
return supply_chain
def getCurrentSupplyLink(self, movement):
"""
Get the current SupplyLink
"""
# Get the current supply link
supply_link_portal_type = "Supply Link"
current_supply_link = movement.getCausalityValue(
portal_type=supply_link_portal_type)
return current_supply_link
security.declareProtected(Permissions.ModifyPortalContent,
'_buildMovementList')
def _buildMovementList(self, applied_rule, movement_dict):
"""
For each movement in the dictionnary, test if the movement already
exists.
If not, create it.
Then, update the movement attributes.
"""
for movement_id in movement_dict.keys():
movement = applied_rule.get(movement_id)
# Create the movement if it does not exist
if movement is None:
movement = applied_rule.newContent(
portal_type=self.simulation_movement_portal_type,
id=movement_id
)
# Update movement properties
movement.edit(**(movement_dict[movement_id]))
class TransformationSourcingRule(Rule):
"""
Transformation Sourcing Rule object make sure
items required in a Transformation are sourced
"""
# CMF Type Definition
meta_type = 'ERP5 Transformation Sourcing Rule'
portal_type = 'Transformation Sourcing Rule'
add_permission = Permissions.AddPortalContent
isPortalContent = 1
isRADContent = 1
# Declarative security
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.View)
# Default Properties
property_sheets = ( PropertySheet.Base
, PropertySheet.XMLObject
, PropertySheet.CategoryCore
, PropertySheet.DublinCore
)
# CMF Factory Type Information
factory_type_information = \
{ 'id' : portal_type
, 'meta_type' : meta_type
, 'description' : """\
An ERP5 Rule..."""
, 'icon' : 'rule_icon.gif'
, 'product' : 'ERP5'
, 'factory' : 'addTransformationSourcingRule'
, 'immediate_view' : 'rule_view'
, 'allow_discussion' : 1
, 'allowed_content_types': ()
, 'filter_content_types' : 1
, 'global_allow' : 1
, 'actions' :
( { 'id' : 'view'
, 'name' : 'View'
, 'category' : 'object_view'
, 'action' : 'rule_view'
, 'permissions' : (
Permissions.View, )
}
, { 'id' : 'list'
, 'name' : 'Object Contents'
, 'category' : 'object_action'
, 'action' : 'folder_contents'
, 'permissions' : (
Permissions.View, )
}
, { 'id' : 'print'
, 'name' : 'Print'
, 'category' : 'object_print'
, 'action' : 'rule_print'
, 'permissions' : (
Permissions.View, )
}
, { 'id' : 'metadata'
, 'name' : 'Metadata'
, 'category' : 'object_view'
, 'action' : 'metadata_edit'
, 'permissions' : (
Permissions.View, )
}
, { 'id' : 'translate'
, 'name' : 'Translate'
, 'category' : 'object_action'
, 'action' : 'translation_template_view'
, 'permissions' : (
Permissions.TranslateContent, )
}
)
}
# Class variable
simulation_movement_portal_type = "Simulation Movement"
security.declareProtected(Permissions.AccessContentsInformation, 'test')
def test(self, movement):
"""
Tests if the rule (still) applies
"""
# Test if some movements are "spendings" movements
# ie. with destination equal to None
destination = movement.getDestination()
if destination is not None:
return 0
# Is the resource sourceable (ie. tissue, composant in Coramy Case)
# This must become generic in the future through path XXXX
resource = movement.getDefaultResourceValue()
if resource is None:
return 0
module = resource.aq_parent
# Source components and workforce at this point
# This must become generic in the future through path XXXX
if module.id in ('tissu', 'composant'):
return 1
# We accept operations at this point
resource = movement.getDefaultResource()
if resource.find('operation/') >= 0:
return 1
return 0
# Test if we must transform
# The test should actually be based on nodes, paths
# and capacities, which is not possible now
result = 1
# Only apply to Order applied rule
root_applied_rule = movement.getRootAppliedRule()
root_rule = root_applied_rule.getSpecialiseValue()
order = root_applied_rule.getCausalityValue()
# Test some properties to see if we are really
# in a 'production' expand.
if (root_rule is None) or\
(root_rule.getPortalType() != "Production Order Rule") or\
(order is None) or\
(movement.getResourceValue() is None) or\
(movement.getSourceValue() is None):
result = 0
else:
supply_chain = self.getSupplyChain(movement.getParent())
parent_supply_link = self.getCurrentSupplyLink(movement)
if not supply_chain.test(parent_supply_link, movement):
result = 0
return result
# Simulation workflow
def reset(self, applied_rule):
......@@ -156,73 +156,50 @@ An ERP5 Rule..."""
def expand(self, applied_rule, **kw):
"""
Expands the current movement downward.
-> new status -> expanded
An applied rule can be expanded only if its parent movement
is expanded.
"""
delivery_line_type = 'Simulation Movement'
# Source that movement from the next node / stock
my_context_movement = applied_rule.getParent()
LOG('TransformationSourcingRule.expand, my_context_movement.getPhysicalPath()',0,my_context_movement.getPhysicalPath())
LOG('TransformationSourcingRule.expand, my_context_movement.getSource()',0,my_context_movement.getSource())
LOG('TransformationSourcingRule.expand, my_context_movement.getTargetSource()',0,my_context_movement.getTargetSource())
LOG('TransformationSourcingRule.expand, my_context_movement.showDict()',0,my_context_movement.showDict())
LOG('TransformationSourcingRule.expand, my_context_movement.getTargetSource',0,my_context_movement.getTargetSource)
if my_context_movement.getSource() is not None:
# We should only expand movements if they have a source
# otherwise, it creates infinite recursion
# This happens for example whenever the source of a movement is acquired
# from an order which is deleted afterwards
# LOG('Sourcing', 0, str(my_context_movement.getDefaultResource()))
new_id = 'transformation_source'
transformation_source = getattr(aq_base(applied_rule), new_id, None)
if transformation_source is None:
my_context_movement.portal_types.constructContent(
type_name = delivery_line_type,
container = applied_rule,
id = new_id
)
transformation_source = applied_rule[new_id]
resource = my_context_movement.getResource()
if resource.find('operation/') >= 0:
# This is an operation - produce it
transformation_source._edit(
target_quantity = my_context_movement.getTargetQuantity(),
target_efficiency = my_context_movement.getTargetEfficiency(),
resource = resource,
target_start_date = my_context_movement.getTargetStartDate(),
target_stop_date = my_context_movement.getTargetStartDate(),
target_source_list = (),
target_source_section_list = (),
quantity_unit = my_context_movement.getQuantityUnit(),
target_destination = my_context_movement.getTargetSource(),
target_destination_section = my_context_movement.getTargetSourceSection(),
deliverable = 0 # We do not need to source explicitely operations
)
transformation_source.setVariationCategoryList(
my_context_movement.getVariationCategoryList())
parent_movement = applied_rule.getParent()
# Calculate the previous supply link
supply_chain = self.getSupplyChain(parent_movement.getParent())
parent_supply_link = self.getCurrentSupplyLink(parent_movement)
previous_supply_link_list = supply_chain.\
getPreviousPackingListSupplyLinkList(
parent_supply_link,
movement=parent_movement)
if len(previous_supply_link_list) == 0:
raise "TransformationSourcingRuleError",\
"Expand must not be called on %r" %\
applied_rule.getRelativeUrl()
else:
# This is a component - source from Stock
transformation_source._edit(
target_quantity = my_context_movement.getTargetQuantity(),
target_efficiency = my_context_movement.getTargetEfficiency(),
resource = resource,
target_start_date = my_context_movement.getTargetStartDate(),
target_stop_date = my_context_movement.getTargetStartDate(),
target_source = 'site/Stock_MP/Gravelines',
target_source_section = 'group/Coramy',
quantity_unit = my_context_movement.getQuantityUnit(),
target_destination = my_context_movement.getTargetSource(),
target_destination_section = my_context_movement.getTargetSourceSection(),
deliverable = 1,
)
transformation_source.setVariationCategoryList(
my_context_movement.getVariationCategoryList())
movement_dict = {}
for previous_supply_link in previous_supply_link_list:
# Calculate the source
source_value = None
source_node = previous_supply_link.getSourceValue()
if source_node is not None:
source_value = source_node.getDestinationValue()
# Generate the dict
movement_dict.update({
"ts": {
'source_value': source_value,
'destination_value': parent_movement.getSourceValue(),
'resource_value': parent_movement.getResourceValue(),
'variation_category_list': parent_movement.\
getVariationCategoryList(),
'quantity': parent_movement.getQuantity(),
'quantity_unit': parent_movement.getQuantityUnit(),
# XXX FIXME not implemented
# 'start_date': parent_movement.getStartDate(),
# 'stop_date': parent_movement.getStopDate(),
'deliverable': 1,
# Save the value of the current supply link
'causality_value': previous_supply_link,
}
})
# Build the movement
self._buildMovementList(applied_rule, movement_dict)
# Create one submovement which sources the transformation
Rule.expand(self, applied_rule, **kw)
......@@ -276,3 +253,9 @@ An ERP5 Rule..."""
return 0
else:
return 1
def isOrderable(self, m):
return 0
from Products.ERP5Type.Utils import monkeyPatch
monkeyPatch(TransformationSourcingRuleMixin, TransformationSourcingRule)
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