def __init__()

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

• 46 lines of code
• 8 McCabe index (conditional complexity)

    def __init__(self, url, index, name, skip_fps, debug=False, enable_saving=True):
        self.url = url
        self.index = index
        self.processed_frame_queue = Queue(maxsize=10)
        self.vs = None
        self.name = name
        self.restricted_areas = []
        self.process_thread = None
        self.running = False

        self.enable_saving = enable_saving  # Flag to enable/disable saving frames and videos
        self.debug = debug
        self.skip_fps = skip_fps

        MODEL_PATH = os.getenv("MODEL_PATH", ".\\models")
        print(f"Model: {MODEL_PATH}")

        # Initialize YOLO model
        device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        print(f"Using device: {device}")
        
        yolo_model_path = os.path.join(MODEL_PATH, "yolov8n.pt")
        self.model = YOLO(yolo_model_path, task='detect').to(device)
        self.min_confidence = 0.6
        print(f"Model loaded on: {next(self.model.parameters()).device}")

        # Initialize OpenVINO age detection
        self.initialize_age_detection(os.path.join(MODEL_PATH, "age-gender-recognition-retail-0013.xml"))

        # Initialize save_lock for thread-safe operations
        self.save_lock = threading.Lock()

        # Directory to store person images
        self.person_image_dir = "/usr/src/app/detected_frames"
        if not os.path.exists(self.person_image_dir):
            os.makedirs(self.person_image_dir)
        
        # Directory to store GIFs
        self.videos_output_dir = "/usr/src/app/detected_persons/videos"
        if not os.path.exists(self.videos_output_dir):
            os.makedirs(self.videos_output_dir)
        
        self.processed_person_dirs = set()
        self.video_creation_thread = None

        # Initialize tracking and statistics
        self.restricted_areas = []
        self.area_stats = {}
        self.age_stats = collections.defaultdict(int)
        self.people_near_areas = {}
        
        # Performance monitoring
        self.fps = 0
        self.processing_times = collections.deque(maxlen=200)
        self.detected_persons = 0
        self.current_shoppers = 0
        self.shoppers = 0
        self.current_shoppers_hashes = []

        # Thread control
        self.processed_frame_queue = Queue(maxsize=32)
        self.last_activity = time.time()
        self.inactivity_threshold = 60

        # Age tracking
        self.age_stats = defaultdict(int)
        for age_group in range(10, 60, 10):
            self.age_stats[age_group] = 0