New Functionality : Support measurement intervals

/reviewed-on klaus/wendelin!10

---

Hint: this cherry-pick created a conflict in "ERP5Site_createDataAnalysisList.py"
with the following commit:
nexedi/wendelin@529d5430

The conflict has been resolved by accepting the solution by the cherry-picked commit,
and then afterwards moving the changed code from the above commit to the relevant position.

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,9 @@ 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()
     # 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 +39,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 = 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 = []
@@ -51,6 +51,7 @@ for movement in portal_catalog(query = query):
       transformation_list.append(transformation)
 
   for transformation in transformation_list:
+   
     is_shared_data_analysis = False
     # Check if analysis already exists
     data_analysis = portal_catalog.getResultValue(
@@ -87,6 +88,7 @@ for movement in portal_catalog(query = 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(
@@ -94,47 +96,61 @@ for movement in portal_catalog(query = 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
-          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())
+
+      # 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())
+            
           for related_movement in related_movement_list:
-            aggregate_set.update(related_movement.getAggregateSet())
-            if related_movement.getUse() == "big_data/ingestion/batch":
-              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":
+            #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():
+        # 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_query_dict = dict(
               portal_type=item_type,
               validation_state="validated",
@@ -146,27 +162,21 @@ for movement in portal_catalog(query = query):
             if data_analysis.getDestinationProjectValue() is not None:
               item_query_dict["item_project_relative_url"] = data_analysis.getDestinationProject()
             item = portal.portal_catalog.getResultValue(**item_query_dict)
-
-            if item is None:
-              module = portal.getDefaultModule(item_type)
-              item = module.newContent(portal_type = item_type,
-                                title = transformation.getTitle(),
-                                reference = "%s-%s" %(transformation.getTitle(),
-                                                      delivery.getReference()),
-                                version = '001')
-              try:
-                item.validate()
-              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())
+          
+          #if transformation_line.getRelativeUrl() == "data_transformation_module/woelfel_r0331_statistic_raw":
+          #  raise TypeError("JUST STOP")
+          if item is None:
+            module = portal.getDefaultModule(item_type)
+            item = module.newContent(portal_type = item_type,
+                              title = transformation.getTitle(),
+                              reference = "%s-%s" %(transformation.getTitle(),
+                                                    delivery.getReference()),
+                              version = '001')
+            try:
+              item.validate()
+            except AttributeError:
+              pass
+          aggregate_set.add(item.getRelativeUrl())
 
       data_analysis_line = data_analysis.newContent(
         portal_type = "Data Analysis Line",
@@ -179,12 +189,12 @@ for movement in portal_catalog(query = 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()