Compare commits

...

3 Commits

Author SHA1 Message Date
5f7af7574c sse iphone fix 4 2025-11-09 21:19:06 +01:00
0c73e36e82 sse iphone fix 2 2025-11-09 20:12:08 +01:00
01b60671db sse iphone fix 1 2025-11-09 20:05:35 +01:00
2 changed files with 312 additions and 146 deletions

View File

@@ -19,6 +19,13 @@ from packages.home_capabilities import LIGHT_VERSION, THERMOSTAT_VERSION, LightS
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# In-memory cache for last known device states
# Will be populated from Redis pub/sub messages
device_states: dict[str, dict[str, Any]] = {}
# Background task reference
background_task: asyncio.Task | None = None
app = FastAPI( app = FastAPI(
title="Home Automation API", title="Home Automation API",
description="API for home automation system", description="API for home automation system",
@@ -49,6 +56,77 @@ async def health() -> dict[str, str]:
return {"status": "ok"} return {"status": "ok"}
async def redis_state_listener():
"""Background task that listens to Redis pub/sub and updates state cache."""
redis_client = None
pubsub = None
try:
redis_url, redis_channel = get_redis_settings()
logger.info(f"Starting Redis state listener for channel {redis_channel}")
redis_client = await aioredis.from_url(redis_url, decode_responses=True)
pubsub = redis_client.pubsub()
await pubsub.subscribe(redis_channel)
logger.info("Redis state listener connected")
while True:
try:
message = await asyncio.wait_for(
pubsub.get_message(ignore_subscribe_messages=True),
timeout=1.0
)
if message and message["type"] == "message":
data = message["data"]
try:
state_data = json.loads(data)
if state_data.get("type") == "state" and state_data.get("device_id"):
device_id = state_data["device_id"]
payload = state_data.get("payload", {})
device_states[device_id] = payload
logger.debug(f"Updated state cache for {device_id}: {payload}")
except Exception as e:
logger.warning(f"Failed to parse state data: {e}")
except asyncio.TimeoutError:
pass # No message, continue
except asyncio.CancelledError:
logger.info("Redis state listener cancelled")
raise
except Exception as e:
logger.error(f"Redis state listener error: {e}")
finally:
if pubsub:
await pubsub.unsubscribe(redis_channel)
await pubsub.close()
if redis_client:
await redis_client.close()
@app.on_event("startup")
async def startup_event():
"""Start background tasks on application startup."""
global background_task
background_task = asyncio.create_task(redis_state_listener())
logger.info("Started background Redis state listener")
@app.on_event("shutdown")
async def shutdown_event():
"""Clean up background tasks on application shutdown."""
global background_task
if background_task:
background_task.cancel()
try:
await background_task
except asyncio.CancelledError:
pass
logger.info("Stopped background Redis state listener")
@app.get("/spec") @app.get("/spec")
async def spec() -> dict[str, dict[str, str]]: async def spec() -> dict[str, dict[str, str]]:
"""Capability specification endpoint. """Capability specification endpoint.
@@ -182,6 +260,16 @@ async def get_devices() -> list[DeviceInfo]:
] ]
@app.get("/devices/states")
async def get_device_states() -> dict[str, dict[str, Any]]:
"""Get current states of all devices from in-memory cache.
Returns:
dict: Dictionary mapping device_id to state payload
"""
return device_states
@app.get("/layout") @app.get("/layout")
async def get_layout() -> dict[str, Any]: async def get_layout() -> dict[str, Any]:
"""Get UI layout configuration. """Get UI layout configuration.
@@ -287,7 +375,13 @@ async def set_device(device_id: str, request: SetDeviceRequest) -> dict[str, str
async def event_generator(request: Request) -> AsyncGenerator[str, None]: async def event_generator(request: Request) -> AsyncGenerator[str, None]:
"""Generate SSE events from Redis Pub/Sub. """Generate SSE events from Redis Pub/Sub with Safari compatibility.
Safari-compatible features:
- Immediate retry hint on connection
- Regular heartbeats every 15s (comment-only, no data)
- Proper flushing after each yield
- Graceful disconnect handling
Args: Args:
request: FastAPI request object for disconnect detection request: FastAPI request object for disconnect detection
@@ -295,17 +389,28 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]:
Yields: Yields:
str: SSE formatted event strings str: SSE formatted event strings
""" """
redis_url, redis_channel = get_redis_settings() redis_client = None
redis_client = await aioredis.from_url(redis_url, decode_responses=True) pubsub = None
pubsub = redis_client.pubsub()
try: try:
await pubsub.subscribe(redis_channel) # Send retry hint immediately for EventSource reconnect behavior
logger.info(f"SSE client connected, subscribed to {redis_channel}") yield "retry: 2500\n\n"
# Create heartbeat tracking # Try to connect to Redis
redis_url, redis_channel = get_redis_settings()
try:
redis_client = await aioredis.from_url(redis_url, decode_responses=True)
pubsub = redis_client.pubsub()
await pubsub.subscribe(redis_channel)
logger.info(f"SSE client connected, subscribed to {redis_channel}")
except Exception as e:
logger.warning(f"Redis unavailable, running in heartbeat-only mode: {e}")
redis_client = None
pubsub = None
# Heartbeat tracking
last_heartbeat = asyncio.get_event_loop().time() last_heartbeat = asyncio.get_event_loop().time()
heartbeat_interval = 25 heartbeat_interval = 15 # Safari-friendly: shorter interval
while True: while True:
# Check if client disconnected # Check if client disconnected
@@ -313,29 +418,67 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]:
logger.info("SSE client disconnected") logger.info("SSE client disconnected")
break break
# Try to get message (non-blocking) # Try to get message from Redis (if available)
message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1) if pubsub:
try:
message = await asyncio.wait_for(
pubsub.get_message(ignore_subscribe_messages=True),
timeout=0.1
)
if message and message["type"] == "message":
data = message["data"]
logger.debug(f"Sending SSE message: {data[:100]}...")
# Update in-memory cache with latest state
try:
state_data = json.loads(data)
if state_data.get("type") == "state" and state_data.get("device_id"):
device_states[state_data["device_id"]] = state_data.get("payload", {})
except Exception as e:
logger.warning(f"Failed to parse state data for cache: {e}")
yield f"event: message\ndata: {data}\n\n"
last_heartbeat = asyncio.get_event_loop().time()
continue # Skip sleep, check for more messages immediately
except asyncio.TimeoutError:
pass # No message, continue to heartbeat check
except Exception as e:
logger.error(f"Redis error: {e}")
# Continue with heartbeats even if Redis fails
# Handle actual data messages # Sleep briefly to avoid busy loop
if message and message["type"] == "message": await asyncio.sleep(0.1)
data = message["data"]
logger.debug(f"Sending SSE message: {data[:100]}...")
yield f"event: message\ndata: {data}\n\n"
last_heartbeat = asyncio.get_event_loop().time()
else:
# No message, sleep a bit to avoid busy loop
await asyncio.sleep(0.1)
# Send heartbeat every 25 seconds # Send heartbeat if interval elapsed
current_time = asyncio.get_event_loop().time() current_time = asyncio.get_event_loop().time()
if current_time - last_heartbeat >= heartbeat_interval: if current_time - last_heartbeat >= heartbeat_interval:
yield "event: ping\ndata: heartbeat\n\n" # Comment-style ping (Safari-compatible, no event type)
yield ": ping\n\n"
last_heartbeat = current_time last_heartbeat = current_time
except asyncio.CancelledError:
logger.info("SSE connection cancelled by client")
raise
except Exception as e:
logger.error(f"SSE error: {e}")
raise
finally: finally:
await pubsub.unsubscribe(redis_channel) # Cleanup Redis connection
await pubsub.aclose() if pubsub:
await redis_client.aclose() try:
await pubsub.unsubscribe(redis_channel)
await pubsub.aclose()
except Exception as e:
logger.error(f"Error closing pubsub: {e}")
if redis_client:
try:
await redis_client.aclose()
except Exception as e:
logger.error(f"Error closing redis: {e}")
logger.info("SSE connection closed") logger.info("SSE connection closed")
@@ -343,23 +486,28 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]:
async def realtime_events(request: Request) -> StreamingResponse: async def realtime_events(request: Request) -> StreamingResponse:
"""Server-Sent Events endpoint for real-time updates. """Server-Sent Events endpoint for real-time updates.
Safari-compatible SSE implementation:
- Immediate retry hint (2.5s reconnect delay)
- Heartbeat every 15s using comment syntax ": ping"
- Proper Cache-Control headers
- No buffering (nginx compatibility)
- Graceful Redis fallback (heartbeat-only mode)
Args: Args:
request: FastAPI request object request: FastAPI request object
Returns: Returns:
StreamingResponse: SSE stream of Redis messages StreamingResponse: SSE stream with Redis messages and heartbeats
""" """
return StreamingResponse( return StreamingResponse(
event_generator(request), event_generator(request),
media_type="text/event-stream", media_type="text/event-stream",
headers={ headers={
"Cache-Control": "no-cache", "Cache-Control": "no-cache, no-transform",
"Connection": "keep-alive", "Connection": "keep-alive",
"X-Accel-Buffering": "no", # Disable nginx buffering "X-Accel-Buffering": "no", # Disable nginx buffering
} }
) )
return {"message": f"Command sent to {device_id}"}
def main() -> None: def main() -> None:

View File

@@ -737,12 +737,27 @@
// API_BASE injected from backend (supports Docker/K8s environments) // API_BASE injected from backend (supports Docker/K8s environments)
window.API_BASE = '{{ api_base }}'; window.API_BASE = '{{ api_base }}';
window.RUNTIME_CONFIG = window.RUNTIME_CONFIG || {};
// Helper function to construct API URLs // Helper function to construct API URLs
function api(url) { function api(url) {
return `${window.API_BASE}${url}`; return `${window.API_BASE}${url}`;
} }
// iOS/Safari Polyfill laden (nur wenn nötig)
(function() {
var isIOS = /iP(hone|od|ad)/.test(navigator.platform) ||
(navigator.userAgent.includes("Mac") && "ontouchend" in document);
if (isIOS && typeof window.EventSourcePolyfill === "undefined") {
var s = document.createElement("script");
s.src = "https://cdn.jsdelivr.net/npm/event-source-polyfill@1.0.31/src/eventsource.min.js";
s.onerror = function() {
console.warn("EventSource polyfill konnte nicht geladen werden");
};
document.head.appendChild(s);
}
})();
let eventSource = null; let eventSource = null;
let currentState = {}; let currentState = {};
let thermostatTargets = {}; let thermostatTargets = {};
@@ -999,167 +1014,170 @@
} }
} }
// Connect to SSE // Safari/iOS-kompatibler SSE Client mit Auto-Reconnect
let reconnectAttempts = 0; let reconnectDelay = 2500;
const maxReconnectDelay = 30000; // Max 30 seconds let reconnectTimer = null;
function connectSSE() { // Global handleSSE function für SSE-Nachrichten
// Close existing connection if any window.handleSSE = function(data) {
console.log('SSE message:', data);
addEvent(data);
// Update device state
if (data.type === 'state' && data.device_id && data.payload) {
const card = document.querySelector(`[data-device-id="${data.device_id}"]`);
if (!card) {
console.warn(`No card found for device ${data.device_id}`);
return;
}
// Check if it's a light
if (data.payload.power !== undefined) {
currentState[data.device_id] = data.payload.power;
updateDeviceUI(
data.device_id,
data.payload.power,
data.payload.brightness
);
}
// Check if it's a thermostat
if (data.payload.mode !== undefined || data.payload.target !== undefined || data.payload.current !== undefined) {
if (data.payload.mode !== undefined) {
thermostatModes[data.device_id] = data.payload.mode;
}
if (data.payload.target !== undefined) {
thermostatTargets[data.device_id] = data.payload.target;
}
updateThermostatUI(
data.device_id,
data.payload.current,
data.payload.target,
data.payload.mode
);
}
}
};
function cleanupSSE() {
if (eventSource) { if (eventSource) {
try { try {
eventSource.close(); eventSource.close();
} catch (e) { } catch(e) {
console.error('Error closing EventSource:', e); console.error('Error closing EventSource:', e);
} }
eventSource = null; eventSource = null;
} }
if (reconnectTimer) {
clearTimeout(reconnectTimer);
reconnectTimer = null;
}
}
function scheduleReconnect() {
if (reconnectTimer) return;
console.log(`Reconnecting in ${reconnectDelay}ms...`);
reconnectTimer = setTimeout(() => {
reconnectTimer = null;
connectSSE();
// Backoff bis 10s
reconnectDelay = Math.min(reconnectDelay * 2, 10000);
}, reconnectDelay);
}
function connectSSE() {
cleanupSSE();
console.log(`Connecting to SSE... (attempt ${reconnectAttempts + 1})`); const REALTIME_URL = (window.RUNTIME_CONFIG && window.RUNTIME_CONFIG.REALTIME_URL)
? window.RUNTIME_CONFIG.REALTIME_URL
: api('/realtime');
console.log('Connecting to SSE:', REALTIME_URL);
try { try {
eventSource = new EventSource(api('/realtime')); // Verwende Polyfill wenn verfügbar, sonst native EventSource
const EventSourceImpl = window.EventSourcePolyfill || window.EventSource;
eventSource = new EventSourceImpl(REALTIME_URL, {
withCredentials: false
});
eventSource.onopen = () => { eventSource.onopen = function() {
console.log('SSE connected successfully'); console.log('SSE connected successfully');
reconnectAttempts = 0; // Reset counter on successful connection reconnectDelay = 2500; // Reset backoff
document.getElementById('connection-status').textContent = 'Verbunden'; document.getElementById('connection-status').textContent = 'Verbunden';
document.getElementById('connection-status').className = 'status connected'; document.getElementById('connection-status').className = 'status connected';
}; };
eventSource.addEventListener('message', (e) => { eventSource.onmessage = function(evt) {
const data = JSON.parse(e.data); if (!evt || !evt.data) return;
console.log('SSE message:', data);
addEvent(data); // Heartbeats beginnen mit ":" -> ignorieren
if (typeof evt.data === "string" && evt.data.charAt(0) === ":") {
// Update device state return;
if (data.type === 'state' && data.device_id && data.payload) {
const card = document.querySelector(`[data-device-id="${data.device_id}"]`);
if (!card) {
console.warn(`No card found for device ${data.device_id}`);
return;
}
// Check if it's a light
if (data.payload.power !== undefined) {
currentState[data.device_id] = data.payload.power;
updateDeviceUI(
data.device_id,
data.payload.power,
data.payload.brightness
);
}
// Check if it's a thermostat
if (data.payload.mode !== undefined || data.payload.target !== undefined || data.payload.current !== undefined) {
if (data.payload.mode !== undefined) {
thermostatModes[data.device_id] = data.payload.mode;
}
if (data.payload.target !== undefined) {
thermostatTargets[data.device_id] = data.payload.target;
}
updateThermostatUI(
data.device_id,
data.payload.current,
data.payload.target,
data.payload.mode
);
}
} }
});
try {
const data = JSON.parse(evt.data);
if (window.handleSSE) {
window.handleSSE(data);
}
} catch (e) {
console.error('Error parsing SSE message:', e);
}
};
eventSource.addEventListener('ping', (e) => { eventSource.onerror = function(error) {
console.log('Heartbeat received');
});
eventSource.onerror = (error) => {
console.error('SSE error:', error, 'readyState:', eventSource?.readyState); console.error('SSE error:', error, 'readyState:', eventSource?.readyState);
document.getElementById('connection-status').textContent = 'Getrennt'; document.getElementById('connection-status').textContent = 'Getrennt';
document.getElementById('connection-status').className = 'status disconnected'; document.getElementById('connection-status').className = 'status disconnected';
if (eventSource) { // Safari/iOS verliert Netz beim App-Switch: ruhig reconnecten
try { scheduleReconnect();
eventSource.close();
} catch (e) {
console.error('Error closing EventSource on error:', e);
}
eventSource = null;
}
// Exponential backoff with max delay
reconnectAttempts++;
const delay = Math.min(
1000 * Math.pow(2, reconnectAttempts - 1),
maxReconnectDelay
);
console.log(`Reconnecting in ${delay}ms... (attempt ${reconnectAttempts})`);
setTimeout(connectSSE, delay);
}; };
} catch (error) { } catch (error) {
console.error('Failed to create EventSource:', error); console.error('Failed to create EventSource:', error);
document.getElementById('connection-status').textContent = 'Getrennt'; document.getElementById('connection-status').textContent = 'Getrennt';
document.getElementById('connection-status').className = 'status disconnected'; document.getElementById('connection-status').className = 'status disconnected';
scheduleReconnect();
reconnectAttempts++;
const delay = Math.min(
1000 * Math.pow(2, reconnectAttempts - 1),
maxReconnectDelay
);
setTimeout(connectSSE, delay);
} }
} }
// Safari/iOS specific: Reconnect when page becomes visible // Visibility-Change Handler für iOS App-Switch
document.addEventListener('visibilitychange', () => { document.addEventListener('visibilitychange', function() {
if (document.visibilityState === 'visible') { if (!document.hidden) {
console.log('Page visible, checking SSE connection...'); // Wenn wieder sichtbar & keine offene Verbindung: verbinden
// Only reconnect if connection is actually dead (CLOSED = 2) if (!eventSource || eventSource.readyState !== 1) {
if (!eventSource || eventSource.readyState === EventSource.CLOSED) { console.log('Page visible again, reconnecting SSE...');
console.log('SSE connection dead, forcing reconnect...');
reconnectAttempts = 0; // Reset for immediate reconnect
connectSSE(); connectSSE();
} else {
console.log('SSE connection OK, readyState:', eventSource.readyState);
} }
} }
}); });
// Safari/iOS specific: Reconnect on page focus // Start SSE connection
window.addEventListener('focus', () => {
console.log('Window focused, checking SSE connection...');
// Only reconnect if connection is actually dead (CLOSED = 2)
if (!eventSource || eventSource.readyState === EventSource.CLOSED) {
console.log('SSE connection dead, forcing reconnect...');
reconnectAttempts = 0; // Reset for immediate reconnect
connectSSE();
} else {
console.log('SSE connection OK, readyState:', eventSource.readyState);
}
});
// Initialize
connectSSE(); connectSSE();
// Load initial device states // Load initial device states
async function loadDevices() { async function loadDevices() {
try { try {
const response = await fetch(api('/devices')); const response = await fetch(api('/devices/states'));
const devices = await response.json(); const states = await response.json();
console.log('Loaded initial device states:', devices); console.log('Loaded initial device states:', states);
// Update UI with initial states // Update UI with initial states
devices.forEach(device => { for (const [deviceId, state] of Object.entries(states)) {
if (device.type === 'light' && device.state) { if (state.power !== undefined) {
currentState[device.id] = device.state.power; // It's a light
updateDeviceUI(device.id, device.state.power, device.state.brightness); currentState[deviceId] = state.power;
} else if (device.type === 'thermostat' && device.state) { updateDeviceUI(deviceId, state.power, state.brightness);
if (device.state.mode) thermostatModes[device.id] = device.state.mode; } else if (state.mode !== undefined || state.target !== undefined) {
if (device.state.target) thermostatTargets[device.id] = device.state.target; // It's a thermostat
updateThermostatUI(device.id, device.state.current, device.state.target, device.state.mode); if (state.mode) thermostatModes[deviceId] = state.mode;
if (state.target) thermostatTargets[deviceId] = state.target;
updateThermostatUI(deviceId, state.current, state.target, state.mode);
} }
}); }
} catch (error) { } catch (error) {
console.error('Failed to load initial device states:', error); console.error('Failed to load initial device states:', error);
} }