Made the todo's linking config to many files

Added configuration settings to be changed to network_psutil.py, packet_sniffer.py and dump_memory.py
Renamed global CONFIG var to config

Signed-off-by: Shahm Najeeb <Nirt_12023@outlook.com>

Author: DefinetlyNotAI

.idea/csv-editor.xml

@@ -12,12 +12,12 @@
 import psutil
 from prettytable import PrettyTable
 
-from logicytics import Log, Execute, Check, Get, FileManagement, Flag, DEBUG, DELETE_LOGS, CONFIG
+from logicytics import Log, Execute, Check, Get, FileManagement, Flag, DEBUG, DELETE_LOGS, config
 
 # Initialization
 log = Log({"log_level": DEBUG, "delete_log": DELETE_LOGS})
 ACTION, SUB_ACTION = None, None
-MAX_WORKERS = CONFIG.getint("Settings", "max_workers", fallback=min(32, (os.cpu_count() or 1) + 4))
+MAX_WORKERS = config.getint("Settings", "max_workers", fallback=min(32, (os.cpu_count() or 1) + 4))
 log.debug(f"MAX_WORKERS: {MAX_WORKERS}")
 
 

CODE/Logicytics.py

@@ -56,14 +56,45 @@ model_debug = false
 
 ###################################################
 
+[DumpMemory Settings]
+# If the file size generated exceeds this limit,
+# the file will be truncated with a message
+# Put 0 to disable the limit - Limit is in MiB - int
+file_size_limit = 0
+# Safety margin to check, it multiplies with the size limit
+# This makes sure that after the file is created, there is still
+# disk space left for other tasks,
+# Make sure its above 1 or else it will fail
+# Put 1 to disable the limit - Limit is in MiB - float
+file_size_safety = 1.5
+
+###################################################
+
+[NetWorkPsutil Settings]
+# Total time this will take will be `sample_count * interval`
+
+# Number of samples to take for feature `measure network bandwidth usage`
+# This is an integer, and should be 1 and above
+sample_count = 5
+# Time between samples in seconds for feature `measure network bandwidth usage`
+# This is a float, and should be above 0
+interval = 1.5
+
+###################################################
+
 [PacketSniffer Settings]
 # The interface to sniff packets on, keep it as WiFi for most cases
 # Autocorrects between WiFi and Wi-Fi
 interface = WiFi
 # The number of packets to sniff,
-packet_count = 10000
-# The time to timeout the sniffing process
+# Must be greater than or equal to 1 - int
+packet_count = 5000
+# The time to timeout the sniffing process only,
+# Must be greater than or equal to 5 - int
 timeout = 10
+# The maximum retry time for the whole process,
+# Must be greater than or equal to 10 and timeout - int
+max_retry_time = 30
 
 ###################################################
 

CODE/config.ini

@@ -5,16 +5,17 @@
 
 import psutil
 
-from logicytics import log
+from logicytics import log, config
 
 # TODO v3.4.1
 #  psutil.virtual_memory(): used, free, percent, total
 #  psutil.swap_memory(): used, free, percent, total
 
-# If the file size exceeds this limit, the file will be truncated with a message
-# Put 0 to disable the limit
-# TODO v3.4.1: Make this take from config.ini
-LIMIT_FILE_SIZE = 20  # Always in MiB
+LIMIT_FILE_SIZE = config.getint("DumpMemory Settings", "file_size_limit")  # Always in MiB
+SAFETY_MARGIN = config.getfloat("DumpMemory Settings", "file_size_safety")  # Always in MiB
+if SAFETY_MARGIN < 1:
+    log.critical("Invalid Safety Margin Inputted - Cannot proceed with dump memory")
+    exit(1)
 
 
 # Capture RAM Snapshot
@@ -41,7 +42,7 @@ def capture_ram_snapshot():
     log.info("Capturing RAM Snapshot...")
     memory = psutil.virtual_memory()
     swap = psutil.swap_memory()
-    with open("Ram_Snapshot.txt", "w") as file:
+    with open("memory_dumps/Ram_Snapshot.txt", "w") as file:
         try:
             file.write(f"Total RAM: {memory.total / (1024 ** 3):.2f} GB\n")
             file.write(f"Used RAM: {memory.used / (1024 ** 3):.2f} GB\n")
@@ -91,13 +92,13 @@ def gather_system_info():
     except Exception as e:
         log.error(f"Error gathering system information: {e}")
         sys_info = {'Error': 'Failed to gather system information'}
-    with open("SystemRam_Info.txt", "w") as file:
+    with open("memory_dumps/SystemRam_Info.txt", "w") as file:
         for key, value in sys_info.items():
             file.write(f"{key}: {value}\n")
     log.info("System Information saved to SystemRam_Info.txt")
 
 
-# Memory Dump (Windows-specific, using psutil)
+# Memory Dump
 def memory_dump():
     """
     Perform a memory dump of the current process, capturing detailed metadata for each readable memory region.
@@ -127,12 +128,12 @@ def memory_dump():
 
     try:
         process = psutil.Process(pid)
-        with open("Ram_Dump.txt", "wb") as dump_file:
+        with open("memory_dumps/Ram_Dump.txt", "wb") as dump_file:
             total_size = 0
             for mem_region in process.memory_maps(grouped=False):
                 # Check available disk space
                 if os.path.exists("Ram_Dump.txt"):
-                    required_space = LIMIT_FILE_SIZE * 1024 * 1024 * 1.5  # 2x safety margin
+                    required_space = LIMIT_FILE_SIZE * 1024 * 1024 * SAFETY_MARGIN  # 2x safety margin
                     free_space = psutil.disk_usage(".").free
                     if free_space < required_space:
                         log.error(f"Not enough disk space. Need {required_space / 1024 / 1024:.2f}MB")
@@ -206,6 +207,7 @@ def main():
     No parameters.
     No return value.
     """
+    os.makedirs("memory_dumps", exist_ok=True)
     log.info("Starting system memory collection tasks...")
     capture_ram_snapshot()
     gather_system_info()

CODE/dump_memory.py

@@ -15,7 +15,7 @@ def __config_data() -> tuple[str, str, list[str], bool, str]:
             - Version (str): System version from configuration
             - Files (list[str]): List of files specified in configuration
             - Delete old logs (bool): Flag indicating whether to delete old log files
-            - CONFIG itself
+            - config itself
 
     Raises:
         SystemExit: If the 'config.ini' file cannot be found in any of the attempted locations
@@ -30,21 +30,21 @@ def _config_path() -> str:
             print("The config.ini file is not found in the expected location.")
             exit(1)
 
-    config = configparser.ConfigParser()
+    config_local = configparser.ConfigParser()
     path = _config_path()
-    config.read(path)
+    config_local.read(path)
 
-    log_using_debug = config.getboolean("Settings", "log_using_debug")
-    delete_old_logs = config.getboolean("Settings", "delete_old_logs")
-    version = config.get("System Settings", "version")
-    files = config.get("System Settings", "files").split(", ")
+    log_using_debug = config_local.getboolean("Settings", "log_using_debug")
+    delete_old_logs = config_local.getboolean("Settings", "delete_old_logs")
+    version = config_local.get("System Settings", "version")
+    files = config_local.get("System Settings", "files").split(", ")
 
     log_using_debug = "DEBUG" if log_using_debug else "INFO"
 
-    return log_using_debug, version, files, delete_old_logs, config
+    return log_using_debug, version, files, delete_old_logs, config_local
 
 
 # Check if the script is being run directly, if not, set up the library
 if __name__ == '__main__':
     exit("This is a library, Please import rather than directly run.")
-DEBUG, VERSION, CURRENT_FILES, DELETE_LOGS, CONFIG = __config_data()
+DEBUG, VERSION, CURRENT_FILES, DELETE_LOGS, config = __config_data()

CODE/logicytics/Config.py

@@ -8,23 +8,23 @@
 from collections import Counter
 from datetime import datetime
 
-from logicytics.Config import CONFIG
+from logicytics.Config import config
 
 # Check if the script is being run directly, if not, set up the library
 if __name__ == '__main__':
     exit("This is a library, Please import rather than directly run.")
 else:
     # Save user preferences?
-    SAVE_PREFERENCES = CONFIG.getboolean("Settings", "save_preferences")
+    SAVE_PREFERENCES = config.getboolean("Settings", "save_preferences")
     # Debug mode for Sentence Transformer
-    DEBUG_MODE = CONFIG.getboolean("Flag Settings", "model_debug")  # Debug mode for Sentence Transformer
+    DEBUG_MODE = config.getboolean("Flag Settings", "model_debug")  # Debug mode for Sentence Transformer
     # File for storing user history data
     HISTORY_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'User_History.json.gz')  # User history file
     # Minimum accuracy threshold for flag suggestions
     MIN_ACCURACY_THRESHOLD = float(
-        CONFIG.get("Flag Settings", "accuracy_min"))  # Minimum accuracy threshold for flag suggestions
-    if not 0 <= MIN_ACCURACY_THRESHOLD <= 100:
-        raise ValueError("accuracy_min must be between 0 and 100")
+        config.get("Flag Settings", "accuracy_min"))  # Minimum accuracy threshold for flag suggestions
+    if not 1 <= MIN_ACCURACY_THRESHOLD <= 99:
+        raise ValueError("accuracy_min must be between 1 and 99")
 
 
 class _Match:
@@ -61,11 +61,11 @@ def __get_sim(user_input: str, all_descriptions: list[str]) -> list[float]:
             logging.getLogger("sentence_transformers").setLevel(logging.ERROR)
 
         try:
-            MODEL = SentenceTransformer(CONFIG.get("Flag Settings", "model_to_use"))
+            MODEL = SentenceTransformer(config.get("Flag Settings", "model_to_use"))
         except Exception as e:
             print(f"Error: {e}")
             print("Please check the model name in the config file.")
-            print(f"Model name {CONFIG.get('Flag Settings', 'model_to_use')} may not be valid.")
+            print(f"Model name {config.get('Flag Settings', 'model_to_use')} may not be valid.")
             exit(1)
 
         user_embedding = MODEL.encode(user_input, convert_to_tensor=True, show_progress_bar=DEBUG_MODE)

CODE/logicytics/Flag.py

@@ -2,7 +2,7 @@
 import traceback
 
 from logicytics.Checks import Check
-from logicytics.Config import DEBUG, VERSION, CURRENT_FILES, DELETE_LOGS, CONFIG
+from logicytics.Config import DEBUG, VERSION, CURRENT_FILES, DELETE_LOGS, config
 from logicytics.Execute import Execute
 from logicytics.FileManagement import FileManagement
 from logicytics.Flag import Flag

CODE/logicytics/__init__.py

@@ -4,14 +4,18 @@
 
 import psutil
 
-from logicytics import log, Execute
+from logicytics import log, Execute, config
 
 
 class NetworkInfo:
     """
     A class to gather and save various network-related information.
     """
 
+    def __init__(self):
+        self.SAMPLE_COUNT = config.getint("NetWorkPsutil Settings", "sample_count")
+        self.INTERVAL = config.getfloat("NetWorkPsutil Settings", "interval")
+
     @log.function
     async def get(self):
         """
@@ -24,7 +28,7 @@ async def get(self):
             self.__fetch_network_interface_stats()
             self.__execute_external_network_command()
             self.__fetch_network_connections_with_process_info()
-            await self.__measure_network_bandwidth_usage()
+            await self.__measure_network_bandwidth_usage(sample_count=self.SAMPLE_COUNT, interval=self.INTERVAL)
             self.__fetch_hostname_and_ip()
         except Exception as e:
             log.error(f"Error getting network info: {e}, Type: {type(e).__name__}")
@@ -133,7 +137,9 @@ async def __measure_network_bandwidth_usage(self, sample_count: int = 5, interva
             sample_count: Number of samples to take (default: 5)
             interval: Time between samples in seconds (default: 1.0)
         """
-        # TODO v3.4.1: Allow config.ini to set values
+        if sample_count < 1 or interval <= 0:
+            log.critical(
+                "Invalid values passed down from configuration for `NetworkInfo.__measure_network_bandwidth_usage()`")
         log.debug("Measuring network bandwidth usage...")
         samples = []
         for _ in range(sample_count):

CODE/network_psutil.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import os
-import warnings
 from time import time
 
 import matplotlib.pyplot as plt
@@ -10,13 +8,7 @@
 from scapy.all import sniff, conf
 from scapy.layers.inet import IP, TCP, UDP, ICMP
 
-from logicytics import log, CONFIG
-
-# Read configuration from config.ini
-CONFIG.read(os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.ini"))
-config = CONFIG['PacketSniffer Settings']
-# Ignore all warnings (Wireshark issue)
-warnings.filterwarnings("ignore")
+from logicytics import log, config
 
 
 class Sniff:
@@ -361,24 +353,24 @@ def correct_interface_name(interface_name: str) -> str:
             }
             return corrections.get(interface_name, interface_name)
 
-        interface = config['interface']
-        packet_count = int(config['packet_count'])
-        timeout = int(config['timeout'])
+        interface = config.get("PacketSniffer Settings", "interface")
+        packet_count = config.getint("PacketSniffer Settings", "packet_count")
+        timeout = config.getint("PacketSniffer Settings", "timeout")
+        max_retry_time = config.getint("PacketSniffer Settings", "max_retry_time")
 
-        if packet_count <= 0 or timeout <= 0:
+        if packet_count < 1 or timeout < 5 or max_retry_time < 10 or max_retry_time < timeout:
             try:
-                log.error(
-                    "Oops! Can't work with these values (Not your fault):\n"
-                    f"          - Packet count: {packet_count} {'❌ (must be > 0)' if packet_count <= 0 else '✅'}\n"
-                    f"          - Timeout: {timeout} {'❌ (must be > 0)' if timeout <= 0 else '✅'}"
+                log.critical(
+                    "Oops! Can't work with these values):\n"
+                    f"          - Packet count: {packet_count} {'❌ (must be > 0)' if packet_count < 1 else '✅'}\n"
+                    f"          - Timeout: {timeout} {'❌ (must be >= 5)' if timeout < 5 else '✅'}\n"
+                    f"          - Max Retry Time: {max_retry_time} {'❌ (must be >= 10 and larger than timeout)' if max_retry_time < 10 or max_retry_time < timeout else '✅'}"
                 )
             except Exception:
-                log.error("Error reading configuration: Improper values for packet count or timeout")
+                log.critical("Error reading configuration: Improper values for packet count or timeout")
             exit(1)
 
         start_time = time()
-        # TODO v3.4.1 -> Config.ini controlled value
-        max_retry_time = 30  # seconds
         for attempt in range(2):  # Try original and corrected name
             try:
                 if time() - start_time > max_retry_time:

CODE/packet_sniffer.py

@@ -23,8 +23,6 @@
 #  Use Asynchronous File Scanning,
 #  Optimize Model Loading and Caching,
 #  Improve Feature Extraction
-
-# TODO: v3.4.1
 #  also add a global variable called MAX_FILE_SIZE, if its none ignore it, else only scan files under that file size (default at 50MB)
 #  add this to config.ini -> max_workers = min(32, os.cpu_count() * 2)
 #  add UNREADABLE_EXTENSIONS as well to config.ini