dandri/meshcore-bot
1
0
Fork 0
mirror of https://github.com/agessaman/meshcore-bot.git synced 2026-08-27 21:10:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add alert command configuration to config.ini.example and README. Update install-service.sh to support macOS and Linux service installations with improved user prompts and error handling. Enhance installation script with detailed steps for setting up the bot as a system service, including virtual environment creation and dependency installation.

Browse Source
This commit is contained in:
agessaman
2025-11-27 21:07:08 -08:00
parent 6de424f894
commit 7982cdce4c
5 changed files with 1408 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+15 -2
View File
@@ -99,7 +99,7 @@ startup_advert = flood # Send advert on startup
# Format: keyword = response_template
# Variables: {sender}, {connection_info}, {snr}, {timestamp}, {path}
test = "Message received from {sender} | {connection_info}"
help = "Bot Help: test, ping, help, hello, cmd, wx, aqi, sun, moon, solar, hfcond, satpass, dice, roll, joke, dadjoke, sports, channels, path, prefix, repeater, stats"
help = "Bot Help: test, ping, help, hello, cmd, wx, aqi, sun, moon, solar, hfcond, satpass, dice, roll, joke, dadjoke, sports, channels, path, prefix, repeater, stats, alert"
```
### Channels
@@ -117,6 +117,16 @@ n2yo_api_key = # Satellite pass data
airnow_api_key = # Air quality data
```
### Alert Command
```ini
[Alert_Command]
alert_enabled = true # Enable/disable alert command
max_incident_age_hours = 24 # Maximum age for incidents (hours)
max_distance_km = 20.0 # Maximum distance for proximity queries (km)
agency.city.<city_name> = <agency_ids> # City-specific agency IDs (e.g., agency.city.seattle = 17D20,17M15)
agency.county.<county_name> = <agency_ids> # County-specific agency IDs (aggregates all city agencies)
```
### Logging
```ini
[Logging]
@@ -159,6 +169,9 @@ The bot responds to these commands:
- `satpass <NORAD> visual` - Visual passes only (must be visually observable)
- `satpass <shortcut>` - Use shortcuts like `iss`, `hst`, `hubble`, `goes18`, `tiangong`
**Emergency Commands:**
- `alert <city|zipcode|street city|lat,lon|county> [all]` - Get active emergency incidents (e.g., `alert seattle`, `alert 98101`, `alert seattle all`)
**Gaming Commands:**
- `dice` - Roll dice (d6 by default, or specify like `dice d20`, `dice 2d6`)
- `roll` - Roll random number (1-100 by default, or specify like `roll 50`)
@@ -214,7 +227,7 @@ Example:
[Keywords]
test = "Message received from {sender} | {connection_info}"
ping = "Pong!"
help = "Bot Help: test, ping, help, hello, cmd, wx, gwx, aqi, sun, moon, solar, solarforecast, hfcond, satpass, dice, roll, joke, dadjoke, sports, channels, path, prefix, repeater, stats, multitest, webviewer"
help = "Bot Help: test, ping, help, hello, cmd, wx, gwx, aqi, sun, moon, solar, solarforecast, hfcond, satpass, dice, roll, joke, dadjoke, sports, channels, path, prefix, repeater, stats, multitest, alert, webviewer"
```
## Hardware Setup
com.meshcore.bot.plist
+50
View File
@@ -0,0 +1,50 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key>
<string>com.meshcore.bot</string>
<key>ProgramArguments</key>
<array>
<string>/usr/local/meshcore-bot/venv/bin/python</string>
<string>/usr/local/meshcore-bot/meshcore_bot.py</string>
</array>
<key>WorkingDirectory</key>
<string>/usr/local/meshcore-bot</string>
<key>RunAtLoad</key>
<true/>
<key>KeepAlive</key>
<dict>
<key>SuccessfulExit</key>
<false/>
</dict>
<key>StandardOutPath</key>
<string>/usr/local/var/log/meshcore-bot/meshcore-bot.log</string>
<key>StandardErrorPath</key>
<string>/usr/local/var/log/meshcore-bot/meshcore-bot.error.log</string>
<key>EnvironmentVariables</key>
<dict>
<key>PYTHONPATH</key>
<string>/usr/local/meshcore-bot</string>
<key>PYTHONUNBUFFERED</key>
<string>1</string>
</dict>
<key>ThrottleInterval</key>
<integer>10</integer>
<key>ProcessType</key>
<string>Background</string>
<key>Nice</key>
<integer>1</integer>
</dict>
</plist>
config.ini.example
+22
View File
@@ -599,6 +599,28 @@ enable_p_shortcut = false
# false: Hacker command is disabled
hacker_enabled = false
[Alert_Command]
# Enable or disable the alert command
# true: Alert command is available
# false: Alert command is disabled
alert_enabled = false
# PulsePoint agency IDs by county/region
# Format: agency.<county_name> = comma-separated agency IDs
# Get agency IDs from: https://web.pulsepoint.org/ or PulsePoint searchagencies API
# You can use any naming convention that makes sense for your region
# Examples: agency.county1, agency.city1, agency.region1, etc.
# Example: County agencies
# agency.county1 =
# Example: City agencies (can have multiple entries for the same region)
# agency.city1 =
# agency.city1_alias =
# Example: Combined region (all counties/agencies)
# agency.region_all =
[Web_Viewer]
# Enable or disable the web data viewer
# true: web viewer is available for viewing bot data
install-service.sh
+392 -76
View File
@@ -1,6 +1,20 @@
#!/bin/bash
# MeshCore Bot Service Installation Script
# This script installs the MeshCore Bot as a systemd service
# This script installs the MeshCore Bot as a system service
# Supports both Linux (systemd) and macOS (launchd)
#
# This script will:
# 1. Create a dedicated system user for the bot (Linux only)
# 2. Copy bot files to installation directory
# 3. Set up proper file permissions
# 4. Install and enable the service (systemd or launchd)
# 5. Create a Python virtual environment with dependencies
#
# Prerequisites:
# - Linux system with systemd OR macOS
# - Python 3.7+ installed
# - sudo access (script will prompt if needed)
# - Run from the meshcore-bot directory
set -e
@@ -9,119 +23,421 @@ RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
CYAN='\033[0;36m'
NC='\033[0m' # No Color
# Configuration
SERVICE_NAME="meshcore-bot"
SERVICE_USER="meshcore"
SERVICE_GROUP="meshcore"
INSTALL_DIR="/opt/meshcore-bot"
LOG_DIR="/var/log/meshcore-bot"
SERVICE_FILE="meshcore-bot.service"
# Detect operating system
OS="$(uname -s)"
IS_MACOS=false
IS_LINUX=false
echo -e "${BLUE}MeshCore Bot Service Installer${NC}"
echo "=================================="
# Check if running as root
if [[ $EUID -ne 0 ]]; then
echo -e "${RED}This script must be run as root (use sudo)${NC}"
exit 1
fi
# Check if systemd is available
if ! command -v systemctl &> /dev/null; then
echo -e "${RED}systemd is not available on this system${NC}"
if [[ "$OS" == "Darwin" ]]; then
IS_MACOS=true
SERVICE_TYPE="launchd"
elif [[ "$OS" == "Linux" ]]; then
IS_LINUX=true
SERVICE_TYPE="systemd"
else
echo "Error: Unsupported operating system: $OS"
echo "This script supports Linux (systemd) and macOS (launchd)"
exit 1
fi
echo -e "${YELLOW}Step 1: Creating service user and group${NC}"
# Create service user and group
if ! id "$SERVICE_USER" &>/dev/null; then
useradd --system --no-create-home --shell /bin/false "$SERVICE_USER"
echo -e "${GREEN}Created user: $SERVICE_USER${NC}"
# Configuration - OS-specific paths
SERVICE_NAME="meshcore-bot"
PLIST_NAME="com.meshcore.bot"
if [[ "$IS_MACOS" == true ]]; then
SERVICE_USER="$(whoami)" # macOS: use current user or _meshcore
SERVICE_GROUP="staff"
INSTALL_DIR="/usr/local/meshcore-bot"
LOG_DIR="/usr/local/var/log/meshcore-bot"
SERVICE_FILE="com.meshcore.bot.plist"
LAUNCHD_DIR="/Library/LaunchDaemons"
else
echo -e "${YELLOW}User $SERVICE_USER already exists${NC}"
SERVICE_USER="meshcore"
SERVICE_GROUP="meshcore"
INSTALL_DIR="/opt/meshcore-bot"
LOG_DIR="/var/log/meshcore-bot"
SERVICE_FILE="meshcore-bot.service"
SYSTEMD_DIR="/etc/systemd/system"
fi
echo -e "${YELLOW}Step 2: Creating directories${NC}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# Function to print section headers
print_section() {
echo ""
echo -e "${CYAN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo -e "${BLUE}$1${NC}"
echo -e "${CYAN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
}
# Function to print info messages
print_info() {
echo -e "${CYAN}ℹ${NC} $1"
}
# Function to print success messages
print_success() {
echo -e "${GREEN}✓${NC} $1"
}
# Function to print warning messages
print_warning() {
echo -e "${YELLOW}⚠${NC} $1"
}
# Function to print error messages
print_error() {
echo -e "${RED}✗${NC} $1"
}
print_section "MeshCore Bot Service Installer"
echo ""
if [[ "$IS_MACOS" == true ]]; then
print_info "Detected macOS - will install as launchd service"
print_info "The bot will start automatically on boot using launchd"
else
print_info "Detected Linux - will install as systemd service"
print_info "The bot will run as a dedicated user and start automatically on boot"
fi
echo ""
# Check if script has execute permissions
if [ ! -x "$0" ]; then
print_warning "Script does not have execute permissions. Attempting to set them..."
chmod +x "$0" 2>/dev/null || {
print_error "Could not set execute permissions. Please run: chmod +x install-service.sh"
exit 1
}
print_success "Execute permissions set"
fi
# Capture original user before sudo (for macOS)
ORIGINAL_USER="${SUDO_USER:-$USER}"
# Check if running as root, if not re-execute with sudo
if [[ $EUID -ne 0 ]]; then
print_warning "This script requires root privileges to install system services"
print_info "Re-executing with sudo..."
echo ""
exec sudo "$0" "$@"
fi
# Verify we're in the right directory
if [ ! -f "meshcore_bot.py" ]; then
print_error "This script must be run from the meshcore-bot directory"
print_error "Expected file not found: meshcore_bot.py"
print_info "Please cd to the meshcore-bot directory and run this script again"
exit 1
fi
# Check for service file
if [ ! -f "$SERVICE_FILE" ]; then
print_error "Service file not found: $SERVICE_FILE"
if [[ "$IS_MACOS" == true ]]; then
print_info "Expected: com.meshcore.bot.plist"
else
print_info "Expected: meshcore-bot.service"
fi
exit 1
fi
# OS-specific service manager checks
if [[ "$IS_MACOS" == true ]]; then
if ! command -v launchctl &> /dev/null; then
print_error "launchctl is not available on this system"
print_error "This script requires macOS with launchd"
exit 1
fi
else
if ! command -v systemctl &> /dev/null; then
print_error "systemd is not available on this system"
print_error "This script requires a Linux system with systemd"
exit 1
fi
fi
# Check if Python 3 is available
if ! command -v python3 &> /dev/null; then
print_error "Python 3 is not installed or not in PATH"
print_error "Please install Python 3.7 or higher before running this script"
exit 1
fi
print_section "Step 1: Setting Up Service User"
if [[ "$IS_MACOS" == true ]]; then
# Use original user if available, otherwise root
if [[ -n "$ORIGINAL_USER" && "$ORIGINAL_USER" != "root" ]]; then
SERVICE_USER="$ORIGINAL_USER"
SERVICE_GROUP="$(id -gn "$ORIGINAL_USER" 2>/dev/null || echo "staff")"
else
SERVICE_USER="root"
SERVICE_GROUP="wheel"
fi
print_info "macOS: Service will run as user '$SERVICE_USER'"
print_info "On macOS, launchd services run as the specified user"
print_success "Using user: $SERVICE_USER"
else
print_info "Creating a dedicated system user '$SERVICE_USER' for security"
print_info "This user will run the bot service with minimal privileges"
# Create service user and group
if ! id "$SERVICE_USER" &>/dev/null; then
useradd --system --no-create-home --shell /bin/false "$SERVICE_USER"
print_success "Created system user: $SERVICE_USER"
else
print_warning "User $SERVICE_USER already exists (skipping creation)"
fi
fi
print_section "Step 2: Creating Installation Directories"
print_info "Creating directory structure for bot installation"
# Create installation directory
if [ -d "$INSTALL_DIR" ]; then
print_warning "Installation directory $INSTALL_DIR already exists"
print_info "Existing files will be backed up and replaced"
BACKUP_DIR="${INSTALL_DIR}.backup.$(date +%Y%m%d_%H%M%S)"
mv "$INSTALL_DIR" "$BACKUP_DIR" 2>/dev/null || {
print_error "Could not backup existing installation. Please remove $INSTALL_DIR manually"
exit 1
}
print_success "Backed up existing installation to $BACKUP_DIR"
fi
mkdir -p "$INSTALL_DIR"
echo -e "${GREEN}Created directory: $INSTALL_DIR${NC}"
print_success "Created installation directory: $INSTALL_DIR"
# Create log directory
mkdir -p "$LOG_DIR"
echo -e "${GREEN}Created directory: $LOG_DIR${NC}"
print_success "Created log directory: $LOG_DIR"
echo -e "${YELLOW}Step 3: Copying bot files${NC}"
print_section "Step 3: Copying Bot Files"
print_info "Copying all bot files to $INSTALL_DIR"
print_info "This includes Python scripts, modules, configuration templates, and documentation"
# Copy bot files to installation directory
cp -r . "$INSTALL_DIR/"
echo -e "${GREEN}Copied bot files to $INSTALL_DIR${NC}"
cp -r . "$INSTALL_DIR/" 2>/dev/null || {
print_error "Failed to copy files. Check permissions and disk space"
exit 1
}
print_success "Copied bot files to $INSTALL_DIR"
echo -e "${YELLOW}Step 4: Setting permissions${NC}"
print_section "Step 4: Setting File Permissions"
print_info "Configuring file ownership and permissions for security"
print_info "The service user will own all files, with appropriate read/write permissions"
# Set ownership
chown -R "$SERVICE_USER:$SERVICE_GROUP" "$INSTALL_DIR"
chown -R "$SERVICE_USER:$SERVICE_GROUP" "$LOG_DIR"
print_success "Set ownership to $SERVICE_USER:$SERVICE_GROUP"
# Set permissions
chmod 755 "$INSTALL_DIR"
chmod 644 "$INSTALL_DIR"/*.py
chmod 644 "$INSTALL_DIR"/*.ini
chmod 644 "$INSTALL_DIR"/*.txt
chmod 755 "$INSTALL_DIR"/modules
chmod 644 "$INSTALL_DIR"/modules/*.py
chmod 755 "$INSTALL_DIR"/modules/commands
chmod 644 "$INSTALL_DIR"/modules/commands/*.py
find "$INSTALL_DIR" -type f -name "*.py" -exec chmod 644 {} \; 2>/dev/null || true
find "$INSTALL_DIR" -type f -name "*.ini" -exec chmod 644 {} \; 2>/dev/null || true
find "$INSTALL_DIR" -type f -name "*.txt" -exec chmod 644 {} \; 2>/dev/null || true
find "$INSTALL_DIR" -type f -name "*.json" -exec chmod 644 {} \; 2>/dev/null || true
find "$INSTALL_DIR" -type d -exec chmod 755 {} \; 2>/dev/null || true
# Make main script executable
chmod 755 "$INSTALL_DIR/meshcore_bot.py"
print_success "Configured file permissions"
echo -e "${GREEN}Set permissions for bot files${NC}"
print_section "Step 5: Installing Service"
if [[ "$IS_MACOS" == true ]]; then
print_info "Installing launchd plist file to enable automatic startup"
print_info "The service will be configured to start on boot and restart on failure"
# Create LaunchDaemons directory if it doesn't exist
mkdir -p "$LAUNCHD_DIR"
# Update plist with actual installation paths and copy to LaunchDaemons
print_info "Updating plist file with installation paths"
# Use a more portable approach for path substitution
if command -v python3 &> /dev/null; then
python3 -c "
import sys
import re
with open('$SERVICE_FILE', 'r') as f:
content = f.read()
content = content.replace('/usr/local/meshcore-bot', '$INSTALL_DIR')
content = content.replace('/usr/local/var/log/meshcore-bot', '$LOG_DIR')
with open('$LAUNCHD_DIR/$SERVICE_FILE', 'w') as f:
f.write(content)
"
else
# Fallback to sed (works on both macOS and Linux)
sed "s|/usr/local/meshcore-bot|$INSTALL_DIR|g; s|/usr/local/var/log/meshcore-bot|$LOG_DIR|g" "$SERVICE_FILE" > "$LAUNCHD_DIR/$SERVICE_FILE"
fi
print_success "Copied and configured plist file to $LAUNCHD_DIR/"
# Set ownership
chown root:wheel "$LAUNCHD_DIR/$SERVICE_FILE"
chmod 644 "$LAUNCHD_DIR/$SERVICE_FILE"
print_success "Set plist permissions"
print_section "Step 6: Loading Service"
print_info "Loading service into launchd"
# Unload if already loaded
launchctl list "$PLIST_NAME" &>/dev/null && launchctl unload "$LAUNCHD_DIR/$SERVICE_FILE" 2>/dev/null || true
# Load the service
launchctl load "$LAUNCHD_DIR/$SERVICE_FILE" 2>/dev/null || {
print_error "Failed to load service. Check plist syntax and permissions."
exit 1
}
print_success "Service '$PLIST_NAME' loaded into launchd"
print_info "Note: The service is loaded but not started yet. You'll start it after configuration."
else
print_info "Installing systemd service file to enable automatic startup"
print_info "The service will be configured to start on boot and restart on failure"
# Copy service file to systemd directory
cp "$SERVICE_FILE" "$SYSTEMD_DIR/"
print_success "Copied service file to $SYSTEMD_DIR/"
# Reload systemd
print_info "Reloading systemd to recognize the new service"
systemctl daemon-reload
print_success "Systemd configuration reloaded"
print_section "Step 6: Enabling Service"
print_info "Enabling service to start automatically on system boot"
# Enable service to start on boot
systemctl enable "$SERVICE_NAME" >/dev/null 2>&1
print_success "Service '$SERVICE_NAME' enabled for automatic startup"
print_info "Note: The service is enabled but not started yet. You'll start it after configuration."
fi
echo -e "${YELLOW}Step 5: Installing systemd service${NC}"
# Copy service file to systemd directory
cp "$SERVICE_FILE" "/etc/systemd/system/"
echo -e "${GREEN}Copied service file to /etc/systemd/system/${NC}"
# Reload systemd
systemctl daemon-reload
echo -e "${GREEN}Reloaded systemd configuration${NC}"
echo -e "${YELLOW}Step 6: Enabling service${NC}"
# Enable service to start on boot
systemctl enable "$SERVICE_NAME"
echo -e "${GREEN}Enabled $SERVICE_NAME service${NC}"
echo -e "${YELLOW}Step 7: Installing Python dependencies in virtual environment${NC}"
print_section "Step 7: Setting Up Python Virtual Environment"
print_info "Creating an isolated Python environment for the bot"
print_info "This ensures dependencies don't conflict with system Python packages"
# Create virtual environment
if [ -d "$INSTALL_DIR/venv" ]; then
print_warning "Virtual environment already exists, removing it..."
rm -rf "$INSTALL_DIR/venv"
fi
python3 -m venv "$INSTALL_DIR/venv"
echo -e "${GREEN}Created virtual environment${NC}"
print_success "Created virtual environment at $INSTALL_DIR/venv"
# Upgrade pip first
print_info "Upgrading pip to latest version"
"$INSTALL_DIR/venv/bin/pip" install --quiet --upgrade pip >/dev/null 2>&1 || true
# Install dependencies in venv
"$INSTALL_DIR/venv/bin/pip" install -r "$INSTALL_DIR/requirements.txt"
echo -e "${GREEN}Installed Python dependencies${NC}"
print_info "Installing Python dependencies from requirements.txt"
print_info "This may take a few minutes depending on your internet connection..."
if [ ! -f "$INSTALL_DIR/requirements.txt" ]; then
print_error "requirements.txt not found in installation directory"
exit 1
fi
"$INSTALL_DIR/venv/bin/pip" install --quiet -r "$INSTALL_DIR/requirements.txt" || {
print_error "Failed to install Python dependencies"
print_info "You may need to check your internet connection or Python version"
exit 1
}
print_success "Installed all Python dependencies"
# Update ownership of venv
chown -R "$SERVICE_USER:$SERVICE_GROUP" "$INSTALL_DIR/venv"
print_success "Set ownership of virtual environment"
print_section "Installation Complete!"
echo ""
echo -e "${GREEN}Installation completed successfully!${NC}"
print_success "MeshCore Bot has been successfully installed as a systemd service!"
echo ""
echo -e "${BLUE}Service Management Commands:${NC}"
echo " Start service: sudo systemctl start $SERVICE_NAME"
echo " Stop service: sudo systemctl stop $SERVICE_NAME"
echo " Restart service: sudo systemctl restart $SERVICE_NAME"
echo " Check status: sudo systemctl status $SERVICE_NAME"
echo " View logs: sudo journalctl -u $SERVICE_NAME -f"
echo " Disable service: sudo systemctl disable $SERVICE_NAME"
echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo -e "${BLUE}📋 Next Steps${NC}"
echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo ""
echo -e "${BLUE}Configuration:${NC}"
echo " Config file: $INSTALL_DIR/config.ini"
echo " Log directory: $LOG_DIR"
echo " Working directory: $INSTALL_DIR"
echo -e "${CYAN}1. Configure the bot:${NC}"
echo " ${YELLOW}sudo nano $INSTALL_DIR/config.ini${NC}"
echo " Edit the configuration file with your bot settings, API keys, and device information"
echo ""
echo -e "${YELLOW}Next steps:${NC}"
echo "1. Edit configuration: sudo nano $INSTALL_DIR/config.ini"
echo "2. Start the service: sudo systemctl start $SERVICE_NAME"
echo "3. Check status: sudo systemctl status $SERVICE_NAME"
if [[ "$IS_MACOS" == true ]]; then
echo -e "${CYAN}2. Start the service:${NC}"
echo " ${YELLOW}sudo launchctl load -w $LAUNCHD_DIR/$SERVICE_FILE${NC}"
echo " Or: ${YELLOW}sudo launchctl start $PLIST_NAME${NC}"
echo ""
echo -e "${CYAN}3. Verify it's running:${NC}"
echo " ${YELLOW}sudo launchctl list | grep $PLIST_NAME${NC}"
echo " Or check logs: ${YELLOW}tail -f $LOG_DIR/meshcore-bot.log${NC}"
echo ""
echo -e "${CYAN}4. View live logs (optional):${NC}"
echo " ${YELLOW}tail -f $LOG_DIR/meshcore-bot.log${NC}"
echo " Press Ctrl+C to exit log view"
else
echo -e "${CYAN}2. Start the service:${NC}"
echo " ${YELLOW}sudo systemctl start $SERVICE_NAME${NC}"
echo ""
echo -e "${CYAN}3. Verify it's running:${NC}"
echo " ${YELLOW}sudo systemctl status $SERVICE_NAME${NC}"
echo " You should see 'active (running)' in green"
echo ""
echo -e "${CYAN}4. View live logs (optional):${NC}"
echo " ${YELLOW}sudo journalctl -u $SERVICE_NAME -f${NC}"
echo " Press Ctrl+C to exit log view"
fi
echo ""
echo -e "${GREEN}The bot will now start automatically on system boot!${NC}"
echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo -e "${BLUE}🔧 Service Management Commands${NC}"
echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo ""
if [[ "$IS_MACOS" == true ]]; then
echo -e " ${CYAN}Start service:${NC} ${YELLOW}sudo launchctl start $PLIST_NAME${NC}"
echo -e " ${CYAN}Stop service:${NC} ${YELLOW}sudo launchctl stop $PLIST_NAME${NC}"
echo -e " ${CYAN}Restart service:${NC} ${YELLOW}sudo launchctl stop $PLIST_NAME && sudo launchctl start $PLIST_NAME${NC}"
echo -e " ${CYAN}Check status:${NC} ${YELLOW}sudo launchctl list | grep $PLIST_NAME${NC}"
echo -e " ${CYAN}View logs:${NC} ${YELLOW}tail -f $LOG_DIR/meshcore-bot.log${NC}"
echo -e " ${CYAN}View error logs:${NC} ${YELLOW}tail -f $LOG_DIR/meshcore-bot.error.log${NC}"
echo -e " ${CYAN}Unload service:${NC} ${YELLOW}sudo launchctl unload $LAUNCHD_DIR/$SERVICE_FILE${NC}"
echo -e " ${CYAN}Load service:${NC} ${YELLOW}sudo launchctl load $LAUNCHD_DIR/$SERVICE_FILE${NC}"
else
echo -e " ${CYAN}Start service:${NC} ${YELLOW}sudo systemctl start $SERVICE_NAME${NC}"
echo -e " ${CYAN}Stop service:${NC} ${YELLOW}sudo systemctl stop $SERVICE_NAME${NC}"
echo -e " ${CYAN}Restart service:${NC} ${YELLOW}sudo systemctl restart $SERVICE_NAME${NC}"
echo -e " ${CYAN}Check status:${NC} ${YELLOW}sudo systemctl status $SERVICE_NAME${NC}"
echo -e " ${CYAN}View logs:${NC} ${YELLOW}sudo journalctl -u $SERVICE_NAME -f${NC}"
echo -e " ${CYAN}View recent logs:${NC} ${YELLOW}sudo journalctl -u $SERVICE_NAME -n 100${NC}"
echo -e " ${CYAN}Disable auto-start:${NC} ${YELLOW}sudo systemctl disable $SERVICE_NAME${NC}"
echo -e " ${CYAN}Enable auto-start:${NC} ${YELLOW}sudo systemctl enable $SERVICE_NAME${NC}"
fi
echo ""
echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo -e "${BLUE}📁 Important File Locations${NC}"
echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo ""
echo -e " ${CYAN}Configuration file:${NC} ${YELLOW}$INSTALL_DIR/config.ini${NC}"
echo -e " ${CYAN}Log directory:${NC} ${YELLOW}$LOG_DIR${NC}"
echo -e " ${CYAN}Installation directory:${NC} ${YELLOW}$INSTALL_DIR${NC}"
if [[ "$IS_MACOS" == true ]]; then
echo -e " ${CYAN}Service plist:${NC} ${YELLOW}$LAUNCHD_DIR/$SERVICE_FILE${NC}"
else
echo -e " ${CYAN}Service file:${NC} ${YELLOW}$SYSTEMD_DIR/$SERVICE_NAME.service${NC}"
fi
echo ""
echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo -e "${BLUE}ℹ️ Additional Information${NC}"
echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
echo ""
print_info "The service is configured to:"
echo " • Start automatically on system boot"
if [[ "$IS_MACOS" == true ]]; then
echo " • Restart automatically if it crashes (with 10 second throttle)"
echo " • Run as user '$SERVICE_USER'"
echo " • Log to: $LOG_DIR/meshcore-bot.log"
echo ""
print_info "After editing config.ini, restart the service for changes to take effect:"
echo " ${YELLOW}sudo launchctl stop $PLIST_NAME && sudo launchctl start $PLIST_NAME${NC}"
else
echo " • Restart automatically if it crashes (with 10 second delay)"
echo " • Run as user '$SERVICE_USER' for security"
echo " • Log to systemd journal (view with journalctl)"
echo ""
print_info "After editing config.ini, restart the service for changes to take effect:"
echo " ${YELLOW}sudo systemctl restart $SERVICE_NAME${NC}"
fi
echo ""
print_success "Installation complete! The bot is ready to configure and start."
echo ""
modules/commands/alert_command.py
+929
View File
@@ -0,0 +1,929 @@
#!/usr/bin/env python3
"""
Alert command for the MeshCore Bot
Provides PulsePoint incident alerts for locations, zip codes, and street addresses
"""
import re
import base64
import hashlib
import json
import requests
from datetime import datetime, timezone
from typing import Optional, List, Dict, Tuple
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from .base_command import BaseCommand
from ..models import MeshMessage
from ..utils import geocode_zipcode_sync, geocode_city_sync, calculate_distance, rate_limited_nominatim_reverse_sync
# Incident type codes -> human readable (short versions for mesh)
CALL_TYPES = {
"AA": "Auto Aid", "MU": "Mutual Aid", "ST": "Strike Team",
"AC": "Aircraft Crash", "AE": "Aircraft Emerg", "AES": "Aircraft Standby", "LZ": "Landing Zone",
"AED": "AED Alarm", "OA": "Alarm", "CMA": "CO Alarm", "FA": "Fire Alarm",
"MA": "Manual Alarm", "SD": "Smoke Detector", "TRBL": "Trouble Alarm", "WFA": "Waterflow",
"FL": "Flooding", "LR": "Ladder Req", "LA": "Lift Assist",
"PA": "Police Assist", "PS": "Public Svc", "SH": "Hydrant",
"EX": "Explosion", "PE": "Pipeline Emerg", "TE": "Transformer",
"AF": "Appliance Fire", "CHIM": "Chimney Fire", "CF": "Commercial Fire",
"WSF": "Structure Fire", "WVEG": "Veg Fire", "CB": "Controlled Burn",
"ELF": "Electrical Fire", "EF": "Extinguished", "FIRE": "Fire",
"FULL": "Full Assignment", "IF": "Illegal Fire", "MF": "Marine Fire",
"OF": "Outside Fire", "PF": "Pole Fire", "GF": "Garbage Fire",
"RF": "Residential Fire", "SF": "Structure Fire", "VEG": "Veg Fire",
"VF": "Vehicle Fire", "WCF": "Working Comm Fire", "WRF": "Working Res Fire",
"BT": "Bomb Threat", "EE": "Electrical Emerg", "EM": "Emergency",
"ER": "Emergency", "GAS": "Gas Leak", "HC": "Hazmat",
"HMR": "Hazmat", "TD": "Tree Down", "WE": "Water Emerg",
"AI": "Arson Inv", "HMI": "Hazmat Inv", "INV": "Investigation",
"OI": "Odor Inv", "SI": "Smoke Inv",
"LO": "Lockout", "CL": "Comm Lockout", "RL": "Res Lockout", "VL": "Vehicle Lockout",
"IFT": "Med Transfer", "ME": "Medical", "MCI": "Mass Casualty",
"EQ": "Earthquake", "FLW": "Flood Warn", "TOW": "Tornado Warn", "TSW": "Tsunami Warn",
"CA": "Community", "FW": "Fire Watch", "NO": "Notification",
"STBY": "Standby", "TEST": "Test", "TRNG": "Training", "UNK": "Unknown",
"AR": "Animal Rescue", "CR": "Cliff Rescue", "CSR": "Confined Space",
"ELR": "Elevator Rescue", "RES": "Rescue", "RR": "Rope Rescue",
"TR": "Tech Rescue", "TNR": "Trench Rescue", "USAR": "Urban SAR",
"VS": "Vessel Sinking", "WR": "Water Rescue",
"TCE": "Major TC", "RTE": "Train Emerg",
"TC": "Traffic Collision", "TCS": "TC w/Structure", "TCT": "TC w/Train",
"WA": "Wires Arcing", "WD": "Wires Down"
}
# Unit dispatch status codes
UNIT_STATUS = {
"DP": "Dispatched",
"ER": "Enroute",
"OS": "On Scene",
"AV": "Available",
"TR": "Transport",
"TA": "Arrived",
"CL": "Cleared"
}
def _derive_key(salt: bytes) -> bytes:
"""Derive AES key from the obfuscated password."""
e = "CommonIncidents"
password = e[13] + e[1] + e[2] + "brady" + "5" + "r" + e.lower()[6] + e[5] + "gs"
hasher = hashlib.md5()
key = b''
block = None
while len(key) < 32:
if block:
hasher.update(block)
hasher.update(password.encode())
hasher.update(salt)
block = hasher.digest()
hasher = hashlib.md5()
key += block
return key[:32]
def _decrypt(data: dict) -> dict:
"""Decrypt PulsePoint's encrypted response."""
ct = base64.b64decode(data["ct"])
iv = bytes.fromhex(data["iv"])
salt = bytes.fromhex(data["s"])
key = _derive_key(salt)
cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
decryptor = cipher.decryptor()
out = decryptor.update(ct) + decryptor.finalize()
out = out[1:out.rindex(b'"')].decode()
out = out.replace(r'\"', r'"')
return json.loads(out)
def _parse_time(iso_str: str) -> datetime:
"""Parse ISO timestamp to datetime and convert to local time."""
if not iso_str:
return None
try:
dt = datetime.fromisoformat(iso_str.replace('Z', '+00:00'))
# Convert to local time if timezone-aware
if dt.tzinfo is not None:
dt = dt.astimezone()
return dt
except:
return None
def _time_ago(dt: datetime) -> str:
"""Format datetime as relative time string."""
if not dt:
return ""
# Use local time for comparison
now = datetime.now().astimezone()
# Ensure dt is timezone-aware (should be after _parse_time conversion)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=now.tzinfo)
diff = now - dt
mins = int(diff.total_seconds() / 60)
if mins < 1:
return "now"
elif mins < 60:
return f"{mins}m ago"
elif mins < 1440:
return f"{mins // 60}h {mins % 60}m ago"
else:
return f"{mins // 1440}d ago"
class AlertCommand(BaseCommand):
"""Handles alert/incident commands using PulsePoint API"""
# Plugin metadata
name = "alert"
keywords = ['alert', 'alerts', 'incident', 'incidents']
description = "Get active emergency incidents (usage: alert seattle, alert 98258, alert 178th seattle, alert seattle all)"
category = "emergency"
cooldown_seconds = 10 # 10 second cooldown to prevent API abuse
def __init__(self, bot):
super().__init__(bot)
self.url_timeout = 10
self.db_manager = bot.db_manager
# Load agencies from config (separate cities and counties)
self.city_agencies, self.county_agencies = self._load_agencies()
# Get max distance from config (default 20km, about 12 miles)
self.max_distance_km = self.get_config_value('Alert_Command', 'max_distance_km', fallback=20.0, value_type='float')
# Get max incident age in hours (default 24 hours) - filter out incidents older than this
self.max_incident_age_hours = self.get_config_value('Alert_Command', 'max_incident_age_hours', fallback=24.0, value_type='float')
def can_execute(self, message: MeshMessage) -> bool:
"""Check if this command can be executed with the given message"""
# Check if alert command is enabled
alert_enabled = self.get_config_value('Alert_Command', 'alert_enabled', fallback=True, value_type='bool')
if not alert_enabled:
return False
# Call parent can_execute() which includes channel checking, cooldown, etc.
return super().can_execute(message)
def _load_agencies(self) -> Tuple[Dict[str, str], Dict[str, str]]:
"""Load agency IDs from config, separating cities and counties"""
cities = {}
counties = {}
if self.bot.config.has_section('Alert_Command'):
for key, value in self.bot.config.items('Alert_Command'):
# New format: agency.city.seattle or agency.county.king
if key.startswith('agency.city.'):
city = key.replace('agency.city.', '').lower()
cities[city] = value.strip()
elif key.startswith('agency.county.'):
county = key.replace('agency.county.', '').lower()
counties[county] = value.strip()
# Legacy format support: agency.* (treat as county for backward compatibility)
elif key.startswith('agency.'):
name = key.replace('agency.', '').lower()
counties[name] = value.strip()
# Old format: agency_* (treat as county for backward compatibility)
elif key.startswith('agency_'):
name = key.replace('agency_', '').lower()
counties[name] = value.strip()
return cities, counties
def _normalize_location_key(self, location: str) -> str:
"""Normalize location name to match config key format (spaces -> underscores)"""
return location.lower().replace(' ', '_')
def _get_agency_ids(self, location: str = None, location_type: str = None) -> Optional[str]:
"""Get agency IDs for a city or county, or default to all configured agencies"""
if location:
location_lower = location.lower()
location_normalized = self._normalize_location_key(location)
# If location_type is specified, only check that type
if location_type == "city":
# Try normalized first (with underscore), then original (with space)
if location_normalized in self.city_agencies:
return self.city_agencies[location_normalized]
if location_lower in self.city_agencies:
return self.city_agencies[location_lower]
# City not found in config, return None to indicate we should use all agencies
return None
elif location_type == "county":
# Try normalized first (with underscore), then original (with space)
if location_normalized in self.county_agencies:
return self.county_agencies[location_normalized]
if location_lower in self.county_agencies:
return self.county_agencies[location_lower]
# County not found, check aliases
aliases = {
'sno': 'snohomish',
'sea': 'king', # 'sea' alias maps to King County
'tac': 'pierce',
'all': 'puget_sound'
}
if location_lower in aliases:
alias_target = aliases[location_lower]
if alias_target in self.county_agencies:
return self.county_agencies[alias_target]
return None
# If no location_type specified, check both (city first, then county)
# Try normalized first (with underscore), then original (with space)
if location_normalized in self.city_agencies:
return self.city_agencies[location_normalized]
if location_lower in self.city_agencies:
return self.city_agencies[location_lower]
if location_normalized in self.county_agencies:
return self.county_agencies[location_normalized]
if location_lower in self.county_agencies:
return self.county_agencies[location_lower]
# Check aliases (these map to counties)
aliases = {
'sno': 'snohomish',
'sea': 'king', # 'sea' alias maps to King County
'tac': 'pierce',
'all': 'puget_sound'
}
if location_lower in aliases:
alias_target = aliases[location_lower]
if alias_target in self.county_agencies:
return self.county_agencies[alias_target]
# Default: combine all agencies from both cities and counties
all_agencies = []
for agency_list in list(self.city_agencies.values()) + list(self.county_agencies.values()):
all_agencies.append(agency_list)
return ",".join(all_agencies)
def _fetch_incidents(self, agency_ids: str) -> List[Dict]:
"""Fetch active incidents from PulsePoint"""
url = "https://api.pulsepoint.org/v1/webapp"
params = {"resource": "incidents", "agencyid": agency_ids}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Origin": "https://web.pulsepoint.org",
"Referer": "https://web.pulsepoint.org/"
}
try:
resp = requests.get(url, params=params, headers=headers, timeout=self.url_timeout)
resp.raise_for_status()
encrypted = resp.json()
decrypted = _decrypt(encrypted)
incidents = []
seen_ids = set() # Track incident IDs to avoid duplicates
# Only fetch active incidents (not recent/cleared)
# Filter by age to exclude very old "active" incidents
now = datetime.now(timezone.utc)
max_age = self.max_incident_age_hours * 3600 # Convert hours to seconds
for inc in decrypted.get("incidents", {}).get("active", []):
incident_id = inc.get("ID")
# Skip if we've already seen this incident ID (deduplication)
if incident_id in seen_ids:
continue
seen_ids.add(incident_id)
call_type = inc.get("PulsePointIncidentCallType", "UNK")
call_time = _parse_time(inc.get("CallReceivedDateTime"))
# Filter out incidents older than max_incident_age_hours
if call_time:
# Ensure timezone-aware for comparison
if call_time.tzinfo is None:
call_time = call_time.replace(tzinfo=timezone.utc)
else:
call_time = call_time.astimezone(timezone.utc)
age_seconds = (now - call_time).total_seconds()
if age_seconds > max_age:
# Incident is too old, skip it
continue
# Parse units with status
units = []
for u in inc.get("Unit", []):
unit_id = u.get("UnitID", "?")
status = u.get("PulsePointDispatchStatus", "?")
units.append({
"id": unit_id,
"status_code": status,
"status": UNIT_STATUS.get(status, status)
})
# Parse address
full_addr = inc.get("FullDisplayAddress", "Unknown")
# Try splitting on ", " first (most common), then on "," if that fails
if ", " in full_addr:
addr_parts = full_addr.split(", ", 1)
elif "," in full_addr:
addr_parts = full_addr.split(",", 1)
else:
addr_parts = [full_addr]
street = addr_parts[0].strip()
city = addr_parts[1].strip() if len(addr_parts) > 1 else ""
incidents.append({
"id": incident_id,
"type_code": call_type,
"type": CALL_TYPES.get(call_type, call_type),
"address": full_addr,
"street": street,
"city": city,
"latitude": float(inc.get("Latitude", 0)),
"longitude": float(inc.get("Longitude", 0)),
"agency": inc.get("AgencyID"),
"time": call_time,
"time_ago": _time_ago(call_time),
"units": units,
"unit_ids": [u["id"] for u in units],
"raw": inc
})
return incidents
except Exception as e:
self.logger.error(f"Error fetching PulsePoint incidents: {e}")
return []
def _parse_query(self, query: str) -> Tuple[str, Optional[str], Optional[float], Optional[float]]:
"""
Parse query string to determine search type.
Returns:
Tuple of (query_type, location, lat, lon)
query_type: "zipcode", "coordinates", "street_city", "city", "county"
"""
query = query.strip()
# Check for coordinates (lat,lon or lat, lon)
coord_match = re.match(r'^(-?\d+\.?\d*),?\s*(-?\d+\.?\d*)$', query)
if coord_match:
try:
lat = float(coord_match.group(1))
lon = float(coord_match.group(2))
if -90 <= lat <= 90 and -180 <= lon <= 180:
return ("coordinates", None, lat, lon)
except:
pass
# Check for zipcode (5 digits)
if re.match(r'^\d{5}$', query):
return ("zipcode", query, None, None)
# Check for street + city pattern (e.g., "178th seattle", "main seattle")
# If query has 2+ words, check if first word looks like a street name
parts = query.split()
if len(parts) >= 2:
first_word = parts[0].lower()
# Check if first word looks like a street name:
# - Ends with a number (e.g., "178th", "5th", "123rd")
# - Ends with street suffix (e.g., "main", "oak", "park" - but these are ambiguous)
# - Is a known street prefix (e.g., "ne", "nw", "se", "sw" for directional)
looks_like_street = (
bool(re.search(r'\d+(st|nd|rd|th)$', first_word)) or # Ends with number + ordinal
first_word in ['ne', 'nw', 'se', 'sw', 'n', 's', 'e', 'w'] or # Directional prefixes
first_word.endswith(('st', 'nd', 'rd', 'th')) # Ordinal suffix
)
if looks_like_street:
# First word looks like a street, try to split into street and city
for i in range(1, len(parts)):
street_part = ' '.join(parts[:i])
city_part = ' '.join(parts[i:])
# Check if city_part looks like a city name (not a street suffix)
# If city_part is a known city/county, or doesn't end with street suffix, it's likely a city
city_lower = city_part.lower()
if (city_lower in self.city_agencies or
city_lower in self.county_agencies or
city_lower in ['sno', 'sea', 'tac', 'all'] or
not city_lower.endswith(('st', 'street', 'ave', 'avenue', 'rd', 'road', 'blvd', 'boulevard',
'dr', 'drive', 'ct', 'court', 'ln', 'lane', 'way', 'pl', 'place'))):
# This looks like street + city
return ("street_city", f"{street_part} {city_part}", None, None)
# If no good split found, assume first word is street, rest is city
street_part = parts[0]
city_part = ' '.join(parts[1:])
return ("street_city", f"{street_part} {city_part}", None, None)
# else: first word doesn't look like a street, treat entire query as city name (fall through)
# Check if it's a known county alias (short codes only)
# County aliases: 'sno' (snohomish), 'sea' (king), 'tac' (pierce), 'all' (all counties)
query_lower = query.lower()
if query_lower in ['sno', 'sea', 'tac', 'all']:
return ("county", query, None, None)
# Normalize query to match config key format (spaces -> underscores)
# Config keys use underscores (e.g., "lake_stevens"), but queries may use spaces
query_normalized = self._normalize_location_key(query)
# Check if it's a configured city name (from config) - check cities first
# Try both normalized (with underscore) and original (with space) formats
if query_normalized in self.city_agencies or query_lower in self.city_agencies:
return ("city", query, None, None)
# Check if it's a configured county name (from config)
# Try both normalized (with underscore) and original (with space) formats
if query_normalized in self.county_agencies or query_lower in self.county_agencies:
return ("county", query, None, None)
# Default: treat as city name (handles single-word and multi-word city names)
return ("city", query, None, None)
def _match_street_name(self, incidents: List[Dict], street_query: str) -> Tuple[List[Dict], List[Dict]]:
"""Split incidents into matched and unmatched by street name"""
street_lower = street_query.lower().strip()
matched = []
unmatched = []
for inc in incidents:
street = inc.get("street", "").lower()
# Check if street query appears in the incident's street name
# This handles cases like "178th" matching "178TH AVE" or "NE 178TH ST"
if street_lower in street:
matched.append(inc)
else:
unmatched.append(inc)
return matched, unmatched
def _matches_city(self, inc: Dict, city_query: str) -> bool:
"""Check if incident matches the city name by substring matching on address field"""
city_query_lower = city_query.lower().strip()
address = inc.get("address", "").lower().strip()
city = inc.get("city", "").lower().strip()
# Check if city name appears in the address field
return city_query_lower in address
def _get_city_match_priority(self, inc: Dict, city_query: str) -> int:
"""
Get priority score for city match (higher = better match).
Returns 0 if no match, higher numbers for better matches.
We prioritize matches where the city name appears at the end of the address
(after a comma), as this is the most reliable indicator. The city field
can be inaccurate (e.g., showing "SEATTLE" for addresses in King County
but not actually in Seattle).
"""
city_query_lower = city_query.lower().strip()
address = inc.get("address", "").lower().strip()
city_query_clean = city_query_lower.split(',')[0].strip()
# Priority 2: City appears at end of address (most reliable - typical format: "STREET, CITY" or "STREET, CITY, STATE")
# This is the most trustworthy match since it's based on the actual address format
import re
end_pattern = r',\s*' + re.escape(city_query_clean) + r'(?:\s*,\s*[A-Z]{2})?$'
if re.search(end_pattern, address, re.IGNORECASE):
return 2
# Priority 1: City appears anywhere in address (substring match, less reliable)
# This catches cases where city might be mentioned but not at the end
if city_query_clean in address:
return 1
# Priority 0: No match
return 0
def _match_city_name(self, incidents: List[Dict], city_query: str) -> Tuple[List[Dict], List[Dict]]:
"""Split incidents into matched and unmatched by city name using simple substring matching on address"""
matched = []
unmatched = []
for inc in incidents:
if self._matches_city(inc, city_query):
matched.append(inc)
else:
unmatched.append(inc)
return matched, unmatched
def _sort_by_time(self, incidents: List[Dict]) -> List[Dict]:
"""Sort incidents by time (most recent first)"""
def get_time_key(inc):
time = inc.get("time")
if time is None:
return datetime.min.replace(tzinfo=timezone.utc)
# Ensure timezone-aware
if time.tzinfo is None:
time = time.replace(tzinfo=timezone.utc)
return time
return sorted(incidents, key=get_time_key, reverse=True)
def _sort_by_distance_then_time(self, incidents: List[Dict], lat: float, lon: float, max_distance: float = None) -> List[Dict]:
"""
Sort incidents by distance first, then by time (most recent first) within same distance.
This maintains distance ordering while sorting by time as secondary criteria.
"""
scored_incidents = []
for inc in incidents:
if not self._has_valid_coordinates(inc):
continue
inc_lat = inc.get("latitude", 0)
inc_lon = inc.get("longitude", 0)
distance = calculate_distance(lat, lon, inc_lat, inc_lon)
inc["_distance"] = distance
# Filter by max_distance if specified
if max_distance is None or distance <= max_distance:
# Get time for secondary sort
time = inc.get("time")
if time is None:
time_key = datetime.min.replace(tzinfo=timezone.utc)
else:
if time.tzinfo is None:
time = time.replace(tzinfo=timezone.utc)
time_key = time
inc["_time_key"] = time_key
scored_incidents.append(inc)
# Sort by distance first, then by time (most recent first)
return sorted(scored_incidents, key=lambda x: (x.get("_distance", float('inf')), -x.get("_time_key", datetime.min).timestamp()))
def _has_valid_coordinates(self, inc: Dict) -> bool:
"""Check if incident has valid coordinates"""
inc_lat = inc.get("latitude", 0)
inc_lon = inc.get("longitude", 0)
# Valid if both are non-zero and within valid ranges
return (inc_lat != 0.0 and inc_lon != 0.0 and
-90 <= inc_lat <= 90 and -180 <= inc_lon <= 180)
def _sort_by_distance(self, incidents: List[Dict], lat: float, lon: float, max_distance: float = None) -> List[Dict]:
"""
Sort incidents by distance from given coordinates.
Args:
incidents: List of incident dicts
lat: Target latitude
lon: Target longitude
max_distance: Optional maximum distance in km (incidents beyond this are excluded)
Returns:
Sorted list of incidents (closest first). Only includes incidents with valid coordinates.
"""
scored_incidents = []
for inc in incidents:
if not self._has_valid_coordinates(inc):
# Skip incidents without valid coordinates - they'll be handled separately
continue
inc_lat = inc.get("latitude", 0)
inc_lon = inc.get("longitude", 0)
distance = calculate_distance(lat, lon, inc_lat, inc_lon)
inc["_distance"] = distance
# Filter by max_distance if specified
if max_distance is None or distance <= max_distance:
scored_incidents.append(inc)
# Sort by distance (closest first)
return sorted(scored_incidents, key=lambda x: x.get("_distance", float('inf')))
def _format_incident_compact(self, inc: Dict) -> str:
"""Format a single incident in compact format"""
# Get first unit with status icon
unit_str = ""
if inc.get("units"):
u = inc["units"][0]
status_icon = {"DP": "⏳", "ER": "🚗", "OS": "📍", "TR": "🏥"}.get(u["status_code"], "")
unit_str = f" [{u['id']}{status_icon}]"
# Shorten city name
city = inc.get("city", "")
if city:
city = city.replace(", WA", "").replace(" COUNTY", " CO")
city_part = f", {city}"
else:
city_part = ""
time_ago = inc.get("time_ago", "")
time_part = f" ({time_ago})" if time_ago else ""
return f"{inc['type']}: {inc['street']}{city_part}{time_part}{unit_str}"
def _format_response(self, incidents: List[Dict], max_length: int = 130) -> str:
"""Format incidents into a single message, limiting to max_length"""
if not incidents:
return "🚨 No active incidents"
lines = ["🚨"]
# Start with emoji (2 chars) + newline (1 char) = 3 chars
current_length = 3
remaining = len(incidents)
for i, inc in enumerate(incidents):
line = self._format_incident_compact(inc)
# Length includes the line content + newline character
line_length = len(line) + 1
# Check if we can fit this line at all
if current_length + line_length > max_length:
# Can't fit this line
if len(lines) > 1: # At least one incident shown
lines.append(f"({remaining} more)")
break
# Check if this is the last incident
is_last = (remaining == 1)
if is_last:
# Last incident, no "(X more)" needed, add it
lines.append(line)
current_length += line_length
remaining -= 1
else:
# Not the last incident, check if we can fit line + "(X more)"
more_text = f" ({remaining - 1} more)"
if current_length + line_length + len(more_text) > max_length:
# Can't fit both line and "(X more)", show count instead
if len(lines) > 1: # At least one incident shown
lines.append(f"({remaining} more)")
break
else:
# Can fit both line and "(X more)", add the line
lines.append(line)
current_length += line_length
remaining -= 1
# Build final message
final_message = "\n".join(lines)
# Safety check: ensure we don't exceed max_length (shouldn't happen, but be safe)
if len(final_message) > max_length:
# Find the last complete line before max_length
# Look for the last newline that would keep us under the limit
last_newline = final_message.rfind('\n', 0, max_length - 15) # Reserve 15 chars for "(X more)"
if last_newline > 0 and len(lines) > 1:
# Truncate at the last complete line
final_message = final_message[:last_newline]
# Add count if there are remaining incidents
if remaining > 0:
final_message += f"\n({remaining} more)"
else:
# Fallback: just truncate (shouldn't happen with proper logic above)
final_message = final_message[:max_length].rstrip()
return final_message
async def _send_all_response(self, message: MeshMessage, incidents: List[Dict]):
"""Send up to 10 incidents in multiple messages, grouping efficiently"""
import asyncio
if not incidents:
await self.send_response(message, "🚨 No active incidents")
return
# Build messages efficiently, grouping incidents to fit within 130 chars
messages = []
header = f"🚨 {len(incidents)} incident(s):"
# Start first message with header
current_lines = [header]
current_length = len(header)
for inc in incidents:
incident_text = self._format_incident_compact(inc)
incident_length = len(incident_text)
# Calculate what the message would look like with this incident added
# Need to account for newline character between lines
test_length = current_length + 1 + incident_length # +1 for newline
# Check if this incident fits in the current message
if test_length <= 130:
# It fits, add it
current_lines.append(incident_text)
current_length = test_length
else:
# Doesn't fit - finalize current message and start new one
if len(current_lines) > 1: # Has at least header + one incident
messages.append("\n".join(current_lines))
else:
# Only header, must add at least one incident even if it exceeds limit
current_lines.append(incident_text)
messages.append("\n".join(current_lines))
current_lines = []
current_length = 0
continue
# Start new message with this incident (no header for subsequent messages)
current_lines = [incident_text]
current_length = incident_length
# Add the last message if it has content
if current_lines:
# If we only have header, add first incident
if len(current_lines) == 1 and len(incidents) > 0:
current_lines.append(self._format_incident_compact(incidents[0]))
messages.append("\n".join(current_lines))
# Send all messages with delays between them
for i, msg in enumerate(messages):
await self.send_response(message, msg)
# Wait between messages (except after the last one)
if i < len(messages) - 1:
await asyncio.sleep(2.0)
async def execute(self, message: MeshMessage) -> bool:
"""Execute the alert command"""
content = message.content.strip()
# Parse command
parts = content.split(None, 1)
if len(parts) < 2:
# No query provided, use default location or show help
await self.send_response(message, "Usage: alert <city|zipcode|street city|lat,lon|county> [all]")
return True
query = parts[1].strip()
# Check for "all" flag at the end
show_all = False
if query.lower().endswith(' all'):
show_all = True
query = query[:-4].strip() # Remove " all" from the end
try:
# Parse the query
query_type, location, lat, lon = self._parse_query(query)
self.logger.debug(f"Parsed query '{query}' as type: {query_type}, location: {location}")
# Get agency IDs based on query type
if query_type == "county":
agency_ids = self._get_agency_ids(location, "county")
elif query_type == "city":
# For city queries, try to get city-specific agencies, fall back to all
agency_ids = self._get_agency_ids(location, "city")
if agency_ids is None:
# No city-specific agencies configured, use all agencies
agency_ids = self._get_agency_ids()
else:
# For other queries (zipcode, coordinates, street_city), use all agencies
agency_ids = self._get_agency_ids() # Default to all
# Fetch incidents
incidents = self._fetch_incidents(agency_ids)
if not incidents:
await self.send_response(message, "🚨 No active incidents")
return True
# Process based on query type
if query_type == "coordinates":
# Sort by distance with configurable max distance, then by time
incidents = self._sort_by_distance_then_time(incidents, lat, lon, max_distance=self.max_distance_km)
elif query_type == "zipcode":
# Geocode zipcode and get city name
zip_lat, zip_lon = geocode_zipcode_sync(self.bot, location)
zip_city = None
if zip_lat and zip_lon:
# Get city name from zipcode via reverse geocoding
try:
reverse_location = rate_limited_nominatim_reverse_sync(self.bot, f"{zip_lat}, {zip_lon}", timeout=10)
if reverse_location and reverse_location.raw:
address = reverse_location.raw.get('address', {})
zip_city = (address.get('city') or
address.get('town') or
address.get('village') or
address.get('hamlet') or
address.get('municipality') or
address.get('suburb') or '')
if zip_city:
zip_city = zip_city.lower().strip()
self.logger.debug(f"Zipcode {location} maps to city: {zip_city}")
except Exception as e:
self.logger.debug(f"Error getting city from zipcode: {e}")
# Split incidents by coordinate validity
with_coords = [inc for inc in incidents if self._has_valid_coordinates(inc)]
without_coords = [inc for inc in incidents if not self._has_valid_coordinates(inc)]
# Prioritize incidents that match the zipcode's city
if zip_city:
# Filter by city name match, then sort by distance and time
matched_coords, _ = self._match_city_name(with_coords, zip_city)
matched_no_coords, _ = self._match_city_name(without_coords, zip_city)
# Sort matched incidents by distance (for those with coords), then time
matched_coords = self._sort_by_distance_then_time(matched_coords, zip_lat, zip_lon, max_distance=None)
matched_no_coords = self._sort_by_time(matched_no_coords)
# If we have matches, show those. Otherwise, show nearby incidents within max distance
if len(matched_coords) > 0 or len(matched_no_coords) > 0:
incidents = matched_coords + matched_no_coords
else:
# No city matches, show nearby incidents within max distance
nearby_coords = self._sort_by_distance_then_time(with_coords, zip_lat, zip_lon, max_distance=self.max_distance_km)
nearby_no_coords = self._sort_by_time(without_coords)
incidents = nearby_coords + nearby_no_coords
else:
# No city name available, just sort by distance
incidents = self._sort_by_distance_then_time(incidents, zip_lat, zip_lon, max_distance=self.max_distance_km)
elif query_type == "street_city":
# Extract street and city
parts = location.split(None, 1)
if len(parts) == 2:
street_query, city_query = parts
# Geocode city first
result = geocode_city_sync(self.bot, city_query, include_address_info=False)
city_lat, city_lon = None, None
if len(result) >= 2:
city_lat, city_lon = result[0], result[1]
# Split incidents by coordinate validity
with_coords = [inc for inc in incidents if self._has_valid_coordinates(inc)]
without_coords = [inc for inc in incidents if not self._has_valid_coordinates(inc)]
# Process incidents with coordinates
if city_lat and city_lon:
# Sort by distance (closest first) with max distance filter
with_coords = self._sort_by_distance(with_coords, city_lat, city_lon, max_distance=self.max_distance_km)
# Prioritize street-matched incidents by re-sorting
matched_street_coords, unmatched_street_coords = self._match_street_name(with_coords, street_query)
# Re-sort both groups by distance then time to maintain proximity ordering
matched_street_coords = self._sort_by_distance_then_time(matched_street_coords, city_lat, city_lon, max_distance=self.max_distance_km)
unmatched_street_coords = self._sort_by_distance_then_time(unmatched_street_coords, city_lat, city_lon, max_distance=self.max_distance_km)
with_coords = matched_street_coords + unmatched_street_coords
else:
# Geocoding failed - fall back to address matching for incidents with coordinates
matched_street_coords, unmatched_street_coords = self._match_street_name(with_coords, street_query)
matched_city_coords, _ = self._match_city_name(matched_street_coords + unmatched_street_coords, city_query)
with_coords = matched_city_coords
# Sort by time
with_coords = self._sort_by_time(with_coords)
# Process incidents without coordinates: match by street name and city name
matched_street, unmatched_street = self._match_street_name(without_coords, street_query)
# Also match by city name in address for those without coordinates
matched_city, _ = self._match_city_name(matched_street + unmatched_street, city_query)
without_coords = matched_city
# Sort by time
without_coords = self._sort_by_time(without_coords)
# Combine: incidents with coordinates (sorted by distance or matched by address) first, then address-matched ones without coordinates
incidents = with_coords + without_coords
elif query_type == "city":
# Filter incidents by city name match - ONLY show matches where city appears at end of address
# This is the most reliable indicator and avoids false positives
matched, _ = self._match_city_name(incidents, location)
# Only keep incidents where city name appears at end of address (Priority 2)
# This ensures we only show incidents that are actually in the queried city
high_priority = []
for inc in matched:
priority = self._get_city_match_priority(inc, location)
if priority >= 2: # Only city at end of address
high_priority.append(inc)
else:
# Log why an incident was excluded (for debugging)
self.logger.debug(f"Excluding incident: {inc.get('address', 'N/A')[:60]} (priority: {priority})")
# Sort by time (most recent first)
incidents = self._sort_by_time(high_priority)
self.logger.debug(f"City query '{location}': {len(incidents)} matches (city at end of address only), {len(matched) - len(high_priority)} excluded")
elif query_type == "county":
# For county queries, return all incidents from that county (no city filtering)
# Sort by time (most recent first)
incidents = self._sort_by_time(incidents)
self.logger.debug(f"County query '{location}': returning {len(incidents)} incidents (no city filtering)")
else:
# Unknown query type - this shouldn't happen, but log it
self.logger.warning(f"Unknown query type: {query_type} for query: {query}")
# Default: sort by time
incidents = self._sort_by_time(incidents)
# Limit to 10 incidents if "all" mode is enabled
if show_all:
incidents = incidents[:10]
await self._send_all_response(message, incidents)
else:
# Format and send response (compact mode)
response = self._format_response(incidents)
await self.send_response(message, response)
return True
except Exception as e:
self.logger.error(f"Error in alert command: {e}")
import traceback
self.logger.error(traceback.format_exc())
await self.send_response(message, f"Error fetching alerts: {str(e)}")
return True