From f9d30b0c8bacd0007f7e3eef39c7b78a086462ef Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 4 Mar 2025 22:00:02 -0500 Subject: [PATCH] Linting --- app/internal/NoiseGatev2.py | 28 ++++++++++++++-------------- app/internal/bot_manager.py | 22 +++++++++++----------- app/internal/logger.py | 20 ++++++++++---------- app/main.py | 7 +------ app/routers/bot.py | 5 ----- app/routers/op25_controller.py | 4 ++-- 6 files changed, 38 insertions(+), 48 deletions(-) diff --git a/app/internal/NoiseGatev2.py b/app/internal/NoiseGatev2.py index 7cf80d2..71e3929 100644 --- a/app/internal/NoiseGatev2.py +++ b/app/internal/NoiseGatev2.py @@ -30,15 +30,15 @@ class AudioStream: if _input: self.paInstance_kwargs['input_device_index'] = _input_device_index else: - LOGGER.warning(f"[AudioStream.__init__]:\tInput was not enabled." - f" Reinitialize with '_input=True'") + LOGGER.warning("[AudioStream.__init__]:\tInput was not enabled." + " Reinitialize with '_input=True'") if _output_device_index: if _output: self.paInstance_kwargs['output_device_index'] = _output_device_index else: - LOGGER.warning(f"[AudioStream.__init__]:\tOutput was not enabled." - f" Reinitialize with '_output=True'") + LOGGER.warning("[AudioStream.__init__]:\tOutput was not enabled." + " Reinitialize with '_output=True'") if _init_on_startup: # Init PyAudio instance @@ -59,15 +59,15 @@ class AudioStream: if self.paInstance_kwargs['input']: self.paInstance_kwargs['input_device_index'] = _new_input_device_index else: - LOGGER.warning(f"[AudioStream.init_stream]:\tInput was not enabled when initialized." - f" Reinitialize with '_input=True'") + LOGGER.warning("[AudioStream.init_stream]:\tInput was not enabled when initialized." + " Reinitialize with '_input=True'") if _new_output_device_index: if self.paInstance_kwargs['output']: self.paInstance_kwargs['output_device_index'] = _new_output_device_index else: - LOGGER.warning(f"[AudioStream.init_stream]:\tOutput was not enabled when initialized." - f" Reinitialize with '_output=True'") + LOGGER.warning("[AudioStream.init_stream]:\tOutput was not enabled when initialized." + " Reinitialize with '_output=True'") self.close_if_open() @@ -80,7 +80,7 @@ class AudioStream: if self.stream.is_active(): self.stream.stop_stream() self.stream.close() - LOGGER.debug(f"[ReopenStream.close_if_open]:\t Stream was open; It was closed.") + LOGGER.debug("[ReopenStream.close_if_open]:\t Stream was open; It was closed.") def list_devices(self, _display_input_devices: bool = True, _display_output_devices: bool = True): LOGGER.info('Getting a list of the devices connected') @@ -126,7 +126,7 @@ class NoiseGate(AudioStream): def run(self) -> None: global voice_connection # Start the audio stream - LOGGER.debug(f"Starting stream") + LOGGER.debug("Starting stream") self.stream.start_stream() # Start the stream to discord self.core() @@ -139,15 +139,15 @@ class NoiseGate(AudioStream): time.sleep(.2) if not voice_connection.is_playing(): - LOGGER.debug(f"Playing stream to discord") + LOGGER.debug("Playing stream to discord") voice_connection.play(self.NGStream, after=self.core) async def close(self): - LOGGER.debug(f"Closing") + LOGGER.debug("Closing") await voice_connection.disconnect() if self.stream.is_active: self.stream.stop_stream() - LOGGER.debug(f"Stopping stream") + LOGGER.debug("Stopping stream") # noinspection PyUnresolvedReferences @@ -155,7 +155,7 @@ class NoiseGateStream(discord.AudioSource): def __init__(self, _stream): super(NoiseGateStream, self).__init__() self.stream = _stream # The actual audio stream object - self.NG_fadeout = 240/20 # Fadeout value used to hold the noisegate after de-triggering + self.NG_fadeout = 240 / 20 # Fadeout value used to hold the noisegate after de-triggering self.NG_fadeout_count = 0 # A count set when the noisegate is triggered and was de-triggered self.process_set_count = 0 # Counts how many processes have been made diff --git a/app/internal/bot_manager.py b/app/internal/bot_manager.py index 3ff2531..60fbe2e 100644 --- a/app/internal/bot_manager.py +++ b/app/internal/bot_manager.py @@ -1,7 +1,7 @@ import asyncio import platform import os -from discord import VoiceClient, VoiceChannel, opus, Activity, ActivityType, Intents +from discord import VoiceClient, VoiceChannel, opus, Activity, ActivityType, Intents from discord.ext import commands from typing import Optional, Dict from internal.NoiseGatev2 import NoiseGate @@ -84,16 +84,16 @@ class DiscordBotManager: if guild_id in self.voice_clients: raise RuntimeError("Already connected to this guild's voice channel.") - + try: voice_client = await channel.connect(timeout=60.0, reconnect=True) - LOGGER.debug(f"Voice Connected.") + LOGGER.debug("Voice Connected.") streamHandler = NoiseGate( _input_device_index=device_id, _voice_connection=voice_client, _noise_gate_threshold=ng_threshold) streamHandler.run() - LOGGER.debug(f"Stream is running.") + LOGGER.debug("Stream is running.") self.voice_clients[guild_id] = voice_client LOGGER.info(f"Joined guild {guild_id} voice channel {channel_id} and stream is running.") except Exception as e: @@ -117,18 +117,18 @@ class DiscordBotManager: script_dir = os.path.dirname(os.path.abspath(__file__)) LOGGER.debug("Processor: ", processor) if os.name == 'nt': - if processor == "AMD64": + if processor == "AMD64": opus.load_opus(os.path.join(script_dir, './opus/libopus_amd64.dll')) - LOGGER.info(f"Loaded OPUS library for AMD64") + LOGGER.info("Loaded OPUS library for AMD64") return "AMD64" else: - if processor == "aarch64": + if processor == "aarch64": opus.load_opus(os.path.join(script_dir, './opus/libopus_aarcch64.so')) - LOGGER.info(f"Loaded OPUS library for aarch64") + LOGGER.info("Loaded OPUS library for aarch64") return "aarch64" - elif processor == "armv7l": + elif processor == "armv7l": opus.load_opus(os.path.join(script_dir, './opus/libopus_armv7l.so')) - LOGGER.info(f"Loaded OPUS library for armv7l") + LOGGER.info("Loaded OPUS library for armv7l") return "armv7l" async def set_presence(self, presence: str): @@ -136,4 +136,4 @@ class DiscordBotManager: try: await self.bot.change_presence(activity=Activity(type=ActivityType.listening, name=presence)) except Exception as pe: - LOGGER.error(f"Unable to set presence: '{pe}'") \ No newline at end of file + LOGGER.error(f"Unable to set presence: '{pe}'") diff --git a/app/internal/logger.py b/app/internal/logger.py index 67ffc98..dd44862 100644 --- a/app/internal/logger.py +++ b/app/internal/logger.py @@ -4,52 +4,52 @@ from logging.handlers import RotatingFileHandler def create_logger(name, level=logging.DEBUG, max_bytes=10485760, backup_count=2): """ Creates a logger with a console and rotating file handlers for both debug and info log levels. - + Args: name (str): The name for the logger. level (int): The logging level for the logger. Defaults to logging.DEBUG. max_bytes (int): Maximum size of the log file in bytes before it gets rotated. Defaults to 10 MB. backup_count (int): Number of backup files to keep. Defaults to 2. - + Returns: logging.Logger: Configured logger. """ # Set the log file paths debug_log_file = "./client.debug.log" info_log_file = "./client.log" - + # Create a logger logger = logging.getLogger(name) logger.setLevel(level) - + # Check if the logger already has handlers to avoid duplicate logs if not logger.hasHandlers(): # Create console handler console_handler = logging.StreamHandler() console_handler.setLevel(level) - + # Create rotating file handler for debug level debug_file_handler = RotatingFileHandler(debug_log_file, maxBytes=max_bytes, backupCount=backup_count) debug_file_handler.setLevel(logging.DEBUG) - + # Create rotating file handler for info level info_file_handler = RotatingFileHandler(info_log_file, maxBytes=max_bytes, backupCount=backup_count) info_file_handler.setLevel(logging.INFO) - + # Create formatter and add it to the handlers formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') console_handler.setFormatter(formatter) debug_file_handler.setFormatter(formatter) info_file_handler.setFormatter(formatter) - + # Add the handlers to the logger logger.addHandler(console_handler) logger.addHandler(debug_file_handler) logger.addHandler(info_file_handler) - + return logger # Example usage: # logger = create_logger('my_logger') # logger.debug('This is a debug message') -# logger.info('This is an info message') \ No newline at end of file +# logger.info('This is an info message') diff --git a/app/main.py b/app/main.py index 3d350a9..99a965e 100644 --- a/app/main.py +++ b/app/main.py @@ -1,9 +1,4 @@ -import asyncio -import discord -from discord.ext import commands -from fastapi import FastAPI, HTTPException -from pydantic import BaseModel -from typing import Optional, Dict +from fastapi import FastAPI import routers.op25_controller as op25_controller import routers.pulse as pulse import routers.bot as bot diff --git a/app/routers/bot.py b/app/routers/bot.py index 9c9e4bf..f26ed91 100644 --- a/app/routers/bot.py +++ b/app/routers/bot.py @@ -1,9 +1,4 @@ -import asyncio -import discord -from discord.ext import commands from fastapi import APIRouter, HTTPException -from pydantic import BaseModel -from typing import Optional, Dict from models import BotConfig, VoiceChannelRequest from internal.bot_manager import DiscordBotManager from internal.logger import create_logger diff --git a/app/routers/op25_controller.py b/app/routers/op25_controller.py index 10ea6f2..3f4b620 100644 --- a/app/routers/op25_controller.py +++ b/app/routers/op25_controller.py @@ -1,12 +1,12 @@ from fastapi import HTTPException, APIRouter -from pydantic import BaseModel import subprocess import os import signal import json import csv -from models import * +from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, AudioConfig, TerminalConfig, TalkgroupTag from internal.logger import create_logger +from typing import List router = APIRouter() LOGGER = create_logger(__name__)