Added modules to support cli.py modules

Author: GenaNiv

src/cli.py

@@ -1,8 +1,27 @@
 import argparse
-from service.commands import EnrollSpeakerCommand, RecognizeSpeakerCommand, ListSpeakersCommand, DeleteSpeakerCommand, CommandHandler
-from bst import BinarySearchTree
-
-def main():
+import os
+
+from service.commands import (
+    EnrollSpeakerCommand,
+    RecognizeSpeakerCommand,
+    ListSpeakersCommand,
+    DeleteSpeakerCommand,
+    CommandHandler
+)
+from file_management.bst import BinarySearchTree
+from file_management.file_management import FileManagementInterface
+
+def setup_environment(base_directory):
+    # Ensure the base directory for models, audio files, and metadata exists
+    if not os.path.exists(os.path.join(base_directory, "models")):
+        os.makedirs(os.path.join(base_directory, "models"))
+    if not os.path.exists(os.path.join(base_directory, "audio_files")):
+        os.makedirs(os.path.join(base_directory, "audio_files"))
+    if not os.path.exists(os.path.join(base_directory, "metadata")):
+        os.makedirs(os.path.join(base_directory, "metadata"))
+    print(f"Environment set up at {base_directory}")
+
+def main(command_line_args=None):
     """CLI entry point."""
     # Initialize Argument Parser
     parser = argparse.ArgumentParser(description="Speaker Recognition CLI Tool")
@@ -27,6 +46,12 @@ def main():
     # Recognize Command
     recognize_parser = subparsers.add_parser('recognize', help='Recognize a speaker from an audio file')
     recognize_parser.add_argument('audio_file', type=str, help='Path to the audio file')
+    recognize_parser.add_argument('--sample_rate', type=int, default=16000, help='Sample rate of the audio file')
+    recognize_parser.add_argument('--frame_size', type=float, default=0.025, help='Frame size in seconds')
+    recognize_parser.add_argument('--frame_step', type=float, default=0.01, help='Frame step (overlap) in seconds')
+    recognize_parser.add_argument('--fft_size', type=int, default=512, help='FFT size for audio processing')
+    recognize_parser.add_argument('--num_filters', type=int, default=26, help='Number of Mel filters')
+    recognize_parser.add_argument('--num_ceps', type=int, default=13, help='Number of MFCC coefficients')
 
     # List Speakers Command
     subparsers.add_parser('list_speakers', help='List all enrolled speakers')
@@ -36,14 +61,19 @@ def main():
     delete_parser.add_argument('speaker_name', type=str, help='Name of the speaker to delete')
 
     # Parse the arguments
-    args = parser.parse_args()
+    args = parser.parse_args(command_line_args)
 
     # Initialize the command handler
     handler = CommandHandler()
 
-    # Binary Search Tree and base directory
-    bst = BinarySearchTree()  # Placeholder for actual binary search tree
-    base_directory = "models/"  # Placeholder for actual base directory
+    # Base directory setup
+    base_directory = "test_environment"  # Placeholder for the base directory
+
+    # Ensure environment setup
+    setup_environment(base_directory)
+
+    # Initialize Binary Search Tree
+    bst = BinarySearchTree()  # Placeholder for actual binary search tree implementation
 
     # Process the command based on the parsed arguments
     if args.command == 'enroll':
@@ -61,21 +91,60 @@ def main():
             n_mixtures=args.n_mixtures
         )
         handler.run(command)
+        
+        # Serialize the BST before exiting the program
+        bst.serialize_bst()
 
     elif args.command == 'recognize':
-        command = RecognizeSpeakerCommand(args.audio_file)
+        command = RecognizeSpeakerCommand(
+            bst=bst,
+            audio_file=args.audio_file,
+            base_directory=base_directory,
+            sample_rate=args.sample_rate,
+            frame_size=args.frame_size,
+            frame_step=args.frame_step,
+            fft_size=args.fft_size,
+            num_filters=args.num_filters,
+            num_ceps=args.num_ceps
+        )
         handler.run(command)
 
     elif args.command == 'list_speakers':
-        command = ListSpeakersCommand()
+        file_management = FileManagementInterface(bst=bst, base_directory=base_directory)
+        command = ListSpeakersCommand(file_management)
         handler.run(command)
 
     elif args.command == 'delete_speaker':
-        command = DeleteSpeakerCommand(args.speaker_name)
+        file_management = FileManagementInterface(bst=bst, base_directory=base_directory)
+        command = DeleteSpeakerCommand(args.speaker_name, file_management)
         handler.run(command)
 
     else:
         parser.print_help()
 
 if __name__ == "__main__":
-    main()
+    #debug_args = [
+    #    'enroll',
+    #    'maria',
+    #    '/home/gena/PROJECTS/voice-recognition-engine/audio_files/maria.wav',
+    #    '--sample_rate', '16000',
+    #    '--num_filters', '40',
+    #    '--num_ceps', '13',
+    #    '--n_fft', '512',
+    #    '--frame_size', '0.025',
+    #    '--frame_step', '0.01',
+    #    '--n_mixtures', '8'
+    #]
+    
+    debug_args = [
+        'recognize',
+        '/home/gena/PROJECTS/voice-recognition-engine/audio_files/leah_recognize.wav',
+        '--sample_rate', '16000',
+        '--frame_size', '0.025',
+        '--frame_step', '0.01',
+        '--fft_size', '512',
+        '--num_filters', '40',
+        '--num_ceps', '13',
+    ]
+
+    main(debug_args)

src/feature_extraction/audio_feature_extractor.py

@@ -1,12 +1,12 @@
 import numpy as np
 import matplotlib.pyplot as plt
 import librosa
-from src.feature_extraction.pre_emphasis import PreEmphasisFilter
-from src.feature_extraction.framing import Framing
-from src.feature_extraction.hamming_window import HammingWindow
-from src.feature_extraction.fft import FFTProcessor
-from src.feature_extraction.mel_filterbank import MelScaleFilterbank
-from src.feature_extraction.dct_processor import DCTProcessor, LogarithmCompression
+from feature_extraction.pre_emphasis import PreEmphasisFilter
+from feature_extraction.framing import Framing
+from feature_extraction.hamming_window import HammingWindow
+from feature_extraction.fft import FFTProcessor
+from feature_extraction.mel_filterbank import MelScaleFilterbank
+from feature_extraction.dct_processor import DCTProcessor, LogarithmCompression
 
 class AudioFeatureExtractor:
     def __init__(self, sample_rate=16000, frame_size=0.025, frame_step=0.01, fft_size=512, num_filters=26, num_ceps=13):

src/file_management/file_management.py

@@ -3,7 +3,7 @@
 import time
 import hashlib
 
-import src.file_management.bst as bst
+import file_management.bst as bst
 
 
 class FileManagementBase(ABC):

src/gmm/gmm_factory.py

@@ -1,4 +1,4 @@
-from src.gmm.gmm_gaussian import GMMGaussianModel 
+from gmm.gmm_gaussian import GMMGaussianModel 
 
 """
 GMM Factory Module

src/gmm/gmm_gaussian.py

@@ -2,7 +2,7 @@
 from sklearn.mixture import GaussianMixture
 import pickle
 
-from src.gmm.gmm_base import GMMModelBase
+from gmm.gmm_base import GMMModelBase
 
 class GMMGaussianModel(GMMModelBase):
     """

src/main.py

@@ -1,77 +1,33 @@
-import time 
-import pathlib
-import os
-import re
-import file_management.bst as bst
-
-
-# The program should generate 5 files with name that have file's uniq ID and time stamps
-
-
-def current_time():
-    t = time.localtime()
-    current_time = time.strftime("%Y_%m_%d_%H_%M_%s", t)
-    return current_time
-
-def read_files(target_directory):
-    """TBD"""
-    files_data = {}
-    for filename in os.listdir(target_directory):
-        file_path = os.path.join(target_directory, filename)
-        
-        if os.path.isfile(file_path) and filename.endswith(".txt"):
-            with open(file_path, "r") as f:
-                lines = f.readlines()
-                # Extract file_id and timestamp from the lines
-                file_id = int(lines[0].split(":")[1].strip())
-                timestamp = lines[1].split(":")[1].strip()
-                files_data[file_id] = timestamp
-    return files_data
-    
-                
-def generate_files(number_of_files, target_directory):
-    """TBD"""
-    file_info = {}
-
-    for i in range(number_of_files):
-        file_info[i + 1] = current_time()
-        
-        file_name = f"{i + 1}_{file_info[i + 1]}.txt"
-        full_path = os.path.join(target_directory, file_name)
-        with open(full_path, "w") as f:
-            f.write("File ID: " + str(i + 1) )
-            f.write("\n")
-            f.write("File last modified: " + file_info[i + 1])
-        time.sleep(1)  # Wait for 1 second between files
+import sys
+from cli import main as cli_main  # Import the CLI entry point
+from web_gui import app as web_app  # Import the Flask app for the Web GUI
+import logging
+import json
+
+def load_config():
+    """Load configuration settings (e.g., port, logging settings)."""
+    with open('config.json', 'r') as config_file:
+        config = json.load(config_file)
+    return config
+
+def setup_logging():
+    """Setup logging configuration."""
+    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+    logging.info("Logging is set up.")
+
+def main():
+    """Main entry point for the system."""
+    # Load config and set up logging
+    config = load_config()
+    setup_logging()
+
+    # Determine execution mode (CLI or Web GUI)
+    if len(sys.argv) > 1 and sys.argv[1] == 'cli':
+        logging.info("Starting CLI interface...")
+        cli_main()  # Run the CLI
+    else:
+        logging.info("Starting Web GUI...")
+        web_app.run(host='0.0.0.0', port=config.get("port", 5000), debug=True)
 
 if __name__ == "__main__":
-
-    
-    #target_directory = "/home/gena/PROJECTS/ServerManagement/config_files"
-    #number_of_files = 100
-    #generate_files(number_of_files, target_directory)
-    
-    # Create binary search tree and insert data
-
-    tree = bst.BinarySearchTree()
-    config_files = read_files("/home/gena/PROJECTS/ServerManagement/config_files")
-    print("Measure how long it take to insert one file to the BST")
-    for file_id, file_timestamp in config_files.items():
-        start_time = time.time()
-        tree.insert(file_id, file_timestamp)
-        end_time = time.time()
-        print(f"Insertion of file ID {file_id} took {end_time - start_time:.6f} seconds")
-
-    print("Measure how long it take to delete a node form the BST")
-    start_time = time.time()
-    file_id = 57
-    tree.delete_node(file_id)
-    end_time = time.time()
-    print(f"Deletion of file ID {file_id} took {end_time - start_time:.6f} seconds")
-
-            
-    
-    
-
-        
-    
+    main()

src/service/commands.py

@@ -0,0 +1,103 @@
+# Updated commands.py based on the provided details
+
+# commands.py
+import os
+from service.speaker_enrollment import SpeakerEnrollment
+from service.speaker_recognition import SpeakerRecognition
+from file_management.file_management import FileManagementInterface
+
+# Base Command class
+class Command:
+    """Base class for all commands."""
+    def execute(self):
+        raise NotImplementedError("Subclasses must implement the execute method.")
+
+# Command for enrolling a speaker
+class EnrollSpeakerCommand(Command):
+    def __init__(self, speaker_name, audio_file, bst, base_directory, 
+                 sample_rate, num_filters, num_ceps, n_fft, 
+                 frame_size, frame_step, n_mixtures):
+        self.speaker_name = speaker_name
+        self.audio_file = audio_file
+        self.bst = bst
+        self.base_directory = base_directory
+        self.sample_rate = sample_rate
+        self.num_filters = num_filters
+        self.num_ceps = num_ceps
+        self.n_fft = n_fft
+        self.frame_size = frame_size
+        self.frame_step = frame_step
+        self.n_mixtures = n_mixtures
+
+    def execute(self):
+        """Execute the enroll command by enrolling a new speaker."""
+        # Initialize SpeakerEnrollment with the provided parameters
+        speaker_enrollment = SpeakerEnrollment(
+            bst=self.bst, 
+            base_directory=self.base_directory, 
+            sample_rate=self.sample_rate, 
+            num_filters=self.num_filters, 
+            num_ceps=self.num_ceps, 
+            n_fft=self.n_fft, 
+            frame_size=self.frame_size, 
+            frame_step=self.frame_step, 
+            n_mixtures=self.n_mixtures
+        )
+
+        # Enroll the speaker using the provided parameters
+        success = speaker_enrollment.enroll_speaker(self.speaker_name, self.audio_file)
+        if success:
+            print(f"Speaker {self.speaker_name} enrolled successfully.")
+        else:
+            print(f"Failed to enroll speaker {self.speaker_name}.")
+
+# Command for recognizing a speaker
+class RecognizeSpeakerCommand(Command):
+    def __init__(self, bst, audio_file, base_directory, sample_rate, frame_size, frame_step, fft_size, num_filters, num_ceps):
+        self.audio_file = audio_file
+        self.recognizer = SpeakerRecognition(
+            bst=bst,
+            base_directory=base_directory,
+            sample_rate=sample_rate,
+            frame_size=frame_size,
+            frame_step=frame_step,
+            fft_size=fft_size,
+            num_filters=num_filters,
+            num_ceps=num_ceps
+        )
+
+    def execute(self):
+        """Execute the recognize command to identify the speaker."""
+        recognized_speaker = self.recognizer.recognize_speaker(self.audio_file)
+        print(f"Recognized Speaker: {recognized_speaker}")
+
+# Command for listing all enrolled speakers
+class ListSpeakersCommand(Command):
+    def __init__(self, file_management):
+        self.file_management = file_management
+
+    def execute(self):
+        """Execute the list speakers command to display all speakers."""
+        speakers = self.file_management.list_all_files()
+        print("Enrolled Speakers:")
+        for speaker in speakers:
+            print(f"- {speaker['file_id']}")
+
+# Command for deleting a speaker
+class DeleteSpeakerCommand(Command):
+    def __init__(self, speaker_name, file_management):
+        self.speaker_name = speaker_name
+        self.file_management = file_management
+
+    def execute(self):
+        """Execute the delete command to remove a speaker."""
+        self.file_management.delete_file(self.speaker_name)
+        print(f"Speaker {self.speaker_name} deleted successfully.")
+
+# CommandHandler to execute the commands
+class CommandHandler:
+    """Handles the execution of commands."""
+    
+    def run(self, command):
+        """Run the given command."""
+        command.execute()