def update_metrics()

in agora/shopper_insights_api/src/prometheus_metrics.py [0:0]

• 32 lines of code
• 15 McCabe index (conditional complexity)

    def update_metrics(self, all_camera_data):
        """Update metrics for all cameras"""
        for camera_name, detection_data in all_camera_data.items():
            # Update basic metrics
            self.total_persons.labels(camera=camera_name).set(detection_data['detected_persons'])
            self.fps.labels(camera=camera_name).set(detection_data['fps'])
            
            # Update age metrics
            for age, count in detection_data['age_stats'].items():
                metric = self._get_age_metric(age, camera_name)
                metric.labels(camera=camera_name).set(count)
            
            # Update area statistics
            for area_id, stats in detection_data['area_stats'].items():
                metric = self._get_area_metric(area_id, camera_name)
                self.total_shoppers.labels(camera=camera_name).set(stats['total_count'])
                self.current_shoppers.labels(camera=camera_name).set(stats['current_count'])
                metric.labels(camera=camera_name).set(stats['current_count'])
                metric.labels(camera=camera_name).set(stats['total_count'])

            # Update average time in area metrics per age group
            for area_id, age_data in detection_data['people_near_areas'].items():
                for times in age_data.items():
                    if not times or len(times) == 0:
                        continue

                    # Calcualte the Average time spent in area - Only consider valid times if the difference is greater than 1 second
                    valid_times = [time for time in times if isinstance(time, dict) and (time.get('end_time', 0) - time.get('start_time', 0)) > 1]
                    total_time = sum(time.get('end_time', 0) - time.get('start_time', 0) for time in valid_times)
                    total_entries = len(valid_times)
                    average_time = total_time / total_entries if total_entries > 0 else 0

                    # Get the Age group of the shoppers
                    age = times[1].get('age', 0)
                    age_group = int(age // 10) * 10

                    metric_key = f'time_avg_age_{age_group}_{camera_name.replace(" ", "_")}'
                    if metric_key not in self.time_in_area_avg_age:
                        self.time_in_area_avg_age[metric_key] = Gauge(
                            f'time_avg_age_{age_group}_{camera_name.replace(" ", "_")}',
                            f'Average time spent by shoppers in age group {age_group}',
                            ['camera']
                        )
                    metric = self.time_in_area_avg_age[metric_key]
                    metric.labels(camera=camera_name).set(average_time)