New Functionality : Support measurement intervals

/reviewed-on klaus/wendelin!10

add  "Id" to parameter dictionary

Pass causality_reference and reference to Data Operation script.

pass causality_relative_url to operation script to support multiple intervals with same sensor at same location

bt5/erp5_wendelin/SkinTemplateItem/portal_skins/erp5_wendelin/DataAnalysis_executeDataOperation.py

@@ -2,6 +2,7 @@ portal = context.getPortalObject()
 operation = None
 use = None
 parameter_dict = {}
+
 context.checkConsistency(fixit=True)
 initial_product = context.getSpecialiseValue(portal_type="Data Transformation").getResourceValue()
 for analysis_line in context.objectValues(portal_type="Data Analysis Line"):
@@ -28,6 +29,12 @@ for analysis_line in context.objectValues(portal_type="Data Analysis Line"):
       parameter[base_category] = analysis_line.getVariationCategoryItemList(
                                    base_category_list=(base_category,))[0][0]
     reference = analysis_line.getReference()
+
+    parameter["Start Date"] = analysis_line.getStartDate()
+    parameter["Stop Date"] = analysis_line.getStopDate()
+    parameter["causality_reference"] = analysis_line.getCausalityReference()
+    parameter["causality_relative_url"] = analysis_line.getCausality()
+    parameter["reference"] = analysis_line.getReference()
     # several lines with same reference wil turn the parameter into a list
     if reference in parameter_dict:
       if not isinstance(parameter_dict[reference], list):
@@ -35,7 +42,9 @@ for analysis_line in context.objectValues(portal_type="Data Analysis Line"):
       parameter_dict[reference].append(parameter)
     else:
       parameter_dict[reference] = parameter
+
 script_id = operation.getScriptId()
+
 out = getattr(operation_analysis_line, script_id)(**parameter_dict)
 
 if out == 1:

bt5/erp5_wendelin/SkinTemplateItem/portal_skins/erp5_wendelin/ERP5Site_createDataAnalysisList.py

@@ -4,7 +4,6 @@ from Products.ERP5Type.Errors import UnsupportedWorkflowMethod
 
 portal = context.getPortalObject()
 portal_catalog = portal.portal_catalog
-
 now = DateTime()
 
 if not include_delivered:
@@ -35,6 +34,7 @@ for movement in portal_catalog(query):
   delivery = movement.getParentValue()
   data_supply = delivery.getSpecialiseValue(portal_type="Data Supply")
   data_supply_list = delivery.getSpecialiseValueList(portal_type="Data Supply")
+
   composed_data_supply = data_supply.asComposedDocument()
   # Get applicable transformation
   transformation_list = []
@@ -48,6 +48,7 @@ for movement in portal_catalog(query):
     validation_state = "validated",
     resource_relative_url = movement.getResource()))
   for transformation in transformation_list:
+   
     is_shared_data_analysis = False
     # Check if analysis already exists
     data_analysis = portal_catalog.getResultValue(
@@ -84,6 +85,7 @@ for movement in portal_catalog(query):
                     destination = delivery.getDestination(),
                     destination_section = delivery.getDestinationSection(),
                     destination_project = delivery.getDestinationProject())
+
     data_analysis.checkConsistency(fixit=True)
     # create input and output lines
     for transformation_line in transformation.objectValues(
@@ -91,47 +93,61 @@ for movement in portal_catalog(query):
                      "Data Transformation Operation Line"]):
       resource = transformation_line.getResourceValue()
       quantity = transformation_line.getQuantity()
+
       if isinstance(quantity, tuple):
         quantity = quantity[0]
       # In case of shared data anylsis only add additional input lines
       if is_shared_data_analysis and quantity > -1:
         continue
+
       aggregate_set = set()
       # manually add device to every line
       aggregate_set.add(movement.getAggregateDevice())
-      if transformation_line.getPortalType() == \
-          "Data Transformation Resource Line":
-        # at the moment, we only check for positive or negative quantity
-        if quantity < 0:
-          # it is an input line
-          # aggregate transformed item from data ingestion batch related to our
-          # movement. If it is an input resource line, then we search for an
-          # ingestion line with the same resource. If it is an operation line
-          # then we search for an ingestion line with resource portal type
-          # Data Product
+
+      # If it is batch processing we additionally get items from the other
+      # batch movements and deliver the other batch movements
+      if transformation_line.getUse() == "big_data/ingestion/batch" and \
+        transformation_line.getPortalType() == \
+          "Data Transformation Resource Line" and quantity < 0:
         batch_relative_url = movement.getAggregateDataIngestionBatch()
+
         if batch_relative_url is not None:
           related_movement_list = portal_catalog(
             portal_type="Data Ingestion Line",
             aggregate_relative_url=batch_relative_url,
             resource_relative_url = resource.getRelativeUrl())
-          else:
-            # get related movements only from current data ingestion
-            related_movement_list = movement.getParentValue().searchFolder(
-              portal_type=["Data Ingestion Line", "Data Analysis Line"],
-              resource_relative_url = resource.getRelativeUrl())
+            
           for related_movement in related_movement_list:
-            aggregate_set.update(related_movement.getAggregateSet())
-            if related_movement.getUse() == "big_data/ingestion/batch":
+            #aggregate_set.update(related_movement.getAggregateSet())
             related_movement.getParentValue().deliver()
+
+
       # create new item based on item_type if it is not already aggregated
       aggregate_type_set = set(
         [portal.restrictedTraverse(a).getPortalType() for a in aggregate_set])
       for item_type in transformation_line.getAggregatedPortalTypeList():
-          # create item if it does note exist yet.
-          # Except if it is a Data Array Line, then it is currently created by
-          # data operation itself (probably this exception is inconsistent)
-          if item_type not in aggregate_type_set and item_type != "Data Array Line":
+        # if item is not yet aggregated to this line, search it by related project
+        # and source If the item is a data configuration or a device configuration
+        # then we do not care the workflow state nor the related resource, nor
+        # the variation nor the related sensor. Data Array Lines are created
+        # by Data Operation.
+
+        if item_type not in aggregate_type_set:
+        
+          if item_type in portal.getPortalDeviceConfigurationTypeList() + portal.getPortalDataConfigurationTypeList():
+            
+            if item_type == "Status Configuration":
+              item = None
+            
+            else:
+              item = portal.portal_catalog.getResultValue(
+                portal_type=item_type,
+                #validation_state="validated",
+                item_project_relative_url=delivery.getDestinationProject(),
+                item_source_relative_url=delivery.getSource())
+  
+          elif item_type != "Data Array Line":
+            
             item = portal.portal_catalog.getResultValue(
               portal_type=item_type,
               validation_state="validated",
@@ -140,6 +156,8 @@ for movement in portal_catalog(query):
               item_project_relative_url=data_analysis.getDestinationProject(),
               item_resource_uid=resource.getUid(),
               item_source_relative_url=data_analysis.getSource())
+          
+
           if item is None:
             module = portal.getDefaultModule(item_type)
             item = module.newContent(portal_type = item_type,
@@ -152,14 +170,6 @@ for movement in portal_catalog(query):
             except AttributeError:
               pass
           aggregate_set.add(item.getRelativeUrl())
-      # find other items such as device configuration and data configuration
-      # from data ingestion and data supply
-      composed = data_analysis.asComposedDocument()
-      line_list = [l for l in delivery.objectValues(portal_type="Data Ingestion Line")]
-      line_list +=  [l for l in composed.objectValues(portal_type="Data Supply Line")]
-      for line in line_list:
-        if line.getResourceValue().getPortalType() == "Data Operation":
-          aggregate_set.update(line.getAggregateList())
 
       data_analysis_line = data_analysis.newContent(
         portal_type = "Data Analysis Line",
@@ -172,12 +182,12 @@ for movement in portal_catalog(query):
         quantity_unit = transformation_line.getQuantityUnit(),
         use = transformation_line.getUse(),
         aggregate_set = aggregate_set)
-
       # for intput lines of first level analysis set causality and specialise
       if quantity < 0 and delivery.getPortalType() == "Data Ingestion":
         data_analysis_line.edit(
           causality_value = delivery,
           specialise_value_list = data_supply_list)
+    
     data_analysis.checkConsistency(fixit=True)
     try:
       data_analysis.start()