Skip to content

Commit a70d8b7

Browse files
author
Tom Softreck
committed
update
1 parent fba158b commit a70d8b7

3 files changed

Lines changed: 180 additions & 22 deletions

File tree

‎src/dialogchain/engine.py‎

Lines changed: 53 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -260,28 +260,59 @@ async def run_route_config(self, route: Dict[str, Any], source: Source, destinat
260260
destination.sent_messages = []
261261

262262
# Process messages from source
263-
async for message in source.receive():
264-
if not self._is_running:
265-
break
266-
267-
try:
268-
# Process message through all processors
269-
processed_message = message
270-
for processor in processors:
271-
try:
272-
processed_message = await processor.process(processed_message)
273-
if processed_message is None:
274-
break # Message was filtered out
275-
except Exception as e:
276-
self.log(f"Error in processor {type(processor).__name__} in route {route_name}: {e}")
277-
processed_message = None
278-
break
279-
280-
# Send to destination if not filtered
281-
if processed_message is not None:
282-
try:
283-
await destination.send(processed_message)
284-
# Append to sent_messages for testing
263+
try:
264+
async for message in source.receive():
265+
if not self._is_running:
266+
self.log(f"🛑 Stopping route {route_name} (engine shutdown)")
267+
break
268+
269+
try:
270+
# Process message through all processors
271+
processed_message = message
272+
for processor in processors:
273+
try:
274+
if hasattr(processor, 'process') and callable(processor.process):
275+
self.log(f"⚙️ Processing message with {type(processor).__name__}")
276+
processed_message = await processor.process(processed_message)
277+
if processed_message is None:
278+
self.log("ℹ️ Message filtered out by processor")
279+
break # Message was filtered out
280+
except Exception as e:
281+
self.log(f"❌ Error in processor {type(processor).__name__} in route {route_name}: {e}")
282+
if self.verbose:
283+
import traceback
284+
self.log(traceback.format_exc())
285+
processed_message = None
286+
break
287+
288+
# Send to destination if not filtered
289+
if processed_message is not None:
290+
try:
291+
self.log(f"📤 Sending message to destination: {type(destination).__name__}")
292+
if hasattr(destination, 'send') and callable(destination.send):
293+
await destination.send(processed_message)
294+
# Store sent messages for testing/verification
295+
if hasattr(destination, 'sent_messages') and isinstance(destination.sent_messages, list):
296+
destination.sent_messages.append(processed_message)
297+
except Exception as e:
298+
self.log(f"❌ Error sending message to destination in route {route_name}: {e}")
299+
if self.verbose:
300+
import traceback
301+
self.log(traceback.format_exc())
302+
except Exception as e:
303+
self.log(f"❌ Error processing message in route {route_name}: {e}")
304+
if self.verbose:
305+
import traceback
306+
self.log(traceback.format_exc())
307+
except asyncio.CancelledError:
308+
self.log(f"🛑 Route {route_name} was cancelled")
309+
raise
310+
except Exception as e:
311+
self.log(f"❌ Error in message loop for route {route_name}: {e}")
312+
if self.verbose:
313+
import traceback
314+
self.log(traceback.format_exc())
315+
raise
285316
destination.sent_messages.append(processed_message)
286317
except Exception as e:
287318
self.log(f"Error sending message to destination in route {route_name}: {e}")

‎test_rtsp.py‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
import cv2
2+
import asyncio
3+
import os
4+
from dotenv import load_dotenv
5+
6+
async def test_rtsp_stream(rtsp_uri):
7+
"""Test RTSP stream connection and display first frame"""
8+
print(f"\n🔍 Testing RTSP connection to: {rtsp_uri}")
9+
10+
cap = cv2.VideoCapture(rtsp_uri)
11+
12+
if not cap.isOpened():
13+
print("❌ Failed to open RTSP stream")
14+
return False
15+
16+
print("✅ Successfully connected to RTSP stream")
17+
print("📊 Stream properties:")
18+
print(f" - Frame width: {int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))}")
19+
print(f" - Frame height: {int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))}")
20+
print(f" - FPS: {cap.get(cv2.CAP_PROP_FPS)}")
21+
22+
# Try to read one frame
23+
ret, frame = cap.read()
24+
if not ret:
25+
print("❌ Failed to read frame from stream")
26+
cap.release()
27+
return False
28+
29+
print("✅ Successfully read frame from stream")
30+
31+
# Save the frame as an image
32+
output_path = "test_frame.jpg"
33+
cv2.imwrite(output_path, frame)
34+
print(f"💾 Saved test frame to: {os.path.abspath(output_path)}")
35+
36+
cap.release()
37+
return True
38+
39+
if __name__ == "__main__":
40+
# Load environment variables
41+
load_dotenv()
42+
43+
# Get RTSP URI from environment or use a test one
44+
rtsp_uri = os.getenv("RTSP_URI", "rtsp://wowzaec2demo.streamlock.net/vod/mp4:BigBuckBunny_115k.mov")
45+
46+
print(f"🚀 Starting RTSP connection test...")
47+
success = asyncio.run(test_rtsp_stream(rtsp_uri))
48+
49+
if success:
50+
print("\n✅ RTSP test completed successfully!")
51+
else:
52+
print("\n❌ RTSP test failed. Check the error messages above.")
53+
54+
print("\n💡 If you're using your own camera, make sure to set the RTSP_URI in your .env file:")
55+
print(f"RTSP_URI=rtsp://username:password@camera-ip:port/stream")
56+
print("\nNote: The test used a public RTSP stream. For your camera, replace with actual credentials and IP.")

‎test_smtp.py‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
import smtplib
2+
import ssl
3+
import os
4+
from email.mime.text import MIMEText
5+
from email.mime.multipart import MIMEMultipart
6+
from dotenv import load_dotenv
7+
8+
def test_smtp_connection():
9+
"""Test SMTP server connection and authentication"""
10+
print("\n🔍 Testing SMTP connection...")
11+
12+
# Load environment variables
13+
load_dotenv()
14+
15+
# Get SMTP settings from environment
16+
smtp_server = os.getenv("SMTP_SERVER")
17+
smtp_port = int(os.getenv("SMTP_PORT", "587"))
18+
smtp_user = os.getenv("SMTP_USER")
19+
smtp_pass = os.getenv("SMTP_PASS")
20+
alert_email = os.getenv("ALERT_EMAIL")
21+
22+
if not all([smtp_server, smtp_user, smtp_pass, alert_email]):
23+
print("❌ Missing required SMTP environment variables. Please set these in your .env file:")
24+
print("SMTP_SERVER=smtp.example.com")
25+
print("SMTP_PORT=587")
26+
print("SMTP_USER=your_username")
27+
print("SMTP_PASS=your_password")
28+
print("ALERT_EMAIL=your@email.com")
29+
return False
30+
31+
try:
32+
# Create a secure SSL context
33+
context = ssl.create_default_context()
34+
35+
# Try to log in to server and send email
36+
with smtplib.SMTP(smtp_server, smtp_port) as server:
37+
print(f"✅ Connected to SMTP server: {smtp_server}:{smtp_port}")
38+
39+
# Start TLS encryption
40+
server.starttls(context=context)
41+
print("✅ Started TLS encryption")
42+
43+
# Login
44+
server.login(smtp_user, smtp_pass)
45+
print("✅ Successfully authenticated with SMTP server")
46+
47+
# Create message
48+
msg = MIMEMultipart()
49+
msg['From'] = smtp_user
50+
msg['To'] = alert_email
51+
msg['Subject'] = 'DialogChain SMTP Test'
52+
msg.attach(MIMEText('This is a test email from DialogChain SMTP test.', 'plain'))
53+
54+
# Send email
55+
server.send_message(msg)
56+
print(f"✅ Sent test email to: {alert_email}")
57+
58+
return True
59+
60+
except Exception as e:
61+
print(f"❌ SMTP test failed: {e}")
62+
return False
63+
64+
if __name__ == "__main__":
65+
print("🚀 Starting SMTP connection test...")
66+
success = test_smtp_connection()
67+
68+
if success:
69+
print("\n✅ SMTP test completed successfully!")
70+
else:
71+
print("\n❌ SMTP test failed. Check the error messages above.")

0 commit comments

Comments
 (0)