199 lines
7.5 KiB
Python
199 lines
7.5 KiB
Python
# --- START OF FILE utils.py ---
|
|
|
|
# utils.py
|
|
"""
|
|
THIS SOFTWARE IS PROVIDED “AS IS” AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
|
|
|
|
Provides utility functions used throughout the application, including
|
|
safe queue management (put, clear) and coordinate formatting (decimal degrees to DMS).
|
|
Uses standardized logging prefixes. Drop counts are now managed within AppState.
|
|
"""
|
|
|
|
# Standard library imports
|
|
import queue
|
|
import logging
|
|
import math
|
|
|
|
# Removed: threading (Lock is now in AppState)
|
|
|
|
# Local application imports
|
|
# Removed: Global counters and lock
|
|
|
|
# --- Queue Management Functions ---
|
|
|
|
|
|
# --- MODIFIED FUNCTION: put_queue ---
|
|
def put_queue(queue_obj, item, queue_name="Unknown", app_instance=None):
|
|
"""
|
|
Safely puts an item into a queue using non-blocking put.
|
|
Increments specific drop counters in the AppState object if the queue is full.
|
|
Discards item if the application is shutting down.
|
|
|
|
Args:
|
|
queue_obj (queue.Queue): The queue instance.
|
|
item (any): The item to put in the queue.
|
|
queue_name (str): Identifier for the queue (for logging/counting: "sar", "mfd", "tkinter", "mouse").
|
|
app_instance (App, optional): Reference to the main app instance which holds the AppState.
|
|
Required for shutdown check and incrementing drop counts.
|
|
"""
|
|
log_prefix = "[Utils Queue Put]" # Specific prefix
|
|
|
|
# Check shutdown flag via app_instance
|
|
if (
|
|
app_instance
|
|
and hasattr(app_instance, "state")
|
|
and app_instance.state.shutting_down
|
|
):
|
|
logging.debug(
|
|
f"{log_prefix} Queue '{queue_name}': Skipping put due to shutdown."
|
|
)
|
|
return # Don't queue if shutting down
|
|
|
|
try:
|
|
# Attempt non-blocking put
|
|
queue_obj.put(item, block=False)
|
|
# DEBUG log for successful put
|
|
logging.debug(
|
|
f"{log_prefix} Item successfully put onto queue '{queue_name}' (Current approx size: {queue_obj.qsize()})."
|
|
)
|
|
except queue.Full:
|
|
# WARNING level is appropriate for dropped items due to full queue
|
|
logging.warning(
|
|
f"{log_prefix} Queue '{queue_name}' is full (maxsize={queue_obj.maxsize}). Dropping item."
|
|
)
|
|
# Increment the counter via the AppState instance
|
|
if app_instance and hasattr(app_instance, "state"):
|
|
# Use the dedicated method in AppState for safe incrementing
|
|
try:
|
|
app_instance.state.increment_dropped_count(queue_name)
|
|
except Exception as e:
|
|
logging.error(
|
|
f"{log_prefix} Failed to increment drop count for queue '{queue_name}': {e}"
|
|
)
|
|
else:
|
|
logging.error(
|
|
f"{log_prefix} Cannot increment drop count for queue '{queue_name}': App instance or state missing."
|
|
)
|
|
pass # Item is discarded
|
|
except Exception as e:
|
|
# Keep EXCEPTION for unexpected errors during put
|
|
logging.exception(
|
|
f"{log_prefix} Unexpected error putting item onto queue '{queue_name}': {e}"
|
|
)
|
|
|
|
|
|
# --- REMOVED FUNCTION: get_dropped_counts ---
|
|
# This is now handled by AppState.get_statistics()
|
|
|
|
|
|
def clear_queue(q):
|
|
"""Removes all items currently in the specified queue."""
|
|
log_prefix = "[Utils Queue Clear]" # Specific prefix
|
|
try:
|
|
q_size_before = q.qsize() # Get approximate size before clearing
|
|
# Clear the underlying deque while holding the mutex for safety
|
|
with q.mutex:
|
|
q.queue.clear()
|
|
# DEBUG log confirming the action
|
|
logging.debug(
|
|
f"{log_prefix} Cleared queue (approx size before: {q_size_before})."
|
|
)
|
|
except Exception as e:
|
|
# Keep EXCEPTION for unexpected errors during clear
|
|
logging.exception(f"{log_prefix} Error clearing queue: {e}")
|
|
|
|
|
|
# --- Coordinate Formatting ---
|
|
|
|
|
|
def decimal_to_dms(decimal_degrees, is_latitude):
|
|
"""
|
|
Converts decimal degrees to a formatted DMS (Degrees, Minutes, Seconds) string.
|
|
|
|
Args:
|
|
decimal_degrees (float): The coordinate value in decimal degrees.
|
|
is_latitude (bool): True if the coordinate is latitude, False for longitude.
|
|
|
|
Returns:
|
|
str: Formatted DMS string (e.g., "40° 51' 54.33\" N") or status string on error.
|
|
"""
|
|
log_prefix = "[Utils DMS Conv]" # Specific prefix
|
|
# DEBUG for function entry and input values
|
|
coord_type = "Latitude" if is_latitude else "Longitude"
|
|
logging.debug(f"{log_prefix} Converting {coord_type} {decimal_degrees} to DMS...")
|
|
|
|
try:
|
|
# Validate input type - Return "N/A" for invalid types
|
|
if (
|
|
not isinstance(decimal_degrees, (int, float))
|
|
or math.isnan(decimal_degrees)
|
|
or math.isinf(decimal_degrees)
|
|
):
|
|
# WARNING if input is invalid type
|
|
logging.warning(
|
|
f"{log_prefix} Invalid input type or value ({decimal_degrees}). Returning 'N/A'."
|
|
)
|
|
return "N/A"
|
|
|
|
# Determine direction and padding
|
|
if is_latitude:
|
|
direction = "N" if decimal_degrees >= 0 else "S"
|
|
degrees_padding = 2
|
|
# Validate range
|
|
if (
|
|
abs(decimal_degrees) > 90.000001
|
|
): # Allow slight tolerance for float issues
|
|
# WARNING for out-of-range input
|
|
logging.warning(
|
|
f"{log_prefix} Latitude {decimal_degrees} out of range (-90 to 90). Returning 'Invalid Lat'."
|
|
)
|
|
return "Invalid Lat"
|
|
else: # Longitude
|
|
direction = "E" if decimal_degrees >= 0 else "W"
|
|
degrees_padding = 3
|
|
# Validate range
|
|
if abs(decimal_degrees) > 180.000001: # Allow slight tolerance
|
|
# WARNING for out-of-range input
|
|
logging.warning(
|
|
f"{log_prefix} Longitude {decimal_degrees} out of range (-180 to 180). Returning 'Invalid Lon'."
|
|
)
|
|
return "Invalid Lon"
|
|
|
|
# Use absolute value for calculations
|
|
decimal_degrees = abs(decimal_degrees)
|
|
|
|
# Calculate components - DEBUG for intermediate values
|
|
degrees = math.floor(decimal_degrees)
|
|
minutes_decimal = (decimal_degrees - degrees) * 60.0
|
|
minutes = math.floor(minutes_decimal)
|
|
seconds = (minutes_decimal - minutes) * 60.0
|
|
# Handle potential floating point inaccuracies near 60 seconds
|
|
if abs(seconds - 60.0) < 1e-9:
|
|
seconds = 0.0
|
|
minutes += 1
|
|
if minutes == 60:
|
|
minutes = 0
|
|
degrees += 1
|
|
|
|
logging.debug(
|
|
f"{log_prefix} Calculated components: D={degrees}, M={minutes}, S={seconds:.4f}"
|
|
)
|
|
|
|
# Format the string - Use f-strings for clarity
|
|
# Ensure seconds formatting handles potential negative sign space if needed (though abs value used now)
|
|
dms_string = f"{degrees:0{degrees_padding}d}° {minutes:02d}' {seconds:05.2f}\" {direction}"
|
|
|
|
# DEBUG for successful formatting result
|
|
logging.debug(f"{log_prefix} Formatting successful: '{dms_string}'")
|
|
return dms_string
|
|
|
|
except Exception as e:
|
|
# Keep EXCEPTION for unexpected errors during conversion/formatting
|
|
logging.exception(
|
|
f"{log_prefix} Error converting {decimal_degrees} to DMS: {e}"
|
|
)
|
|
return "Error DMS" # Return specific error string
|
|
|
|
|
|
# --- END OF FILE utils.py ---
|