Skip to content

K
klaus_wendelin
Commit 12a03f27 authored by Eteri's avatar Eteri Committed by Eteri
Browse files
Options

erp5_wendelin: add ingestion script to append original data to data stream 

without modification.

Add default ingestion policy and parsing script to parse dotted reference
by sensor and data product

Add api to data stream for unpacking msgpack and extracting fluentd
style datetime
parent e571a68b
Expand all Show whitespace changes
Inline Side-by-side
Showing with 747 additions and 1 deletion
bt5/erp5_wendelin/DocumentTemplateItem/portal_components/document.erp5.DataStream.py
View file @ 12a03f27
......@@ -26,6 +26,9 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import msgpack
import struct
import numpy as np
from AccessControl import ClassSecurityInfo
from Products.ERP5Type import Permissions, PropertySheet
from erp5.component.document.BigFile import BigFile
......@@ -57,6 +60,33 @@ class DataStream(BigFile):
for chunk in data.iterate(start_offset, end_offset - start_offset):
yield chunk
def readMsgpackChunkList(self, start_offset, end_offset):
"""
Read chunks of msgpack data from a Data Stream and return (unpacked_data list, offset)
"""
unpacker = msgpack.Unpacker()
data = self._baseGetData()
pos = start_offset
data_list = []
for chunk in data.iterate(start_offset, end_offset - start_offset):
unpacker.feed(chunk)
while True:
pos = start_offset + unpacker.tell()
try:
#yield unpacker.unpack()
data_list.append(unpacker.unpack())
except msgpack.exceptions.OutOfData:
break
#raise StopIteration(pos)
return data_list, pos
def extractDateTime(self, date_time_holder):
if isinstance(date_time_holder, int):
return np.datetime64(date_time_holder, 's')
# if it is not in, we Expect msgpack.ExtType
s, ns = struct.unpack(">II", date_time_holder.data)
return np.datetime64(s, 's') + np.timedelta64(ns, 'ns')
def readChunkList(self, start_offset, end_offset):
"""
Read chunks of data from a Data Stream and return them.
......
bt5/erp5_wendelin/PathTemplateItem/portal_callables/DataIngestionLine_writeIngestionToDataStream.xml 0 → 100644
View file @ 12a03f27
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PyData Script" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_local_properties</string> </key>
<value>
<tuple>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>reference</string> </value>
</item>
<item>
<key> <string>type</string> </key>
<value> <string>string</string> </value>
</item>
</dictionary>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>comment</string> </value>
</item>
<item>
<key> <string>type</string> </key>
<value> <string>string</string> </value>
</item>
</dictionary>
</tuple>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>data_chunk, out_stream={}</string> </value>
</item>
<item>
<key> <string>comment</string> </key>
<value> <string>This script is used with default simple Wendelin model defined in erp5_wendelin_data.\n
For other models it will need adjustments.</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string>Ingestion operation script to be used with Fluentd Input Bin plugin: Append given data chunk to given data stream</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>DataIngestionLine_writeIngestionToDataStream</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>PyData Script</string> </value>
</item>
<item>
<key> <string>reference</string> </key>
<value> <string>DataIngestionLine_writeIngestionToDataStream</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>DataIngestionLine_writeIngestionToDataStream</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
bt5/erp5_wendelin/PathTemplateItem/portal_callables/IngestionPolicy_parseSimpleIngestionTag.py 0 → 100644
View file @ 12a03f27
sensor_reference = reference.split('.')[0]
data_product_reference = reference.split('.')[1]
return {'resource_reference' : data_product_reference,
'specialise_reference': sensor_reference,
'reference': sensor_reference }
bt5/erp5_wendelin/PathTemplateItem/portal_callables/IngestionPolicy_parseSimpleIngestionTag.xml 0 → 100644
View file @ 12a03f27
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PyData Script" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_local_properties</string> </key>
<value>
<tuple>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>reference</string> </value>
</item>
<item>
<key> <string>type</string> </key>
<value> <string>string</string> </value>
</item>
</dictionary>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>comment</string> </value>
</item>
<item>
<key> <string>type</string> </key>
<value> <string>string</string> </value>
</item>
</dictionary>
</tuple>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>reference</string> </value>
</item>
<item>
<key> <string>comment</string> </key>
<value> <string>tag example: sensor_1.sample-data</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string>tag example: sensor_1.sample-data</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>IngestionPolicy_parseSimpleIngestionTag</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>PyData Script</string> </value>
</item>
<item>
<key> <string>reference</string> </key>
<value> <string>IngestionPolicy_parseSimpleIngestionTag</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>IngestionPolicy_parseSimpleIngestionTag</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
bt5/erp5_wendelin/bt/template_keep_last_workflow_history_only_path_list
View file @ 12a03f27
portal_callables/DataIngestionLine_writeFluentdIngestionToDataStream
portal_callables/IngestionPolicy_parseSimpleFluentdTag
portal_callables/DataIngestionLine_writeIngestionToDataStream
portal_callables/IngestionPolicy_parseSimpleIngestionTag
\ No newline at end of file
bt5/erp5_wendelin/bt/template_keep_workflow_path_list
View file @ 12a03f27
......@@ -3,6 +3,7 @@ web_page_module/wendelin_js
portal_gadgets/WendelinInformationGadget
portal_ingestion_policies/default_http_json
portal_ingestion_policies/default_http_json/**
portal_ingestion_policies/wendelin-ingestion
data_supply_module/default_http_json
data_supply_module/default_http_json/**
data_product_module/default_http_json
......
bt5/erp5_wendelin/bt/template_path_list
View file @ 12a03f27
......@@ -4,7 +4,9 @@ data_supply_module/default_http_json
data_supply_module/default_http_json/**
portal_alarms/wendelin_handle_analysis
portal_callables/DataIngestionLine_writeFluentdIngestionToDataStream
portal_callables/DataIngestionLine_writeIngestionToDataStream
portal_callables/IngestionPolicy_parseSimpleFluentdTag
portal_callables/IngestionPolicy_parseSimpleIngestionTag
portal_categories/business_application
portal_categories/business_application/**
portal_categories/quantity_unit
......@@ -19,5 +21,6 @@ portal_gadgets/WendelinInformationGadget
portal_gadgets/WendelinInformationGadget/**
portal_ingestion_policies/default_http_json
portal_ingestion_policies/default_http_json/**
portal_ingestion_policies/wendelin-ingestion
web_page_module/wendelin_information_gadget.html
web_page_module/wendelin_js
\ No newline at end of file
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