Skip to content

Commit ecec3e8

Browse files
tsifttsift
committed
Python linting
1 parent ceffc0c commit ecec3e8

1 file changed

Lines changed: 35 additions & 34 deletions

File tree

‎python/docs/examples/ingestion.ipynb‎

Lines changed: 35 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,7 @@
147147
"source": [
148148
"async def advanced_flowbuilder_example():\n",
149149
" \"\"\"Example showing advanced FlowBuilderPy usage with channel indices for maximum performance.\"\"\"\n",
150-
" from sift_stream_bindings import FlowBuilderPy, TimeValuePy, ValuePy, ChannelIndexPy\n",
150+
" from sift_stream_bindings import FlowBuilderPy, TimeValuePy, ValuePy\n",
151151
"\n",
152152
" connection_config = SiftConnectionConfig(\n",
153153
" api_key=\"my_api_key\",\n",
@@ -188,7 +188,7 @@
188188
" # Get the mapping from channel names to ChannelIndexPy\n",
189189
" # This allows us to avoid hash lookups by using indices directly\n",
190190
" channel_index_map = descriptor.mapping()\n",
191-
" \n",
191+
"\n",
192192
" # Pre-compute channel indices and value conversion methods\n",
193193
" # This creates a list of (ChannelIndexPy, conversion_method) tuples\n",
194194
" # that can be reused for each flow, avoiding hash operations\n",
@@ -204,15 +204,15 @@
204204
" for i in range(10):\n",
205205
" # Create a FlowBuilderPy from the descriptor\n",
206206
" flow_builder = FlowBuilderPy(descriptor)\n",
207-
" \n",
207+
"\n",
208208
" # Attach the run ID directly to the flow builder\n",
209209
" flow_builder.attach_run_id(run_id)\n",
210-
" \n",
210+
"\n",
211211
" # Set channel values using set() with pre-computed indices\n",
212212
" # This avoids hash lookups and provides better performance\n",
213213
" motor_temp_value = 50.0 + random.random() * 5.0\n",
214214
" tank_pressure_value = 2000.0 + random.random() * 100.0\n",
215-
" \n",
215+
"\n",
216216
" # If the raw data class used provides in-order iteration over the raw data, you can also iterate\n",
217217
" # over the values and encoding information directly. Since the value indices are used, the\n",
218218
" # additional per-channel hash lookup is not needed, further improving performance.\n",
@@ -227,10 +227,10 @@
227227
" values = [motor_temp_value, tank_pressure_value]\n",
228228
" for (channel_index, conversion_method), value in zip(channel_indices_and_methods, values):\n",
229229
" flow_builder.set(channel_index, conversion_method(value))\n",
230-
" \n",
230+
"\n",
231231
" # Build the request with current timestamp\n",
232232
" request = flow_builder.request(TimeValuePy.now())\n",
233-
" \n",
233+
"\n",
234234
" # Send the request (non-blocking version)\n",
235235
" ingest_client.send_requests_nonblocking([request])\n",
236236
"\n",
@@ -267,6 +267,7 @@
267267
"async def high_performance_batch_example():\n",
268268
" \"\"\"Example showing high-performance batch sending with FlowBuilderPy using channel indices.\"\"\"\n",
269269
" from datetime import timedelta\n",
270+
"\n",
270271
" from sift_stream_bindings import FlowBuilderPy, TimeValuePy, ValuePy\n",
271272
"\n",
272273
" connection_config = SiftConnectionConfig(\n",
@@ -320,20 +321,20 @@
320321
"\n",
321322
" start_time = datetime.now(timezone.utc)\n",
322323
" requests = []\n",
323-
" \n",
324+
"\n",
324325
" for i in range(num_flows):\n",
325326
" # Calculate timestamp for each sample (spaced 0.1 seconds apart)\n",
326327
" timestamp_secs = int((start_time + timedelta(seconds=i / sample_rate_hz)).timestamp())\n",
327328
" timestamp = TimeValuePy.from_timestamp(timestamp_secs, 0)\n",
328-
" \n",
329+
"\n",
329330
" # Create FlowBuilderPy and build request using pre-computed indices\n",
330331
" flow_builder = FlowBuilderPy(descriptor)\n",
331332
" flow_builder.attach_run_id(run_id)\n",
332-
" \n",
333+
"\n",
333334
" # Generate values\n",
334335
" motor_temp_value = 50.0 + random.random() * 5.0\n",
335336
" tank_pressure_value = 2000.0 + random.random() * 100.0\n",
336-
" \n",
337+
"\n",
337338
" # Use indices directly - no hash operations!\n",
338339
" values = [motor_temp_value, tank_pressure_value]\n",
339340
" for (channel_index, conversion_method), value in zip(channel_indices_and_methods, values):\n",
@@ -377,11 +378,7 @@
377378
"metadata": {},
378379
"outputs": [],
379380
"source": [
380-
"import asyncio\n",
381-
"import random\n",
382-
"import time\n",
383381
"from dataclasses import dataclass\n",
384-
"from datetime import datetime, timezone\n",
385382
"\n",
386383
"from sift_stream_bindings import FlowBuilderPy, FlowDescriptorPy, TimeValuePy\n",
387384
"\n",
@@ -428,10 +425,10 @@
428425
" # Queues for the pipeline\n",
429426
" queue1: asyncio.Queue[RawDataMessage] = asyncio.Queue()\n",
430427
" queue2: asyncio.Queue[tuple[RawDataMessage, FlowDescriptorPy]] = asyncio.Queue()\n",
431-
" \n",
428+
"\n",
432429
" # Cache for flow descriptors (flow_name -> FlowDescriptorPy)\n",
433430
" descriptor_cache: dict[str, FlowDescriptorPy] = {}\n",
434-
" \n",
431+
"\n",
435432
" # Cache for flow configs (flow_name -> FlowConfig)\n",
436433
" # In a real scenario, you'd derive this from your raw data schema\n",
437434
" flow_config_cache: dict[str, FlowConfig] = {\n",
@@ -461,7 +458,7 @@
461458
" for i in range(20):\n",
462459
" # Simulate different flows arriving\n",
463460
" flow_name = \"onboard_sensors\" if i % 2 == 0 else \"navigation\"\n",
464-
" \n",
461+
"\n",
465462
" if flow_name == \"onboard_sensors\":\n",
466463
" raw_data = RawDataMessage(\n",
467464
" flow_name=flow_name,\n",
@@ -480,7 +477,7 @@
480477
" \"gps_lon\": -122.4194 + random.random() * 0.01,\n",
481478
" },\n",
482479
" )\n",
483-
" \n",
480+
"\n",
484481
" queue1.put_nowait(raw_data)\n",
485482
" await asyncio.sleep(0.1)\n",
486483
"\n",
@@ -497,22 +494,26 @@
497494
" continue\n",
498495
"\n",
499496
" flow_name = raw_data.flow_name\n",
500-
" \n",
497+
"\n",
501498
" # Check if descriptor is cached\n",
502499
" if flow_name not in descriptor_cache:\n",
503-
" \n",
504-
" # For this example, the flow configs are pre-defined above. \n",
505-
" # \n",
500+
"\n",
501+
" # For this example, the flow configs are pre-defined above.\n",
502+
" #\n",
506503
" # Though in practice, these would often be dynamically generated based on\n",
507504
" # the raw data schema.\n",
508505
" if flow_name not in flow_config_cache:\n",
509506
" raise ValueError(f\"Flow config not found for {flow_name}\")\n",
510-
" \n",
507+
"\n",
511508
" flow_config = flow_config_cache[flow_name]\n",
512-
" \n",
509+
"\n",
513510
" # Convert to Rust FlowConfigPy format\n",
514-
" from sift_stream_bindings import FlowConfigPy, ChannelConfigPy, ChannelDataTypePy\n",
515-
" \n",
511+
" from sift_stream_bindings import (\n",
512+
" ChannelConfigPy,\n",
513+
" ChannelDataTypePy,\n",
514+
" FlowConfigPy,\n",
515+
" )\n",
516+
"\n",
516517
" channel_configs_py = [\n",
517518
" ChannelConfigPy(\n",
518519
" name=ch.name,\n",
@@ -524,20 +525,20 @@
524525
" )\n",
525526
" for ch in flow_config.channels\n",
526527
" ]\n",
527-
" \n",
528+
"\n",
528529
" flow_config_py = FlowConfigPy(\n",
529530
" name=flow_config.name,\n",
530531
" channels=channel_configs_py,\n",
531532
" )\n",
532-
" \n",
533+
"\n",
533534
" # Register the new flow\n",
534535
" await ingest_client.add_new_flows([flow_config_py])\n",
535-
" \n",
536+
"\n",
536537
" # Get the descriptor and cache it\n",
537538
" descriptor = ingest_client.get_flow_descriptor(flow_name)\n",
538539
" descriptor_cache[flow_name] = descriptor\n",
539540
" print(f\"Registered new flow: {flow_name}\")\n",
540-
" \n",
541+
"\n",
541542
" # Push to Queue 2 with the descriptor\n",
542543
" await queue2.put((raw_data, descriptor_cache[flow_name]))\n",
543544
"\n",
@@ -556,15 +557,15 @@
556557
" # Create FlowBuilderPy and set values\n",
557558
" flow_builder = FlowBuilderPy(descriptor)\n",
558559
" flow_builder.attach_run_id(run_id)\n",
559-
" \n",
560+
"\n",
560561
" # Set all channel values from raw data.\n",
561562
" for channel_name, value in raw_data.channel_values.items():\n",
562563
" flow_builder.set_with_key(channel_name, value)\n",
563-
" \n",
564+
"\n",
564565
" # Convert timestamp to TimeValuePy\n",
565566
" timestamp_secs = int(raw_data.timestamp.timestamp())\n",
566567
" timestamp = TimeValuePy.from_timestamp(timestamp_secs, 0)\n",
567-
" \n",
568+
"\n",
568569
" # Build request and send\n",
569570
" request = flow_builder.request(timestamp)\n",
570571
" await ingest_client.send_requests([request])\n",

0 commit comments

Comments
 (0)