From 4edfda48a15e2010baf5b37aee8c4e503084ff73 Mon Sep 17 00:00:00 2001 From: Kunal-Somani Date: Sat, 21 Mar 2026 12:10:43 +0530 Subject: [PATCH] refactor(logging): migrate service classes to centralized get_logger() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove coloredlogs.install() calls from EmotionsAnalysisImp.__init__() and FirebaseImp.__init__() — reinstalling the log handler on every object instantiation was causing duplicate log entries - Replace instance-level self.logger with module-level logger via get_logger(__name__) for consistency with utils/logger.py (PR #21) - Switch all f-string log calls to % formatting (Python logging best practice — avoids string interpolation when log level is disabled) - Add debug log to FirebaseImp._initialize_app() for the already- initialized path - Add logger.info() to upload_to_firestore() on successful Firestore write - Remove unused logging and coloredlogs imports from both files --- services/data/firebase_imp.py | 29 ++++------- .../emotion_analysis/emotion_analysis_imp.py | 50 +++++++++---------- 2 files changed, 36 insertions(+), 43 deletions(-) diff --git a/services/data/firebase_imp.py b/services/data/firebase_imp.py index da8ca94..4838730 100644 --- a/services/data/firebase_imp.py +++ b/services/data/firebase_imp.py @@ -1,51 +1,44 @@ import firebase_admin -import logging -import coloredlogs - from firebase_admin import firestore, storage from services.data.firebase_service import FirebaseService +from utils.logger import get_logger +logger = get_logger(__name__) -class FirebaseImp(FirebaseService): - logger = logging.getLogger(__name__) +class FirebaseImp(FirebaseService): def __init__(self, storage_bucket: str): self.storage_bucket = storage_bucket self._initialize_app() - - coloredlogs.install( - level="INFO", - fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s" - ) - - self.logger = logging.getLogger(__name__) self.db = firestore.client() self.storage_client = storage.bucket() def _initialize_app(self): if not firebase_admin._apps: + logger.info("Initializing Firebase app with bucket: %s", self.storage_bucket) firebase_admin.initialize_app( options={ "storageBucket": self.storage_bucket } ) + else: + logger.debug("Firebase app already initialized, skipping.") - def download_video_from_storage(self, video_name: str): - self.logger.info(f"Attempting to download video: {video_name} from storage.") - + def download_video_from_storage(self, video_name: str) -> str: + logger.info("Attempting to download video: %s from storage.", video_name) blob = self.storage_client.blob(video_name) video_path = f"static/videos/{video_name}" blob.download_to_filename(video_path) - - self.logger.info(f"Video downloaded successfully to: {video_path}") + logger.info("Video downloaded successfully to: %s", video_path) return video_path - def upload_to_firestore(self, data): + def upload_to_firestore(self, data) -> str: doc_ref = self.db.collection("VideoAnalysis").document() formatted_data = { f"fragment{idx + 1}": result_dict for idx, result_dict in enumerate(data) } doc_ref.set(formatted_data) + logger.info("Uploaded analysis results to Firestore document: %s", doc_ref.id) return doc_ref.id diff --git a/services/emotion_analysis/emotion_analysis_imp.py b/services/emotion_analysis/emotion_analysis_imp.py index 69f751b..539955a 100644 --- a/services/emotion_analysis/emotion_analysis_imp.py +++ b/services/emotion_analysis/emotion_analysis_imp.py @@ -1,42 +1,44 @@ import os from schemas.emotion_schema import GetEmotionPercentagesResponse from services.emotion_analysis.emotion_analysis_service import EmotionsAnalysisService -import logging -import coloredlogs +from utils.logger import get_logger from utils.utils import load_model, load_face_cascade, extract_features, predict_emotion, getPercentages import cv2 +logger = get_logger(__name__) + + class EmotionsAnalysisImp(EmotionsAnalysisService): def __init__(self, model_path: str): self.model = load_model(model_path) self.face_cascade = load_face_cascade() - coloredlogs.install(level="INFO", fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s") - self.logger = logging.getLogger(__name__) def get_emotion_percentages(self, video_path: str) -> GetEmotionPercentagesResponse: predictions = [] labels = {0: 'Angry', 1: 'Disgusted', 2: 'Fearful', 3: 'Happy', 4: 'Neutral', 5: 'Sad', 6: 'Surprised'} - self.logger.info(f"Loading video from path: {video_path}") + logger.info("Loading video from path: %s", video_path) if not os.path.exists(video_path): - self.logger.error(f"Video file does not exist: {video_path}") + logger.error("Video file does not exist: %s", video_path) directory = os.path.dirname(video_path) if os.path.exists(directory): - self.logger.info(f"Contents of the directory {directory}:") + logger.info("Contents of directory %s:", directory) for item in os.listdir(directory): - self.logger.info(f" - {item}") + logger.info(" - %s", item) else: - self.logger.error(f"Directory does not exist: {directory}") - return GetEmotionPercentagesResponse(Angry=0, Disgusted=0, Fearful=0, Happy=0, Neutral=0, Sad=0, Surprised=0) + logger.error("Directory does not exist: %s", directory) + return GetEmotionPercentagesResponse( + Angry=0, Disgusted=0, Fearful=0, Happy=0, Neutral=0, Sad=0, Surprised=0 + ) video = cv2.VideoCapture(video_path) if not video.isOpened(): - self.logger.error(f"Failed to open video file: {video_path}") - return GetEmotionPercentagesResponse(Angry=0, Disgusted=0, Fearful=0, Happy=0, Neutral=0, Sad=0, Surprised=0) + logger.error("Failed to open video file: %s", video_path) + return GetEmotionPercentagesResponse( + Angry=0, Disgusted=0, Fearful=0, Happy=0, Neutral=0, Sad=0, Surprised=0 + ) last_processed_second = -1 - - frame_count = 0 processed_frames = 0 face_count = 0 @@ -47,14 +49,13 @@ def get_emotion_percentages(self, video_path: str) -> GetEmotionPercentagesRespo break timestamp_ms = video.get(cv2.CAP_PROP_POS_MSEC) - current_second = int(timestamp_ms / 500 ) # 2 frame per second + current_second = int(timestamp_ms / 500) # 2 frames per second if current_second == last_processed_second: continue last_processed_second = current_second - - frame_count += 1 + frame_count += 1 processed_frames += 1 gray = cv2.cvtColor(im, cv2.COLOR_BGR2GRAY) faces = self.face_cascade.detectMultiScale(gray, 1.3, 5) @@ -66,23 +67,22 @@ def get_emotion_percentages(self, video_path: str) -> GetEmotionPercentagesRespo img = extract_features(image) pred = predict_emotion(self.model, img) prediction_label = labels[pred.argmax()] - self.logger.info(f"Prediction for frame {frame_count}: {prediction_label}") + logger.info("Prediction for frame %d: %s", frame_count, prediction_label) predictions.append(prediction_label) except cv2.error as e: - self.logger.error(f"OpenCV error: {e}") - pass + logger.error("OpenCV error on frame %d: %s", frame_count, e) video.release() - self.logger.info(f"Total frames in video: {frame_count}") - self.logger.info(f"Frames actually processed: {processed_frames}") - self.logger.info(f"Total faces detected: {face_count}") + logger.info("Total frames in video: %d", frame_count) + logger.info("Frames actually processed: %d", processed_frames) + logger.info("Total faces detected: %d", face_count) if not predictions: - self.logger.warning("No faces detected or no predictions made.") + logger.warning("No faces detected or no predictions made.") percentages = getPercentages(predictions) - self.logger.info(f"Percentages of emotions detected: {percentages}") + logger.info("Emotion percentages: %s", percentages) return GetEmotionPercentagesResponse( Angry=percentages['Angry'], Disgusted=percentages['Disgusted'],