1import itertools
2import math
3import random
4import time
5from datetime import datetime
6
7
8# -------------------------------------------------------------
9# 1. Infinite Mathematical Generators
10# -------------------------------------------------------------
11def infinite_primes():
12 """Generates an unbounded sequence of prime numbers on-demand."""
13 yield 2
14 candidate = 3
15 primes_found = [2]
16
17 while True:
18 is_prime = True
19 limit = math.isqrt(candidate)
20 for p in primes_found:
21 if p > limit:
22 break
23 if candidate % p == 0:
24 is_prime = False
25 break
26
27 if is_prime:
28 primes_found.append(candidate)
29 yield candidate
30
31 candidate += 2
32
33
34def infinite_fibonacci():
35 """Generates an unbounded Fibonacci sequence."""
36 a, b = 0, 1
37 while True:
38 yield a
39 a, b = b, a + b
40
41
42# -------------------------------------------------------------
43# 2. Infinite IoT Sensor Telemetry Stream
44# -------------------------------------------------------------
45def iot_sensor_stream(device_id: str):
46 """Simulates a live IoT industrial sensor producing telemetry readings."""
47 reading_id = 1
48 base_temp = 45.0
49 base_pressure = 101.3
50
51 while True:
52 # Simulate realistic telemetry fluctuations
53 temperature = round(base_temp + random.uniform(-4.5, 6.0), 2)
54 pressure = round(base_pressure + random.uniform(-2.0, 3.5), 2)
55
56 # Flag alert conditions
57 status = "CRITICAL" if temperature > 50.0 else "NORMAL"
58
59 telemetry = {
60 "reading_id": reading_id,
61 "device_id": device_id,
62 "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
63 "temperature_c": temperature,
64 "pressure_kpa": pressure,
65 "status": status
66 }
67
68 yield telemetry
69 reading_id += 1
70
71
72# -------------------------------------------------------------
73# 3. Stream Utilities: Windowing & Chunking
74# -------------------------------------------------------------
75def chunk_stream(stream, chunk_size: int):
76 """Batches an infinite stream into fixed-size finite chunks."""
77 while True:
78 chunk = list(itertools.islice(stream, chunk_size))
79 if not chunk:
80 break
81 yield chunk
82
83
84# -------------------------------------------------------------
85# 4. Stream Consumer & Demonstration
86# -------------------------------------------------------------
87def run_streaming_demo():
88 print("=" * 65)
89 print(" INFINITE SEQUENCE & STREAM GENERATOR ENGINE")
90 print("=" * 65)
91
92 # 1. Pull first 10 Primes using itertools.islice
93 print("\n--- First 10 Prime Numbers from Infinite Stream ---")
94 prime_stream = infinite_primes()
95 first_10_primes = list(itertools.islice(prime_stream, 10))
96 print(f"Primes: {first_10_primes}")
97
98 # 2. Pull first 12 Fibonacci numbers
99 print("\n--- First 12 Fibonacci Numbers from Infinite Stream ---")
100 fib_stream = infinite_fibonacci()
101 first_12_fib = list(itertools.islice(fib_stream, 12))
102 print(f"Fibonacci: {first_12_fib}")
103
104 # 3. Processing IoT Telemetry Stream (Filtered Pipeline)
105 print("\n--- Processing Live IoT Telemetry (Capturing 3 Critical Alerts) ---")
106 sensor = iot_sensor_stream("THERMAL-SENS-09")
107
108 # Lazy filter pipeline for critical anomalies
109 alert_pipeline = (
110 packet for packet in sensor
111 if packet["status"] == "CRITICAL"
112 )
113
114 # Consumer loop pulling 3 anomalies
115 captured_alerts = []
116 for _ in range(3):
117 alert_packet = next(alert_pipeline)
118 captured_alerts.append(alert_packet)
119 print(f" [ALERT] #{alert_packet['reading_id']} | "
120 f"Time: {alert_packet['timestamp']} | "
121 f"Temp: {alert_packet['temperature_c']}°C (Threshold Exceeded!)")
122
123 # 4. Chunking Demonstration
124 print("\n--- Chunking Sensor Stream into Batches of 3 Packets ---")
125 batched_sensor = chunk_stream(iot_sensor_stream("DEVICE-B"), chunk_size=3)
126
127 # Process only 2 batches
128 for batch_idx, batch in enumerate(itertools.islice(batched_sensor, 2), start=1):
129 print(f"Batch #{batch_idx} ({len(batch)} items):")
130 for reading in batch:
131 print(f" -> ID: {reading['reading_id']} | Temp: {reading['temperature_c']}°C | Pressure: {reading['pressure_kpa']} kPa")
132
133 print("\nStream demonstration finished safely without memory exhaustion.")
134
135
136if __name__ == "__main__":
137 run_streaming_demo()