def handle_frame(data):
global last_ocr_time, last_optical_flow_time, last_location_update, prev_gray,
alerts, navigation_steps, prev_frame, last_sound_time
logger = [Link](f'CaregiverLogger_{[Link]("caregiver_username",
"unknown")}_{session["username"]}')
[Link]("Received video frame from client")
try:
if not isinstance(data, str) or ',' not in data or not
[Link]('data:image'):
[Link]("Invalid frame data format")
raise ValueError("Invalid frame format")
try:
img_data = base64.b64decode([Link](',')[1])
npimg = [Link](img_data, dtype=np.uint8)
frame = [Link](npimg, cv2.IMREAD_COLOR)
if frame is None:
raise ValueError("Failed to decode image")
frame = [Link](frame, 1)
frame = [Link](frame, (FRAME_WIDTH, FRAME_HEIGHT))
except Exception as e:
[Link](f"Frame decoding error: {str(e)}")
emit('error', {'message': 'Failed to decode frame'},
namespace='/video_feed')
return
gray = [Link](frame, cv2.COLOR_BGR2GRAY)
if [Link]() - last_location_update > location_update_interval:
try:
user_data["location"] = [Link]('me').latlng
user_data["indoor_location"] = recognize_place(frame)
[Link](f"Location updated: {user_data['location']}, indoor:
{user_data['indoor_location']}")
last_location_update = [Link]()
except Exception as e:
[Link](f"Location update error: {str(e)}")
obstacle_results = [[] for _ in range(4)]
def detect_obstacles(idx, func, name, obstacle_results):
try:
detected = func(frame, gray)
distance = estimate_distance(
100 if name in ["Speed breaker", "Pothole"] else 50 if name ==
"Curb" else 200,
[Link]()) if detected else None
obstacle_results[idx].append((name, detected, distance))
except Exception as e:
[Link](f"Error detecting {name}: {str(e)}")
obstacle_results[idx].append((name, False, None))
obstacle_threads = [
[Link](target=detect_obstacles, args=(0,
detect_speed_breakers, "Speed breaker", obstacle_results)),
[Link](target=detect_obstacles, args=(1, detect_potholes,
"Pothole", obstacle_results)),
[Link](target=detect_obstacles, args=(2, detect_curbs,
"Curb", obstacle_results)),
[Link](target=detect_obstacles, args=(3, detect_stairs,
"Stairs", obstacle_results))
]
for t in obstacle_threads:
[Link]()
for t in obstacle_threads:
[Link]()
obstacles = [result[0] for result in obstacle_results if result]
try:
color = detect_color(frame)
currency = detect_currency(frame)
signage = read_text(frame, gray)
light_level = detect_light_level(gray)
except Exception as e:
[Link](f"Feature detection error: {str(e)}")
color = currency = signage = light_level = None
try:
mp_image = [Link](image_format=[Link], data=frame)
detection_result = object_detector.detect(mp_image)
except Exception as e:
[Link](f"Object detection error: {str(e)}")
detection_result = type('obj', (), {'detections': []})()
crowd_density = analyze_crowd_density(detection_result)
crowd_direction = navigate_crowd(detection_result)
moving_objects = None
flow = None
if [Link]() - last_optical_flow_time > optical_flow_interval and
prev_frame is not None:
moving_objects, flow = compute_optical_flow(prev_frame, frame)
last_optical_flow_time = [Link]()
sound = detect_sound() if [Link]() - last_sound_time > sound_interval
else None
if sound:
[Link](f"Sound detected: {sound}")
last_sound_time = [Link]()
try:
pose_results = [Link](frame)
except Exception as e:
[Link](f"Pose estimation error: {str(e)}")
pose_results = type('obj', (), {'pose_landmarks': None})()
user_behavior = "No user detected"
person_behavior = ("No person detected", None)
if pose_results.pose_landmarks:
user_behavior, user_speed, user_gesture = analyze_user_behavior(
pose_results.pose_landmarks, FRAME_WIDTH, FRAME_HEIGHT,
user_positions, user_head_orientations, user_speeds
)
person_behavior = predict_person_behavior(pose_results.pose_landmarks,
prev_person_positions, FRAME_WIDTH)
if person_behavior[1] is not None:
prev_person_positions.append(person_behavior[1])
person_near_objects = detect_person_near_object(detection_result,
pose_results.pose_landmarks, FRAME_WIDTH, FRAME_HEIGHT)
depth = estimate_depth(pose_results.pose_landmarks, FRAME_WIDTH)
object_distance, object_direction = None, None
if user_data["find_object"]:
object_distance, object_direction = locate_object(detection_result,
user_data["find_object"])
scene_desc = describe_scene(detection_result, user_data["indoor_location"],
crowd_density)
emergency = detect_emergency(pose_results.pose_landmarks, detection_result,
crowd_density)
if user_data["destination"] and user_data["indoor_location"] != "unknown":
navigation_steps = navigate_indoors(user_data["indoor_location"],
user_data["destination"])
text = None
if [Link]() - last_ocr_time > ocr_interval:
text = read_text(frame, gray)
last_ocr_time = [Link]()
emergency_detected = False
for detection in detection_result.detections:
category = [Link][0]
distance = estimate_distance(detection.bounding_box.width,
category.category_name)
if distance and distance < 2 and [Link] > 0.7:
emergency_detected = True
[Link](f"Emergency detected: {category.category_name} too
close at {distance}m")
[Link](f"Warning: {category.category_name} detected at
{distance} meters")
speed_breaker_detected = any(o[1] for o in obstacles if o[0] == "Speed
breaker")
road_context = is_road_context(frame, speed_breaker_detected, gray)
overlays = {"objects": [], "navigation": []}
for detection in detection_result.detections:
bbox = detection.bounding_box
category = [Link][0]
label = classify_vehicle(category.category_name)
distance = estimate_distance([Link], category.category_name)
if distance and distance < 8 and [Link] > 0.7:
direction = get_direction(bbox, FRAME_WIDTH, FRAME_HEIGHT)
overlays["objects"].append({
"label": label,
"score": [Link],
"direction": direction,
"distance": round(distance, 1)
})
for obstacle, detected, distance in obstacles:
if detected and distance and distance < 5:
overlays["objects"].append({
"label": obstacle,
"score": 0.9,
"distance": round(distance, 1)
})
if navigation_steps and road_context:
overlays["navigation"].append({"type": "arrow", "direction": "left",
"x": 25, "y": 25})
frame_with_detections = draw_object_detections(frame, detection_result,
user_behavior, navigation_steps, road_context, crowd_density, moving_objects)
alerts = []
for obstacle, detected, distance in obstacles:
if detected and distance and distance < 5:
[Link](f"{obstacle} ahead at {distance} meters")
for person_near_object in person_near_objects: # Fixed variable
name for clarity
[Link](
f"Person {person_near_object['interaction']}
{person_near_object['object_type']} at {person_near_object['object_distance']}
meters"
)
if emergency:
[Link](f"Emergency: {emergency}")
emergency_detected = True
try: # Start of try block for frame encoding
_, buffer = [Link]('.jpg', frame_with_detections)
frame_b64 = base64.b64encode(buffer).decode('utf-8')
frame_data = f"data:image/jpeg;base64,{frame_b64}"
except Exception as e: # Properly formatted except clause
[Link](f"Frame encoding error: {str(e)}")
emit('error', {'message': 'Failed to encode frame'},
namespace='/video_feed')
return
try: # Separate try block for announce_detections
announce_detections(
detection_result,
user_behavior,
person_behavior,
text,
obstacles,
navigation_steps,
crowd_density,
moving_objects,
person_near_objects,
color,
currency,
signage,
light_level,
object_distance,
object_direction,
scene_desc,
)
except Exception as e: # Fixed: Removed extra parenthesis
[Link](f"Announce detections error: {str(e)}")
# Dashboard data preparation
dashboard_data = {
'user_behavior': user_behavior,
'alerts': alerts[:5],
'objects': [f"{o['label']} at {o['distance']}m
({o['direction']})" for o in overlays["objects"]],
'emergency': emergency,
'crowd_density': crowd_density[0],
'person_count': crowd_density[2],
'avg_distance': round(crowd_density[3], 1) if crowd_density[3]
else 0,
'scene': scene_desc,
'location': user_data["indoor_location"],
'timestamp': [Link]().strftime("%Y-%m-%d %H:%M:%S")
}
[Link](
'dashboard_update',
dashboard_data,
namespace=f'/caregiver_dashboard/{session["username"]}'
)
emit('frame', {
'image': frame_data,
'overlays': overlays,
'alerts': alerts,
'user_behavior': user_behavior,
'emergency': emergency
}, namespace='/video_feed')
prev_frame = [Link]()
if emergency_detected:
[Link](
'emergency',
{'message': emergency or "Emergency situation detected"},
namespace=f'/caregiver_dashboard/{session["username"]}'
)
except Exception as e: # Outer try-except for entire frame processing
[Link](f"Frame processing error: {str(e)}")
emit('error', {'message': str(e)}, namespace='/video_feed')