updated pub_sub with chunking code
This commit is contained in:
179
code snippets/chunked_sync_function_gateway.ipynb
Normal file
179
code snippets/chunked_sync_function_gateway.ipynb
Normal file
@@ -0,0 +1,179 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 1,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import json, time"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 6,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"payload = {\n",
|
||||
" \"Controller A\": [\n",
|
||||
" {\n",
|
||||
" \"ts\": 1700149800000,\n",
|
||||
" \"values\": {\n",
|
||||
" \"Temperature\": 22,\n",
|
||||
" \"Pressure\": 101.3,\n",
|
||||
" \"one\": 12,\n",
|
||||
" \"two\": 2,\n",
|
||||
" \"three\": 3,\n",
|
||||
" \"4\": 4,\n",
|
||||
" \"5\": 5,\n",
|
||||
" \"6\": 6,\n",
|
||||
" \"7\": 7,\n",
|
||||
" \"Temperature1\": 22,\n",
|
||||
" \"Pressure1\": 101.3,\n",
|
||||
" \"one1\": 12,\n",
|
||||
" \"two1\": 2,\n",
|
||||
" \"three1\": 3,\n",
|
||||
" \"41\": 4,\n",
|
||||
" \"51\": 5,\n",
|
||||
" \"61\": 6,\n",
|
||||
" \"71\": 7,\n",
|
||||
" \"Temperature2\": 22,\n",
|
||||
" \"Pressure2\": 101.3,\n",
|
||||
" \"one2\": 12,\n",
|
||||
" \"two2\": 2,\n",
|
||||
" \"three2\": 3,\n",
|
||||
" \"42\": 4,\n",
|
||||
" \"52\": 5,\n",
|
||||
" \"62\": 6,\n",
|
||||
" \"72\": 7,\n",
|
||||
" \"Temperature3\": 22,\n",
|
||||
" \"Pressure3\": 101.3,\n",
|
||||
" \"one3\": 12,\n",
|
||||
" \"two3\": 2,\n",
|
||||
" \"three3\": 3,\n",
|
||||
" \"43\": 4,\n",
|
||||
" \"53\": 5,\n",
|
||||
" \"63\": 6,\n",
|
||||
" \"73\": 7\n",
|
||||
" }\n",
|
||||
" }\n",
|
||||
" ],\n",
|
||||
" \"Controller B\": [\n",
|
||||
" {\n",
|
||||
" \"ts\": 1700149800000,\n",
|
||||
" \"values\": {\n",
|
||||
" \"Humidity\": 45\n",
|
||||
" }\n",
|
||||
" }\n",
|
||||
" ]\n",
|
||||
"}"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 3,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"attributes_payload = {\n",
|
||||
" \"Controller A\": {\n",
|
||||
" \"latestReportTime\": 1700149800000\n",
|
||||
" },\n",
|
||||
" \"Controller B\": {\n",
|
||||
" \"latestReportTime\": 1700149800000\n",
|
||||
" }\n",
|
||||
"}"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 10,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def chunk_payload(payload, chunk_size=20, is_attributes_payload=False):\n",
|
||||
" if is_attributes_payload:\n",
|
||||
" # For attributes payload, chunk the controllers\n",
|
||||
" controllers = list(payload.items())\n",
|
||||
" for i in range(0, len(controllers), chunk_size):\n",
|
||||
" yield dict(controllers[i:i + chunk_size])\n",
|
||||
" else:\n",
|
||||
" # For data payload, chunk the values within each controller\n",
|
||||
" for controller, data in payload.items():\n",
|
||||
" for entry in data:\n",
|
||||
" ts = entry['ts']\n",
|
||||
" values = entry['values']\n",
|
||||
" chunked_values = list(values.items())\n",
|
||||
" for i in range(0, len(chunked_values), chunk_size):\n",
|
||||
" yield {\n",
|
||||
" \"controller\": controller,\n",
|
||||
" \"ts\": ts,\n",
|
||||
" \"values\": dict(chunked_values[i:i + chunk_size])\n",
|
||||
" }\n",
|
||||
"\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 12,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"__topic__ {\"controller\": \"Controller A\", \"ts\": 1700149800000, \"values\": {\"Temperature\": 22, \"Pressure\": 101.3, \"one\": 12, \"two\": 2, \"three\": 3, \"4\": 4, \"5\": 5, \"6\": 6, \"7\": 7, \"Temperature1\": 22, \"Pressure1\": 101.3, \"one1\": 12, \"two1\": 2, \"three1\": 3, \"41\": 4, \"51\": 5, \"61\": 6, \"71\": 7, \"Temperature2\": 22, \"Pressure2\": 101.3}} __qos__\n",
|
||||
"__topic__ {\"controller\": \"Controller A\", \"ts\": 1700149800000, \"values\": {\"one2\": 12, \"two2\": 2, \"three2\": 3, \"42\": 4, \"52\": 5, \"62\": 6, \"72\": 7, \"Temperature3\": 22, \"Pressure3\": 101.3, \"one3\": 12, \"two3\": 2, \"three3\": 3, \"43\": 4, \"53\": 5, \"63\": 6, \"73\": 7}} __qos__\n",
|
||||
"__topic__ {\"controller\": \"Controller B\", \"ts\": 1700149800000, \"values\": {\"Humidity\": 45}} __qos__\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"for chunk in chunk_payload(payload=payload):\n",
|
||||
" print(\"__topic__\", json.dumps(chunk), \"__qos__\")\n",
|
||||
" time.sleep(2)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 11,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"__topic__ {\"Controller A\": {\"latestReportTime\": 1700149800000}, \"Controller B\": {\"latestReportTime\": 1700149800000}} __qos__\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"for chunk in chunk_payload(payload=attributes_payload, is_attributes_payload=True):\n",
|
||||
" print(\"__topic__\", json.dumps(chunk), \"__qos__\")\n",
|
||||
" time.sleep(2)"
|
||||
]
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": "tbDataCollector",
|
||||
"language": "python",
|
||||
"name": "python3"
|
||||
},
|
||||
"language_info": {
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"file_extension": ".py",
|
||||
"mimetype": "text/x-python",
|
||||
"name": "python",
|
||||
"nbconvert_exporter": "python",
|
||||
"pygments_lexer": "ipython3",
|
||||
"version": "3.10.5"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 2
|
||||
}
|
||||
Reference in New Issue
Block a user