2 Commits

Author SHA1 Message Date
serversdwn
1b8c63025f doc: update readme v1.4 2026-03-13 17:52:59 -04:00
serversdwn
0807e09047 feat: windows installer with remote updates and remote management added. 2026-03-13 17:40:28 -04:00
8 changed files with 1154 additions and 93 deletions

View File

@@ -6,6 +6,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
--- ---
## [1.4.0] - 2026-03-12
### Added
- `series3_tray.py` — system tray launcher using `pystray` + `Pillow`. Color-coded icon (green=OK, amber=Pending, red=Missing, purple=Error, grey=Starting). Right-click menu shows live status, unit count, last scan age, Open Log Folder, and Exit.
- `run_watcher(state, stop_event)` in `series3_watcher.py` for background thread use by the tray. Shared `state` dict updated on every scan cycle with status, unit list, last scan time, and last error.
- Interruptible sleep in watcher loop — tray exit is immediate, no waiting out the full scan interval.
### Changed
- `main()` now calls `run_watcher()` — standalone behavior unchanged.
- `requirements.txt` updated to document tray dependencies (`pystray`, `Pillow`); watcher itself remains stdlib-only.
---
## [1.3.0] - 2026-03-12 ## [1.3.0] - 2026-03-12
### Changed ### Changed

182
README.md
View File

@@ -1,84 +1,136 @@
# Series 3 Watcher v1.3 # Series 3 Watcher v1.4
A lightweight Python script that monitors Instantel **Series 3 (Minimate)** call-in activity on a Blastware server. Monitors Instantel **Series 3 (Minimate)** call-in activity on a Blastware server. Runs as a **system tray app** that starts automatically on login, reports heartbeats to terra-view, and self-updates from Gitea.
It scans the event folder, reads `.MLG` headers to identify unit IDs, and prints a live status table showing: ---
- Last event received ## Deployment (Recommended — Installer)
- Age since last call-in
- OK / Pending / Missing states
- Detected units (no roster required)
- Optional API heartbeat to Seismograph Fleet Manager backend
This script is part of the larger **Seismograph Fleet Manager** project. The easiest way to deploy to a field machine is the pre-built Windows installer.
1. Download `series3-watcher-setup.exe` from the [latest release](https://gitea.serversdown.net/serversdown/series3-watcher/releases) on Gitea.
2. Run the installer on the target machine. It installs to `C:\Program Files\Series3Watcher\` and adds a shortcut to the user's Startup folder.
3. On first launch the **Setup Wizard** opens automatically — fill in the terra-view URL and Blastware path, then click **Save & Start**.
4. A coloured dot appears in the system tray. Done.
The watcher will auto-start on every login from that point on.
### Auto-Updates
The watcher checks [Gitea](https://gitea.serversdown.net/serversdown/series3-watcher) for a newer release approximately every 5 minutes. When a newer `.exe` is found it downloads it silently, swaps the file, and relaunches — no user action required.
Updates can also be pushed remotely from terra-view → **Settings → Developer → Watcher Manager**.
---
## Building from Source
Prerequisites (on a **Python 3.8** machine — required for Win7 compatibility):
```
pip install pystray Pillow pyinstaller
```
Then run:
```
build.bat
```
This produces `dist\series3-watcher.exe`. To build the installer, open `installer.iss` in [Inno Setup](https://jrsoftware.org/isinfo.php) and click Build.
---
## Running Without the Installer (Dev / Debug)
```
pip install -r requirements.txt
python series3_tray.py # tray app (recommended)
python series3_watcher.py # console-only, no tray
```
`config.ini` must exist in the same directory. Copy `config-template.ini` to `config.ini` and edit it, or just run `series3_tray.py` — the wizard will create it on first run.
---
## Configuration
All settings live in `config.ini`. The Setup Wizard covers every field, but here's the reference:
### API / terra-view
| Key | Description |
|-----|-------------|
| `API_ENABLED` | `true` to send heartbeats to terra-view |
| `API_URL` | terra-view heartbeat endpoint, e.g. `http://192.168.1.10:8000/api/series3/heartbeat` |
| `API_INTERVAL_SECONDS` | How often to POST (default `300`) |
| `SOURCE_ID` | Identifier for this machine (defaults to hostname) |
| `SOURCE_TYPE` | Always `series3_watcher` |
### Paths
| Key | Description |
|-----|-------------|
| `SERIES3_PATH` | Blastware autocall folder, e.g. `C:\Blastware 10\Event\autocall home` |
| `MAX_EVENT_AGE_DAYS` | Ignore `.MLG` files older than this (default `365`) |
| `LOG_FILE` | Path to the log file |
### Scanning
| Key | Description |
|-----|-------------|
| `SCAN_INTERVAL_SECONDS` | How often to scan the folder (default `300`) |
| `OK_HOURS` | Age threshold for OK status (default `12`) |
| `MISSING_HOURS` | Age threshold for Missing status (default `24`) |
| `MLG_HEADER_BYTES` | Bytes to read from each `.MLG` header for unit ID (default `2048`) |
| `RECENT_WARN_DAYS` | Log unsniffable files newer than this window |
### Logging
| Key | Description |
|-----|-------------|
| `ENABLE_LOGGING` | `true` / `false` |
| `LOG_RETENTION_DAYS` | Auto-clear log after this many days (default `30`) |
| `COLORIZE` | ANSI colours in console — leave `false` on Win7 |
---
## Tray Icon
| Colour | Meaning |
|--------|---------|
| Grey | Starting / no scan yet |
| Green | All detected units OK |
| Yellow | At least one unit Pending |
| Red | At least one unit Missing, or error |
Right-click the icon for: status, per-unit list, Settings, Open Log Folder, Exit.
---
## terra-view Integration
When `API_ENABLED = true`, the watcher POSTs a telemetry payload to terra-view on each heartbeat interval. terra-view updates the emitter table and tracks the watcher process itself (version, last seen, log tail) in the Watcher Manager.
To view connected watchers: **Settings → Developer → Watcher Manager**.
--- ---
## Requirements ## Requirements
- Python 3.8 (Windows 7 compatible) - Windows 7 or later
- Blastware 10 event folder available locally - Python 3.8 (only needed if running from source — not needed with the installer)
- `config.ini` in the same directory as the script - Blastware 10 event folder accessible on the local machine
Install dependencies with:
`pip install -r requirements.txt`
---
## Usage
Run the agent from the folder containing the script:
`python series3_watcher.py`
The script will:
1. Scan the Blastware event folder for `.MLG` files (within a max age window).
2. Sniff each file header for the unit ID.
3. Print a status line for each detected unit (OK / Pending / Missing).
4. Optionally POST a heartbeat payload on an interval when `API_ENABLED=true`.
5. Write logs into the `agent_logs/` folder and auto-clean old logs.
---
## Config
All settings are stored in `config.ini`.
Key fields:
- `SERIES3_PATH` — folder containing `.MLG` files
- `SCAN_INTERVAL_SECONDS` — how often to scan
- `OK_HOURS` / `MISSING_HOURS` — thresholds for status
- `MLG_HEADER_BYTES` — how many bytes to sniff from each `.MLG` header
- `RECENT_WARN_DAYS` — log unsniffable files newer than this window
- `MAX_EVENT_AGE_DAYS` — ignore events older than this many days
- `API_ENABLED` — enable/disable heartbeat POST
- `API_URL` — heartbeat endpoint
- `API_INTERVAL_SECONDS` — heartbeat frequency
- `SOURCE_ID` / `SOURCE_TYPE` — identifiers included in the API payload
- `LOG_RETENTION_DAYS` — auto-delete logs older than this many days
- `COLORIZE` — ANSI color output (off by default for Win7)
---
## Logs
Logs are stored under `agent_logs/`.
Git ignores all log files but keeps the folder itself.
--- ---
## Versioning ## Versioning
This repo follows **Semantic Versioning (SemVer)**. Follows **Semantic Versioning**. Current release: **v1.4.0**.
See `CHANGELOG.md` for full history.
Current release: **v1.3.0** — renamed to series3-watcher.
See `CHANGELOG.md` for details.
--- ---
## License ## License
Private / internal project. Private / internal — Terra-Mechanics Inc.

16
build.bat Normal file
View File

@@ -0,0 +1,16 @@
@echo off
echo Building series3-watcher.exe...
pip install pyinstaller pystray Pillow
REM Check whether icon.ico exists alongside this script.
REM If it does, include it; otherwise build without a custom icon.
if exist "%~dp0icon.ico" (
pyinstaller --onefile --windowed --name series3-watcher --icon="%~dp0icon.ico" series3_tray.py
) else (
echo [INFO] icon.ico not found -- building without custom icon.
pyinstaller --onefile --windowed --name series3-watcher series3_tray.py
)
echo.
echo Done. Check dist\series3-watcher.exe
pause

37
installer.iss Normal file
View File

@@ -0,0 +1,37 @@
; Inno Setup script for Series 3 Watcher
; Run through Inno Setup Compiler after building dist\series3-watcher.exe
[Setup]
AppName=Series 3 Watcher
AppVersion=1.4.0
AppPublisher=Terra-Mechanics Inc.
DefaultDirName={pf}\Series3Watcher
DefaultGroupName=Series 3 Watcher
OutputBaseFilename=series3-watcher-setup
Compression=lzma
SolidCompression=yes
; Require admin rights so we can write to Program Files
PrivilegesRequired=admin
[Tasks]
Name: "desktopicon"; Description: "Create a &desktop icon"; GroupDescription: "Additional icons:"; Flags: unchecked
[Files]
; Main executable — built by build.bat / PyInstaller
Source: "dist\series3-watcher.exe"; DestDir: "{app}"; Flags: ignoreversion
[Icons]
; Start Menu shortcut
Name: "{group}\Series 3 Watcher"; Filename: "{app}\series3-watcher.exe"
; Start Menu uninstall shortcut
Name: "{group}\Uninstall Series 3 Watcher"; Filename: "{uninstallexe}"
; Desktop shortcut (optional — controlled by [Tasks] above)
Name: "{commondesktop}\Series 3 Watcher"; Filename: "{app}\series3-watcher.exe"; Tasks: desktopicon
; Startup folder shortcut so the tray app launches on login
Name: "{userstartup}\Series 3 Watcher"; Filename: "{app}\series3-watcher.exe"
[Run]
; Offer to launch the app after install (unchecked by default)
Filename: "{app}\series3-watcher.exe"; \
Description: "Launch Series 3 Watcher"; \
Flags: nowait postinstall skipifsilent unchecked

View File

@@ -1 +1,5 @@
# Python 3.8.10 standard library only (no external packages required). # series3_watcher.py — stdlib only, no external packages required.
# series3_tray.py — required for system tray mode:
pystray>=0.19.5
Pillow>=9.0.0

407
series3_tray.py Normal file
View File

@@ -0,0 +1,407 @@
"""
Series 3 Watcher — System Tray Launcher v1.4.0
Requires: pystray, Pillow, tkinter (stdlib)
Run with: pythonw series3_tray.py (no console window)
or: python series3_tray.py (with console, for debugging)
Put a shortcut to this in shell:startup for auto-start on login.
Python 3.8 compatible — no walrus operators, no f-string = specifier,
no match statements, no 3.9+ syntax.
"""
import os
import sys
import subprocess
import tempfile
import threading
import urllib.request
import urllib.error
from datetime import datetime
import pystray
from PIL import Image, ImageDraw
import series3_watcher as watcher
# --------------- Auto-updater ---------------
GITEA_BASE = "https://gitea.serversdown.net"
GITEA_USER = "serversdown"
GITEA_REPO = "series3-watcher"
GITEA_API_URL = "{}/api/v1/repos/{}/{}/releases?limit=1&page=1".format(
GITEA_BASE, GITEA_USER, GITEA_REPO
)
# Populated from watcher version string at startup
_CURRENT_VERSION = getattr(watcher, "VERSION", "0.0.0")
def _version_tuple(v):
"""Convert '1.4.0' -> (1, 4, 0) for comparison. Non-numeric parts -> 0."""
parts = []
for p in str(v).lstrip("v").split(".")[:3]:
try:
parts.append(int(p))
except ValueError:
parts.append(0)
while len(parts) < 3:
parts.append(0)
return tuple(parts)
def check_for_update():
"""
Query Gitea for the latest release.
Returns (tag, download_url) if an update is available, else (None, None).
"""
import json as _json
try:
req = urllib.request.Request(
GITEA_API_URL,
headers={"User-Agent": "series3-watcher/{}".format(_CURRENT_VERSION)},
)
with urllib.request.urlopen(req, timeout=8) as resp:
releases = _json.loads(resp.read().decode("utf-8"))
if not releases:
return None, None
latest = releases[0]
tag = latest.get("tag_name", "")
if _version_tuple(tag) <= _version_tuple(_CURRENT_VERSION):
return None, None
# Find the .exe asset
assets = latest.get("assets", [])
for asset in assets:
name = asset.get("name", "")
if name.lower().endswith(".exe"):
return tag, asset.get("browser_download_url")
return tag, None
except Exception:
return None, None
def apply_update(download_url):
"""
Download new .exe to a temp file, write a swap .bat, launch it, exit.
The bat waits for us to exit, then swaps the files and relaunches.
"""
exe_path = os.path.abspath(sys.executable if getattr(sys, "frozen", False) else sys.argv[0])
try:
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".exe", prefix="s3w_update_")
os.close(tmp_fd)
req = urllib.request.Request(
download_url,
headers={"User-Agent": "series3-watcher/{}".format(_CURRENT_VERSION)},
)
with urllib.request.urlopen(req, timeout=60) as resp:
with open(tmp_path, "wb") as f:
f.write(resp.read())
bat_fd, bat_path = tempfile.mkstemp(suffix=".bat", prefix="s3w_swap_")
os.close(bat_fd)
bat_content = (
"@echo off\r\n"
"ping 127.0.0.1 -n 4 > nul\r\n"
"copy /Y \"{new}\" \"{exe}\"\r\n"
"start \"\" \"{exe}\"\r\n"
"del \"%~f0\"\r\n"
).format(new=tmp_path, exe=exe_path)
with open(bat_path, "w") as f:
f.write(bat_content)
subprocess.Popen(
["cmd", "/C", bat_path],
creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, "CREATE_NO_WINDOW") else 0,
)
return True
except Exception:
return False
# --------------- Paths ---------------
HERE = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(HERE, "config.ini")
# --------------- Icon drawing ---------------
COLORS = {
"ok": (60, 200, 80), # green
"pending": (230, 180, 0), # amber
"missing": (210, 40, 40), # red
"error": (160, 40, 200), # purple
"starting": (120, 120, 120), # grey
}
ICON_SIZE = 64
def make_icon(status):
"""Draw a solid filled circle on a transparent background."""
color = COLORS.get(status, COLORS["starting"])
img = Image.new("RGBA", (ICON_SIZE, ICON_SIZE), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
margin = 6
draw.ellipse(
[margin, margin, ICON_SIZE - margin, ICON_SIZE - margin],
fill=color,
)
return img
# --------------- First-run check ---------------
def ensure_config():
"""
If config.ini is missing, launch the first-run wizard.
Returns True if config is ready, False if user cancelled.
"""
if os.path.exists(CONFIG_PATH):
return True
# Import here to avoid pulling in tkinter unless needed
from settings_dialog import show_dialog
saved = show_dialog(CONFIG_PATH, wizard=True)
if not saved:
_show_cancel_message()
return False
return True
def _show_cancel_message():
"""Show a plain messagebox telling the user the app cannot start."""
try:
import tkinter as tk
from tkinter import messagebox
root = tk.Tk()
root.withdraw()
messagebox.showwarning(
"Series 3 Watcher",
"No configuration was saved.\nThe application will now exit.",
)
root.destroy()
except Exception:
pass
# --------------- Tray app ---------------
class WatcherTray:
def __init__(self):
self.state = {}
self.stop_event = threading.Event()
self._watcher_thread = None
self._icon = None
# Lock guards _rebuild_menu calls from the updater thread
self._menu_lock = threading.Lock()
# --- Watcher thread management ---
def _start_watcher(self):
self.stop_event.clear()
self._watcher_thread = threading.Thread(
target=watcher.run_watcher,
args=(self.state, self.stop_event),
daemon=True,
name="watcher",
)
self._watcher_thread.start()
def _stop_watcher(self):
self.stop_event.set()
if self._watcher_thread is not None:
self._watcher_thread.join(timeout=10)
self._watcher_thread = None
def _restart_watcher(self):
"""Stop any running watcher and start a fresh one."""
self._stop_watcher()
self.stop_event = threading.Event()
self.state["status"] = "starting"
self.state["units"] = []
self.state["last_scan"] = None
self.state["last_error"] = None
self._start_watcher()
# --- Menu item callbacks ---
def _open_settings(self, icon, item):
"""Open the settings dialog. On save, restart watcher thread."""
from settings_dialog import show_dialog
saved = show_dialog(CONFIG_PATH, wizard=False)
if saved:
self._restart_watcher()
# Rebuild menu so status label refreshes
if self._icon is not None:
with self._menu_lock:
self._icon.menu = self._build_menu()
def _open_logs(self, icon, item):
log_dir = self.state.get("log_dir")
if not log_dir:
log_dir = HERE
if os.path.exists(log_dir):
subprocess.Popen(["explorer", log_dir])
else:
parent = os.path.dirname(log_dir)
if os.path.exists(parent):
subprocess.Popen(["explorer", parent])
else:
subprocess.Popen(["explorer", HERE])
def _exit(self, icon, item):
self.stop_event.set()
icon.stop()
# --- Dynamic menu text helpers ---
def _status_text(self):
status = self.state.get("status", "starting")
last_err = self.state.get("last_error")
last_scan = self.state.get("last_scan")
if status == "error":
return "Status: Error — {}".format(last_err or "unknown")
if status == "starting":
return "Status: Starting..."
if last_scan is not None:
age_secs = int((datetime.now() - last_scan).total_seconds())
if age_secs < 60:
age_str = "{}s ago".format(age_secs)
else:
age_str = "{}m ago".format(age_secs // 60)
unit_count = len(self.state.get("units", []))
return "Status: {} | {} unit(s) | scan {}".format(
status.upper(), unit_count, age_str
)
return "Status: {}".format(status.upper())
def _build_units_submenu(self):
units = self.state.get("units", [])
if not units:
items = [pystray.MenuItem("No units detected", None, enabled=False)]
else:
items = []
for u in units:
label = "{uid}{status} ({age:.1f}h ago)".format(
uid=u["uid"],
status=u["status"],
age=u["age_hours"],
)
items.append(pystray.MenuItem(label, None, enabled=False))
return pystray.Menu(*items)
def _build_menu(self):
# Capture current text/submenu at build time; pystray will call
# callables each render, but static strings are fine for infrequent
# menu rebuilds. We use callables for the dynamic items so that the
# text shown on hover/open is current.
status_text = self._status_text()
units_submenu = self._build_units_submenu()
return pystray.Menu(
pystray.MenuItem(status_text, None, enabled=False),
pystray.Menu.SEPARATOR,
pystray.MenuItem("Units", units_submenu),
pystray.Menu.SEPARATOR,
pystray.MenuItem("Settings...", self._open_settings),
pystray.MenuItem("Open Log Folder", self._open_logs),
pystray.Menu.SEPARATOR,
pystray.MenuItem("Exit", self._exit),
)
# --- Icon / menu update loop ---
def _icon_updater(self):
"""Periodically refresh the tray icon color and menu to match watcher state."""
last_status = None
update_check_counter = 0 # check for updates every ~5 min (30 * 10s ticks)
while not self.stop_event.is_set():
status = self.state.get("status", "starting")
if self._icon is not None:
# Always rebuild menu every cycle so unit list and scan age stay fresh
with self._menu_lock:
self._icon.menu = self._build_menu()
if status != last_status:
self._icon.icon = make_icon(status)
self._icon.title = "Series 3 Watcher — {}".format(status.upper())
last_status = status
# Check if terra-view signalled an update via heartbeat response
if self.state.get("update_available"):
self.state["update_available"] = False
self._do_update()
return # exit loop; swap bat will relaunch
# Periodic Gitea update check (every ~5 min)
update_check_counter += 1
if update_check_counter >= 30:
update_check_counter = 0
tag, url = check_for_update()
if tag and url:
self._do_update(url)
return # exit loop; swap bat will relaunch
self.stop_event.wait(timeout=10)
def _do_update(self, download_url=None):
"""Notify tray icon then apply update. If url is None, fetch it first."""
if download_url is None:
_, download_url = check_for_update()
if not download_url:
return
if self._icon is not None:
self._icon.title = "Series 3 Watcher — Updating..."
self._icon.icon = make_icon("starting")
success = apply_update(download_url)
if success:
self.stop_event.set()
if self._icon is not None:
self._icon.stop()
# If update failed, just keep running silently
# --- Entry point ---
def run(self):
self._start_watcher()
icon_img = make_icon("starting")
self._icon = pystray.Icon(
name="series3_watcher",
icon=icon_img,
title="Series 3 Watcher — Starting...",
menu=self._build_menu(),
)
updater = threading.Thread(
target=self._icon_updater, daemon=True, name="icon-updater"
)
updater.start()
self._icon.run()
# --------------- Entry point ---------------
def main():
if not ensure_config():
sys.exit(0)
app = WatcherTray()
app.run()
if __name__ == "__main__":
main()

View File

@@ -1,5 +1,5 @@
""" """
Series 3 Watcher — v1.3.0 Series 3 Watcher — v1.4.0
Environment: Environment:
- Python 3.8 (Windows 7 compatible) - Python 3.8 (Windows 7 compatible)
@@ -12,15 +12,14 @@ Key Features:
- Safe .MLG header sniff for unit IDs (BE#### / BA####) - Safe .MLG header sniff for unit IDs (BE#### / BA####)
- Standardized SFM Telemetry JSON payload (source-agnostic) - Standardized SFM Telemetry JSON payload (source-agnostic)
- Periodic HTTP heartbeat POST to SFM backend - Periodic HTTP heartbeat POST to SFM backend
- NEW in v1.2.0: - Tray-friendly: run_watcher(state, stop_event) for background thread use
- No local roster / CSV dependency
- Only scans .MLG files newer than MAX_EVENT_AGE_DAYS
""" """
import os import os
import re import re
import time import time
import json import json
import threading
import configparser import configparser
import urllib.request import urllib.request
import urllib.error import urllib.error
@@ -95,7 +94,7 @@ def log_message(path: str, enabled: bool, msg: str) -> None:
with open(path, "a", encoding="utf-8") as f: with open(path, "a", encoding="utf-8") as f:
f.write("{} {}\n".format(datetime.now(timezone.utc).isoformat(), msg)) f.write("{} {}\n".format(datetime.now(timezone.utc).isoformat(), msg))
except Exception: except Exception:
# Logging must never crash the agent # Logging must never crash the watcher
pass pass
@@ -205,7 +204,7 @@ def scan_latest(
# If unsniffable but very recent, log for later inspection # If unsniffable but very recent, log for later inspection
if (recent_cutoff is not None) and (mtime >= recent_cutoff): if (recent_cutoff is not None) and (mtime >= recent_cutoff):
if logger: if logger:
logger(f"[unsniffable-recent] {fpath}") logger("[unsniffable-recent] {}".format(fpath))
continue # skip file if no unit ID found in header continue # skip file if no unit ID found in header
cache[fpath] = (mtime, uid) cache[fpath] = (mtime, uid)
@@ -217,16 +216,25 @@ def scan_latest(
# --- API heartbeat / SFM telemetry helpers --- # --- API heartbeat / SFM telemetry helpers ---
def send_api_payload(payload: dict, api_url: str) -> None: VERSION = "1.4.0"
def send_api_payload(payload: dict, api_url: str) -> Optional[dict]:
"""POST payload to API. Returns parsed JSON response dict, or None on failure."""
if not api_url: if not api_url:
return return None
data = json.dumps(payload).encode("utf-8") data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(api_url, data=data, headers={"Content-Type": "application/json"}) req = urllib.request.Request(api_url, data=data, headers={"Content-Type": "application/json"})
try: try:
with urllib.request.urlopen(req, timeout=5) as res: with urllib.request.urlopen(req, timeout=5) as res:
print(f"[API] POST success: {res.status}") print("[API] POST success: {}".format(res.status))
try:
return json.loads(res.read().decode("utf-8"))
except Exception:
return None
except urllib.error.URLError as e: except urllib.error.URLError as e:
print(f"[API] POST failed: {e}") print("[API] POST failed: {}".format(e))
return None
def build_sfm_payload(units_dict: Dict[str, Dict[str, Any]], cfg: Dict[str, Any]) -> dict: def build_sfm_payload(units_dict: Dict[str, Dict[str, Any]], cfg: Dict[str, Any]) -> dict:
@@ -280,10 +288,38 @@ def build_sfm_payload(units_dict: Dict[str, Dict[str, Any]], cfg: Dict[str, Any]
return payload return payload
# --------------- Main loop ------------------ # --------------- Watcher loop (tray-friendly) ----------------
def main() -> None: def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None:
"""
Main watcher loop. Runs in a background thread when launched from the tray.
state dict is written on every scan cycle:
state["status"] — "ok" | "pending" | "missing" | "error" | "starting"
state["units"] — list of dicts: {uid, status, age_hours, last, fname}
state["last_scan"] — datetime of last successful scan (or None)
state["last_error"] — last error string (or None)
state["log_dir"] — directory containing the log file
state["cfg"] — loaded config dict
"""
here = os.path.dirname(__file__) or "." here = os.path.dirname(__file__) or "."
cfg = load_config(os.path.join(here, "config.ini")) config_path = os.path.join(here, "config.ini")
state["status"] = "starting"
state["units"] = []
state["last_scan"] = None
state["last_error"] = None
state["log_dir"] = None
state["cfg"] = {}
try:
cfg = load_config(config_path)
except Exception as e:
state["status"] = "error"
state["last_error"] = "Config load failed: {}".format(e)
return
state["cfg"] = cfg
state["log_dir"] = os.path.dirname(cfg["LOG_FILE"]) or here
WATCH_PATH = cfg["WATCH_PATH"] WATCH_PATH = cfg["WATCH_PATH"]
SCAN_INTERVAL = int(cfg["SCAN_INTERVAL"]) SCAN_INTERVAL = int(cfg["SCAN_INTERVAL"])
@@ -315,12 +351,10 @@ def main() -> None:
), ),
) )
# cache for scanning
sniff_cache: Dict[str, Tuple[float, str]] = {} sniff_cache: Dict[str, Tuple[float, str]] = {}
last_api_ts: float = 0.0 last_api_ts: float = 0.0
while True: while not stop_event.is_set():
try: try:
now_local = datetime.now().isoformat() now_local = datetime.now().isoformat()
now_utc = datetime.now(timezone.utc).isoformat() now_utc = datetime.now(timezone.utc).isoformat()
@@ -342,24 +376,41 @@ def main() -> None:
) )
now_epoch = time.time() now_epoch = time.time()
# Detected units summary (no roster dependency) # Build per-unit status list for the tray
unit_list = []
worst = "ok" # tracks worst status across all units
if latest: if latest:
print("\nDetected Units (within last {} days):".format(MAX_EVENT_AGE_DAYS)) print("\nDetected Units (within last {} days):".format(MAX_EVENT_AGE_DAYS))
for uid in sorted(latest.keys()): for uid in sorted(latest.keys()):
info = latest[uid] info = latest[uid]
age_hours = (now_epoch - info["mtime"]) / 3600.0 age_hours = (now_epoch - info["mtime"]) / 3600.0
if age_hours > MISSING_HOURS: if age_hours > MISSING_HOURS:
status, col = "Missing", C_MIS status, col = "missing", C_MIS
elif age_hours > OK_HOURS: elif age_hours > OK_HOURS:
status, col = "Pending", C_PEN status, col = "pending", C_PEN
else: else:
status, col = "OK", C_OK status, col = "ok", C_OK
# escalate worst status
if status == "missing":
worst = "missing"
elif status == "pending" and worst != "missing":
worst = "pending"
unit_list.append({
"uid": uid,
"status": status,
"age_hours": age_hours,
"last": fmt_last(info["mtime"]),
"fname": info["fname"],
})
line = ( line = (
"{col}{uid:<8} {status:<8} Age: {age:<7} Last: {last} (File: {fname}){rst}".format( "{col}{uid:<8} {status:<8} Age: {age:<7} Last: {last} (File: {fname}){rst}".format(
col=col, col=col,
uid=uid, uid=uid,
status=status, status=status.capitalize(),
age=fmt_age(now_epoch, info["mtime"]), age=fmt_age(now_epoch, info["mtime"]),
last=fmt_last(info["mtime"]), last=fmt_last(info["mtime"]),
fname=info["fname"], fname=info["fname"],
@@ -375,25 +426,47 @@ def main() -> None:
ENABLE_LOGGING, ENABLE_LOGGING,
"[info] no recent MLG activity within {} days".format(MAX_EVENT_AGE_DAYS), "[info] no recent MLG activity within {} days".format(MAX_EVENT_AGE_DAYS),
) )
worst = "missing"
# Update shared state for tray
state["status"] = worst
state["units"] = unit_list
state["last_scan"] = datetime.now()
state["last_error"] = None
# ---- API heartbeat to SFM ---- # ---- API heartbeat to SFM ----
if cfg.get("API_ENABLED", False): if cfg.get("API_ENABLED", False):
now_ts = time.time() now_ts = time.time()
interval = int(cfg.get("API_INTERVAL_SECONDS", 300)) interval = int(cfg.get("API_INTERVAL_SECONDS", 300))
if now_ts - last_api_ts >= interval: if now_ts - last_api_ts >= interval:
payload = build_sfm_payload(latest, cfg) hb_payload = build_sfm_payload(latest, cfg)
send_api_payload(payload, cfg.get("API_URL", "")) hb_payload["version"] = VERSION
response = send_api_payload(hb_payload, cfg.get("API_URL", ""))
last_api_ts = now_ts last_api_ts = now_ts
# Surface update signal to tray
if response and response.get("update_available"):
state["update_available"] = True
except KeyboardInterrupt:
print("\nStopping...")
break
except Exception as e: except Exception as e:
err = "[loop-error] {}".format(e) err = "[loop-error] {}".format(e)
print(err) print(err)
log_message(LOG_FILE, ENABLE_LOGGING, err) log_message(LOG_FILE, ENABLE_LOGGING, err)
state["status"] = "error"
state["last_error"] = str(e)
time.sleep(SCAN_INTERVAL) # Interruptible sleep: wake immediately if stop_event fires
stop_event.wait(timeout=SCAN_INTERVAL)
# --------------- Main (standalone) ------------------
def main() -> None:
state = {}
stop_event = threading.Event()
try:
run_watcher(state, stop_event)
except KeyboardInterrupt:
print("\nStopping...")
stop_event.set()
if __name__ == "__main__": if __name__ == "__main__":

459
settings_dialog.py Normal file
View File

@@ -0,0 +1,459 @@
"""
Series 3 Watcher — Settings Dialog v1.4.0
Provides a Tkinter settings dialog that doubles as a first-run wizard.
Public API:
show_dialog(config_path, wizard=False) -> bool
Returns True if the user saved, False if they cancelled.
Python 3.8 compatible — no walrus operators, no f-string = specifier,
no match statements, no 3.9+ syntax.
No external dependencies beyond stdlib + tkinter.
"""
import os
import configparser
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
from socket import gethostname
# --------------- Defaults (mirror config-template.ini) ---------------
DEFAULTS = {
"API_ENABLED": "true",
"API_URL": "",
"API_INTERVAL_SECONDS": "300",
"SOURCE_ID": "", # empty = use hostname at runtime
"SOURCE_TYPE": "series3_watcher",
"SERIES3_PATH": r"C:\Blastware 10\Event\autocall home",
"MAX_EVENT_AGE_DAYS": "365",
"SCAN_INTERVAL_SECONDS":"300",
"OK_HOURS": "12",
"MISSING_HOURS": "24",
"MLG_HEADER_BYTES": "2048",
"ENABLE_LOGGING": "true",
"LOG_FILE": r"C:\SeismoEmitter\agent_logs\series3_watcher.log",
"LOG_RETENTION_DAYS": "30",
"COLORIZE": "false",
}
# --------------- Config I/O ---------------
def _load_config(config_path):
"""
Load existing config.ini. Returns a flat dict of string values.
Falls back to DEFAULTS for any missing key.
"""
values = dict(DEFAULTS)
if not os.path.exists(config_path):
return values
cp = configparser.ConfigParser(inline_comment_prefixes=(";", "#"))
cp.optionxform = str
try:
cp.read(config_path, encoding="utf-8")
except Exception:
return values
# Accept either [agent] section or a bare file
section = None
if cp.has_section("agent"):
section = "agent"
elif cp.sections():
section = cp.sections()[0]
if section:
for k in DEFAULTS:
if cp.has_option(section, k):
values[k] = cp.get(section, k).strip()
return values
def _save_config(config_path, values):
"""Write all values to config_path under [agent] section."""
cp = configparser.ConfigParser()
cp.optionxform = str
cp["agent"] = {}
for k, v in values.items():
cp["agent"][k] = v
config_dir = os.path.dirname(config_path)
if config_dir and not os.path.exists(config_dir):
os.makedirs(config_dir)
with open(config_path, "w", encoding="utf-8") as f:
cp.write(f)
# --------------- Spinbox helper ---------------
def _make_spinbox(parent, from_, to, width=8):
"""Create a ttk.Spinbox; fall back to tk.Spinbox on older ttk."""
try:
sb = ttk.Spinbox(parent, from_=from_, to=to, width=width)
except AttributeError:
# ttk.Spinbox added in Python 3.7 but not available everywhere
sb = tk.Spinbox(parent, from_=from_, to=to, width=width)
return sb
# --------------- Field helpers ---------------
def _add_label_entry(frame, row, label_text, var, hint=None, readonly=False):
"""Add a label + entry row to a grid frame. Returns the Entry widget."""
tk.Label(frame, text=label_text, anchor="w").grid(
row=row, column=0, sticky="w", padx=(8, 4), pady=4
)
state = "readonly" if readonly else "normal"
entry = ttk.Entry(frame, textvariable=var, width=42, state=state)
entry.grid(row=row, column=1, sticky="ew", padx=(0, 8), pady=4)
if hint and not var.get():
# Show placeholder hint in grey; clear on focus
entry.config(foreground="grey")
entry.insert(0, hint)
def _on_focus_in(event, e=entry, h=hint, v=var):
if e.get() == h:
e.delete(0, tk.END)
e.config(foreground="black")
def _on_focus_out(event, e=entry, h=hint, v=var):
if not e.get():
e.config(foreground="grey")
e.insert(0, h)
v.set("")
entry.bind("<FocusIn>", _on_focus_in)
entry.bind("<FocusOut>", _on_focus_out)
return entry
def _add_label_spinbox(frame, row, label_text, var, from_, to):
"""Add a label + spinbox row to a grid frame. Returns the Spinbox widget."""
tk.Label(frame, text=label_text, anchor="w").grid(
row=row, column=0, sticky="w", padx=(8, 4), pady=4
)
sb = _make_spinbox(frame, from_=from_, to=to, width=8)
sb.grid(row=row, column=1, sticky="w", padx=(0, 8), pady=4)
sb.delete(0, tk.END)
sb.insert(0, var.get())
def _on_change(*args):
var.set(sb.get())
sb.config(command=_on_change)
sb.bind("<KeyRelease>", _on_change)
return sb
def _add_label_check(frame, row, label_text, var):
"""Add a checkbox row to a grid frame."""
cb = ttk.Checkbutton(frame, text=label_text, variable=var)
cb.grid(row=row, column=0, columnspan=2, sticky="w", padx=(8, 8), pady=4)
return cb
def _add_label_browse_entry(frame, row, label_text, var, browse_fn):
"""Add a label + entry + Browse button row."""
tk.Label(frame, text=label_text, anchor="w").grid(
row=row, column=0, sticky="w", padx=(8, 4), pady=4
)
inner = tk.Frame(frame)
inner.grid(row=row, column=1, sticky="ew", padx=(0, 8), pady=4)
inner.columnconfigure(0, weight=1)
entry = ttk.Entry(inner, textvariable=var, width=36)
entry.grid(row=0, column=0, sticky="ew")
btn = ttk.Button(inner, text="Browse...", command=browse_fn, width=9)
btn.grid(row=0, column=1, padx=(4, 0))
return entry
# --------------- Main dialog class ---------------
class SettingsDialog:
def __init__(self, parent, config_path, wizard=False):
self.config_path = config_path
self.wizard = wizard
self.saved = False
self.root = parent
if wizard:
self.root.title("Series 3 Watcher — Setup")
else:
self.root.title("Series 3 Watcher — Settings")
self.root.resizable(False, False)
# Center on screen
self.root.update_idletasks()
self._values = _load_config(config_path)
self._build_vars()
self._build_ui()
# Make dialog modal
self.root.grab_set()
self.root.protocol("WM_DELETE_WINDOW", self._on_cancel)
# --- Variable setup ---
def _build_vars(self):
v = self._values
# Connection
self.var_api_enabled = tk.BooleanVar(value=v["API_ENABLED"].lower() in ("1","true","yes","on"))
self.var_api_url = tk.StringVar(value=v["API_URL"])
self.var_api_interval = tk.StringVar(value=v["API_INTERVAL_SECONDS"])
self.var_source_id = tk.StringVar(value=v["SOURCE_ID"])
self.var_source_type = tk.StringVar(value=v["SOURCE_TYPE"])
# Paths
self.var_series3_path = tk.StringVar(value=v["SERIES3_PATH"])
self.var_max_event_age_days = tk.StringVar(value=v["MAX_EVENT_AGE_DAYS"])
self.var_log_file = tk.StringVar(value=v["LOG_FILE"])
# Scanning
self.var_scan_interval = tk.StringVar(value=v["SCAN_INTERVAL_SECONDS"])
self.var_ok_hours = tk.StringVar(value=v["OK_HOURS"])
self.var_missing_hours = tk.StringVar(value=v["MISSING_HOURS"])
self.var_mlg_header_bytes = tk.StringVar(value=v["MLG_HEADER_BYTES"])
# Logging
self.var_enable_logging = tk.BooleanVar(value=v["ENABLE_LOGGING"].lower() in ("1","true","yes","on"))
self.var_log_retention_days = tk.StringVar(value=v["LOG_RETENTION_DAYS"])
self.var_colorize = tk.BooleanVar(value=v["COLORIZE"].lower() in ("1","true","yes","on"))
# --- UI construction ---
def _build_ui(self):
outer = tk.Frame(self.root, padx=10, pady=8)
outer.pack(fill="both", expand=True)
if self.wizard:
welcome = (
"Welcome to Series 3 Watcher!\n\n"
"No configuration file was found. Please review the settings below\n"
"and click \"Save & Start\" when you are ready."
)
lbl = tk.Label(
outer, text=welcome, justify="left",
wraplength=460, fg="#1a5276", font=("TkDefaultFont", 9, "bold"),
)
lbl.pack(fill="x", pady=(0, 8))
# Notebook
nb = ttk.Notebook(outer)
nb.pack(fill="both", expand=True)
self._build_tab_connection(nb)
self._build_tab_paths(nb)
self._build_tab_scanning(nb)
self._build_tab_logging(nb)
# Buttons
btn_frame = tk.Frame(outer)
btn_frame.pack(fill="x", pady=(10, 0))
save_label = "Save & Start" if self.wizard else "Save"
btn_save = ttk.Button(btn_frame, text=save_label, command=self._on_save, width=14)
btn_save.pack(side="right", padx=(4, 0))
btn_cancel = ttk.Button(btn_frame, text="Cancel", command=self._on_cancel, width=10)
btn_cancel.pack(side="right")
def _tab_frame(self, nb, title):
"""Create a new tab in nb, return a scrollable inner frame."""
outer = tk.Frame(nb, padx=4, pady=4)
nb.add(outer, text=title)
outer.columnconfigure(1, weight=1)
return outer
def _build_tab_connection(self, nb):
f = self._tab_frame(nb, "Connection")
_add_label_check(f, 0, "API Enabled", self.var_api_enabled)
_add_label_entry(
f, 1, "terra-view URL", self.var_api_url,
hint="http://192.168.x.x:8000/api/heartbeat",
)
_add_label_spinbox(f, 2, "API Interval (sec)", self.var_api_interval, 30, 3600)
source_id_hint = "Defaults to hostname ({})".format(gethostname())
_add_label_entry(f, 3, "Source ID", self.var_source_id, hint=source_id_hint)
_add_label_entry(f, 4, "Source Type", self.var_source_type, readonly=True)
def _build_tab_paths(self, nb):
f = self._tab_frame(nb, "Paths")
def browse_series3():
d = filedialog.askdirectory(
title="Select Blastware Event Folder",
initialdir=self.var_series3_path.get() or "C:\\",
)
if d:
self.var_series3_path.set(d.replace("/", "\\"))
_add_label_browse_entry(f, 0, "Series3 Path", self.var_series3_path, browse_series3)
_add_label_spinbox(f, 1, "Max Event Age (days)", self.var_max_event_age_days, 1, 3650)
def browse_log():
p = filedialog.asksaveasfilename(
title="Select Log File",
defaultextension=".log",
filetypes=[("Log files", "*.log"), ("Text files", "*.txt"), ("All files", "*.*")],
initialfile=os.path.basename(self.var_log_file.get() or "series3_watcher.log"),
initialdir=os.path.dirname(self.var_log_file.get() or "C:\\"),
)
if p:
self.var_log_file.set(p.replace("/", "\\"))
_add_label_browse_entry(f, 2, "Log File", self.var_log_file, browse_log)
def _build_tab_scanning(self, nb):
f = self._tab_frame(nb, "Scanning")
_add_label_spinbox(f, 0, "Scan Interval (sec)", self.var_scan_interval, 10, 3600)
_add_label_spinbox(f, 1, "OK Hours", self.var_ok_hours, 1, 168)
_add_label_spinbox(f, 2, "Missing Hours", self.var_missing_hours, 1, 168)
_add_label_spinbox(f, 3, "MLG Header Bytes", self.var_mlg_header_bytes, 256, 65536)
def _build_tab_logging(self, nb):
f = self._tab_frame(nb, "Logging")
_add_label_check(f, 0, "Enable Logging", self.var_enable_logging)
_add_label_spinbox(f, 1, "Log Retention (days)", self.var_log_retention_days, 1, 365)
_add_label_check(f, 2, "Colorize console output", self.var_colorize)
# --- Validation helpers ---
def _get_int_var(self, var, name, min_val, max_val, default):
"""Parse a StringVar as int, clamp to range, return clamped value or None on error."""
raw = var.get().strip()
try:
val = int(raw)
except ValueError:
messagebox.showerror(
"Validation Error",
"{} must be an integer (got: {!r}).".format(name, raw),
)
return None
if val < min_val or val > max_val:
messagebox.showerror(
"Validation Error",
"{} must be between {} and {} (got {}).".format(name, min_val, max_val, val),
)
return None
return val
# --- Save / Cancel ---
def _on_save(self):
# Validate numeric fields before writing
checks = [
(self.var_api_interval, "API Interval", 30, 3600, 300),
(self.var_max_event_age_days, "Max Event Age Days", 1, 3650, 365),
(self.var_scan_interval, "Scan Interval", 10, 3600, 300),
(self.var_ok_hours, "OK Hours", 1, 168, 12),
(self.var_missing_hours, "Missing Hours", 1, 168, 24),
(self.var_mlg_header_bytes, "MLG Header Bytes", 256, 65536, 2048),
(self.var_log_retention_days, "Log Retention Days", 1, 365, 30),
]
int_values = {}
for var, name, mn, mx, dflt in checks:
result = self._get_int_var(var, name, mn, mx, dflt)
if result is None:
return # validation failed; keep dialog open
int_values[name] = result
# Resolve source_id placeholder
source_id = self.var_source_id.get().strip()
# Strip placeholder hint if user left it
if source_id.startswith("Defaults to hostname"):
source_id = ""
# Resolve api_url placeholder
api_url = self.var_api_url.get().strip()
if api_url.startswith("http://192.168"):
# This is the hint — only keep it if it looks like a real URL
pass # leave as-is; user may have typed it intentionally
if api_url == "http://192.168.x.x:8000/api/heartbeat":
api_url = ""
values = {
"API_ENABLED": "true" if self.var_api_enabled.get() else "false",
"API_URL": api_url,
"API_INTERVAL_SECONDS": str(int_values["API Interval"]),
"SOURCE_ID": source_id,
"SOURCE_TYPE": self.var_source_type.get().strip() or "series3_watcher",
"SERIES3_PATH": self.var_series3_path.get().strip(),
"MAX_EVENT_AGE_DAYS": str(int_values["Max Event Age Days"]),
"SCAN_INTERVAL_SECONDS":str(int_values["Scan Interval"]),
"OK_HOURS": str(int_values["OK Hours"]),
"MISSING_HOURS": str(int_values["Missing Hours"]),
"MLG_HEADER_BYTES": str(int_values["MLG Header Bytes"]),
"ENABLE_LOGGING": "true" if self.var_enable_logging.get() else "false",
"LOG_FILE": self.var_log_file.get().strip(),
"LOG_RETENTION_DAYS": str(int_values["Log Retention Days"]),
"COLORIZE": "true" if self.var_colorize.get() else "false",
}
try:
_save_config(self.config_path, values)
except Exception as e:
messagebox.showerror("Save Error", "Could not write config.ini:\n{}".format(e))
return
self.saved = True
self.root.destroy()
def _on_cancel(self):
self.saved = False
self.root.destroy()
# --------------- Public API ---------------
def show_dialog(config_path, wizard=False):
"""
Open the settings dialog.
Parameters
----------
config_path : str
Absolute path to config.ini (read if it exists, written on Save).
wizard : bool
If True, shows first-run welcome message and "Save & Start" button.
Returns
-------
bool
True if the user saved, False if they cancelled.
"""
root = tk.Tk()
root.withdraw() # hide blank root window
# Create a Toplevel that acts as the dialog window
top = tk.Toplevel(root)
top.deiconify()
dlg = SettingsDialog(top, config_path, wizard=wizard)
# Center after build
top.update_idletasks()
w = top.winfo_reqwidth()
h = top.winfo_reqheight()
sw = top.winfo_screenwidth()
sh = top.winfo_screenheight()
x = (sw - w) // 2
y = (sh - h) // 2
top.geometry("{}x{}+{}+{}".format(w, h, x, y))
root.wait_window(top)
root.destroy()
return dlg.saved