14. Real-Time vs. Periodic URL Generation Strategies
14.1 Real-Time Generation (Event-Driven)
When to Use:
- Critical alerts requiring immediate attention
- Security-related events
- Equipment failures
- Safety threshold violations
- Urgent maintenance requirements
Architecture:
IoT Event → Immediate Processing → aePiot URL → Instant NotificationImplementation Example:
python
class RealTimeURLGenerator:
"""Generate aePiot URLs immediately upon IoT events"""
def __init__(self):
self.notification_service = NotificationService()
def process_critical_event(self, event):
"""Process critical events in real-time"""
# Validate event criticality
if event['severity'] != 'CRITICAL':
return None
# Generate URL immediately
aepiot_url = self.generate_url(
title=f"CRITICAL: {event['type']}",
description=event['message'],
link=f"https://dashboard.example.com/events/{event['id']}"
)
# Distribute without delay
self.notification_service.send_urgent(
url=aepiot_url,
recipients=self.get_escalation_list(event['type']),
channels=['sms', 'push', 'email']
)
return aepiot_url
def generate_url(self, title, description, link):
"""Quick URL generation"""
from urllib.parse import quote
return f"https://aepiot.com/backlink.html?title={quote(title)}&description={quote(description)}&link={quote(link)}"14.2 Periodic Generation (Scheduled Reports)
When to Use:
- Daily/weekly status summaries
- Routine maintenance schedules
- Non-urgent monitoring reports
- Resource consumption reports
- Historical trend analysis
Architecture:
Scheduled Task → Aggregate Data → Generate Multiple URLs → Batch DistributionImplementation Example:
python
from datetime import datetime, timedelta
import schedule
from urllib.parse import quote
class PeriodicURLGenerator:
"""Generate aePiot URLs on schedule"""
def __init__(self):
self.device_manager = DeviceManager()
def daily_report(self):
"""Generate daily summary report with aePiot URLs"""
devices = self.device_manager.get_all_devices()
report_urls = []
for device in devices:
# Get 24-hour summary
summary = self.device_manager.get_daily_summary(device['id'])
title = quote(f"Daily Report - {device['name']}")
description = quote(
f"Uptime: {summary['uptime']}% | "
f"Events: {summary['event_count']} | "
f"Avg Value: {summary['avg_value']}"
)
link = quote(f"https://dashboard.example.com/reports/{device['id']}/daily/{datetime.now().date()}")
aepiot_url = f"https://aepiot.com/backlink.html?title={title}&description={description}&link={link}"
report_urls.append({
'device_id': device['id'],
'url': aepiot_url
})
# Send consolidated email with all URLs
self.send_daily_digest(report_urls)
return report_urls
def setup_schedule(self):
"""Configure scheduled tasks"""
# Daily report at 8 AM
schedule.every().day.at("08:00").do(self.daily_report)
# Weekly summary every Monday at 9 AM
schedule.every().monday.at("09:00").do(self.weekly_summary)
# Monthly maintenance review first day of month
schedule.every().day.at("07:00").do(self.check_monthly_maintenance)
def run_scheduler(self):
"""Run the scheduler"""
while True:
schedule.run_pending()
time.sleep(60)14.3 Hybrid Approach (Intelligent Triggering)
Best Practice: Combine both strategies with intelligent triggering:
python
class IntelligentURLGenerator:
"""Intelligently decide between real-time and batched generation"""
def __init__(self):
self.event_queue = []
self.last_batch_time = datetime.now()
self.batch_interval = timedelta(minutes=15)
def process_event(self, event):
"""Decide processing strategy based on event characteristics"""
# Critical events: immediate processing
if self.is_critical(event):
return self.process_immediately(event)
# Non-critical: add to batch queue
self.event_queue.append(event)
# Process batch if interval reached
if datetime.now() - self.last_batch_time >= self.batch_interval:
return self.process_batch()
return None
def is_critical(self, event):
"""Determine if event requires immediate processing"""
critical_conditions = [
event.get('severity') == 'CRITICAL',
event.get('type') in ['fire_alarm', 'security_breach', 'equipment_failure'],
event.get('value', 0) > event.get('critical_threshold', float('inf'))
]
return any(critical_conditions)
def process_immediately(self, event):
"""Process and distribute immediately"""
url = self.generate_url(event)
self.distribute_urgent(url, event)
return url
def process_batch(self):
"""Process queued events as batch"""
if not self.event_queue:
return []
urls = []
for event in self.event_queue:
url = self.generate_url(event)
urls.append(url)
self.distribute_batch(urls)
# Clear queue and reset timer
self.event_queue = []
self.last_batch_time = datetime.now()
return urls14.4 Rate Limiting and Throttling
Prevent URL Spam:
python
from collections import defaultdict
from datetime import datetime, timedelta
class RateLimitedGenerator:
"""Prevent excessive URL generation for same device"""
def __init__(self):
self.url_history = defaultdict(list)
self.rate_limit = 5 # Max URLs per device per hour
self.time_window = timedelta(hours=1)
def can_generate_url(self, device_id):
"""Check if URL generation is allowed"""
now = datetime.now()
# Remove old entries outside time window
self.url_history[device_id] = [
timestamp for timestamp in self.url_history[device_id]
if now - timestamp < self.time_window
]
# Check rate limit
if len(self.url_history[device_id]) >= self.rate_limit:
return False
return True
def generate_with_limit(self, device_id, title, description, link):
"""Generate URL only if within rate limit"""
if not self.can_generate_url(device_id):
print(f"Rate limit exceeded for device {device_id}")
return None
# Generate URL
from urllib.parse import quote
url = f"https://aepiot.com/backlink.html?title={quote(title)}&description={quote(description)}&link={quote(link)}"
# Record generation time
self.url_history[device_id].append(datetime.now())
return url15. Multi-Device and Fleet Management
15.1 Fleet-Wide URL Management
Complete Fleet Manager Implementation:
python
import sqlite3
from datetime import datetime
from urllib.parse import quote
import json
class IoTFleetManager:
"""Manage aePiot URLs for entire IoT device fleet"""
def __init__(self, db_path='fleet_urls.db'):
self.db = sqlite3.connect(db_path)
self.init_database()
def init_database(self):
"""Initialize database schema"""
cursor = self.db.cursor()
# Devices table
cursor.execute('''
CREATE TABLE IF NOT EXISTS devices (
device_id TEXT PRIMARY KEY,
device_type TEXT,
location TEXT,
installation_date TEXT,
status TEXT,
metadata TEXT
)
''')
# Generated URLs table
cursor.execute('''
CREATE TABLE IF NOT EXISTS generated_urls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_id TEXT,
timestamp TEXT,
event_type TEXT,
aepiot_url TEXT,
distributed BOOLEAN,
access_count INTEGER DEFAULT 0,
FOREIGN KEY (device_id) REFERENCES devices(device_id)
)
''')
# Access log table
cursor.execute('''
CREATE TABLE IF NOT EXISTS access_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url_id INTEGER,
access_timestamp TEXT,
user_id TEXT,
source TEXT,
FOREIGN KEY (url_id) REFERENCES generated_urls(id)
)
''')
self.db.commit()
def register_device(self, device_id, device_type, location, metadata=None):
"""Register new device in fleet"""
cursor = self.db.cursor()
cursor.execute('''
INSERT OR REPLACE INTO devices
(device_id, device_type, location, installation_date, status, metadata)
VALUES (?, ?, ?, ?, ?, ?)
''', (
device_id,
device_type,
location,
datetime.now().isoformat(),
'active',
json.dumps(metadata) if metadata else '{}'
))
self.db.commit()
# Generate permanent QR code for device
return self.generate_device_qr(device_id)
def generate_fleet_event_url(self, device_id, event_type, event_data):
"""Generate aePiot URL for device event"""
# Get device info
device_info = self.get_device_info(device_id)
if not device_info:
raise ValueError(f"Device {device_id} not found in fleet")
# Construct URL components
title = f"{event_type} - {device_info['device_type']} ({device_info['location']})"
description = self.format_event_description(event_type, event_data)
link = f"https://dashboard.example.com/fleet/{device_id}/events/{event_type}"
# Generate aePiot URL
aepiot_url = f"https://aepiot.com/backlink.html?title={quote(title)}&description={quote(description)}&link={quote(link)}"
# Store in database
cursor = self.db.cursor()
cursor.execute('''
INSERT INTO generated_urls
(device_id, timestamp, event_type, aepiot_url, distributed)
VALUES (?, ?, ?, ?, ?)
''', (device_id, datetime.now().isoformat(), event_type, aepiot_url, False))
self.db.commit()
return aepiot_url
def get_device_info(self, device_id):
"""Retrieve device information"""
cursor = self.db.cursor()
cursor.execute('SELECT * FROM devices WHERE device_id = ?', (device_id,))
row = cursor.fetchone()
if not row:
return None
return {
'device_id': row[0],
'device_type': row[1],
'location': row[2],
'installation_date': row[3],
'status': row[4],
'metadata': json.loads(row[5]) if row[5] else {}
}
def get_fleet_status_urls(self):
"""Generate status URLs for entire fleet"""
cursor = self.db.cursor()
cursor.execute('SELECT device_id FROM devices WHERE status = ?', ('active',))
devices = cursor.fetchall()
fleet_urls = []
for (device_id,) in devices:
device_info = self.get_device_info(device_id)
title = quote(f"Status - {device_info['device_type']} ({device_info['location']})")
description = quote(f"Device ID: {device_id} | Installation: {device_info['installation_date']}")
link = quote(f"https://dashboard.example.com/fleet/{device_id}")
aepiot_url = f"https://aepiot.com/backlink.html?title={title}&description={description}&link={link}"
fleet_urls.append({
'device_id': device_id,
'url': aepiot_url
})
return fleet_urls
def format_event_description(self, event_type, event_data):
"""Format event data into readable description"""
descriptions = {
'temperature_alert': f"Temperature: {event_data.get('value')}°F (Threshold: {event_data.get('threshold')}°F)",
'maintenance_due': f"Maintenance cycle {event_data.get('cycle')} required",
'battery_low': f"Battery: {event_data.get('battery')}%",
'offline': f"Device offline since {event_data.get('offline_since')}",
'online': f"Device back online at {event_data.get('online_at')}"
}
return descriptions.get(event_type, json.dumps(event_data))
def generate_device_qr(self, device_id):
"""Generate QR code for device access"""
import qrcode
device_info = self.get_device_info(device_id)
title = quote(f"{device_info['device_type']} - {device_info['location']}")
description = quote(f"Device ID: {device_id}")
link = quote(f"https://dashboard.example.com/fleet/{device_id}")
aepiot_url = f"https://aepiot.com/backlink.html?title={title}&description={description}&link={link}"
# Generate QR code
qr = qrcode.QRCode(version=1, box_size=10, border=4)
qr.add_data(aepiot_url)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
qr_path = f"qr_codes/device_{device_id}.png"
img.save(qr_path)
return qr_path
def get_device_url_history(self, device_id, days=30):
"""Get URL generation history for device"""
cursor = self.db.cursor()
cutoff_date = (datetime.now() - timedelta(days=days)).isoformat()
cursor.execute('''
SELECT timestamp, event_type, aepiot_url, access_count
FROM generated_urls
WHERE device_id = ? AND timestamp > ?
ORDER BY timestamp DESC
''', (device_id, cutoff_date))
history = []
for row in cursor.fetchall():
history.append({
'timestamp': row[0],
'event_type': row[1],
'url': row[2],
'access_count': row[3]
})
return history
def get_fleet_analytics(self):
"""Get analytics for entire fleet"""
cursor = self.db.cursor()
# Total devices
cursor.execute('SELECT COUNT(*) FROM devices')
total_devices = cursor.fetchone()[0]
# Active devices
cursor.execute('SELECT COUNT(*) FROM devices WHERE status = ?', ('active',))
active_devices = cursor.fetchone()[0]
# URLs generated last 24 hours
yesterday = (datetime.now() - timedelta(days=1)).isoformat()
cursor.execute('SELECT COUNT(*) FROM generated_urls WHERE timestamp > ?', (yesterday,))
recent_urls = cursor.fetchone()[0]
# Most active devices
cursor.execute('''
SELECT device_id, COUNT(*) as event_count
FROM generated_urls
GROUP BY device_id
ORDER BY event_count DESC
LIMIT 10
''')
top_devices = cursor.fetchall()
return {
'total_devices': total_devices,
'active_devices': active_devices,
'urls_24h': recent_urls,
'top_devices': [{'device_id': d[0], 'events': d[1]} for d in top_devices]
}
# Usage Example
fleet_manager = IoTFleetManager()
# Register devices
fleet_manager.register_device('TEMP-001', 'Temperature Sensor', 'Warehouse A')
fleet_manager.register_device('MOTION-002', 'Motion Detector', 'Entrance')
# Generate event URL
url = fleet_manager.generate_fleet_event_url(
'TEMP-001',
'temperature_alert',
{'value': 92, 'threshold': 85}
)
# Get fleet analytics
analytics = fleet_manager.get_fleet_analytics()
print(f"Fleet Analytics: {analytics}")15.2 Geographic Distribution Management
Multi-Location Fleet Manager:
python
class GeographicFleetManager(IoTFleetManager):
"""Manage fleet across multiple geographic locations"""
def __init__(self, db_path='geo_fleet.db'):
super().__init__(db_path)
self.init_location_tables()
def init_location_tables(self):
"""Add location-specific tables"""
cursor = self.db.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS locations (
location_id TEXT PRIMARY KEY,
location_name TEXT,
address TEXT,
contact_email TEXT,
contact_phone TEXT,
timezone TEXT
)
''')
self.db.commit()
def generate_location_summary_url(self, location_id):
"""Generate summary URL for all devices at location"""
cursor = self.db.cursor()
cursor.execute('SELECT * FROM devices WHERE location = ?', (location_id,))
devices = cursor.fetchall()
title = quote(f"Location Summary - {location_id}")
description = quote(f"Total Devices: {len(devices)} | Status: All Systems")
link = quote(f"https://dashboard.example.com/locations/{location_id}")
aepiot_url = f"https://aepiot.com/backlink.html?title={title}&description={description}&link={link}"
return aepiot_urlEnd of Part 4
Continue to Part 5 for industry-specific use cases, scaling considerations, and performance optimization.
Support Resources:
- For detailed implementation guidance: ChatGPT
- For complex integration scripts: Claude.ai
