Initial commit: Shelly Manager with Textual CLI, Streamlit UI, and comprehensive .gitignore

Shelly device management app with mDNS/subnet discovery, inventory,
configuration, and mass operations for Gen1/Gen2+ devices.

Includes .gitignore excluding runtime data (device DB, user config),
AI conversation history, build artifacts, and common Python/OS patterns.
This commit is contained in:
2026-03-23 21:51:59 +01:00
commit 71803418e5
152 changed files with 23405 additions and 0 deletions
@@ -0,0 +1 @@
"""Textual screens."""
@@ -0,0 +1,89 @@
"""Main device table."""
from __future__ import annotations
from textual import work
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Vertical
from textual.screen import Screen
from textual.widgets import DataTable, Footer, Header, Static
from shelly_manager.cli.screens.device_detail import DeviceDetailScreen
from shelly_manager.cli.widgets.filter_bar import FilterBar
from shelly_manager.core.device_manager import DeviceManager
from shelly_manager.core.model_names import format_model_plain
from shelly_manager.core.models import ShellyDevice
class DashboardScreen(Screen[None]):
"""Device list with filters."""
BINDINGS = [
Binding("r", "refresh", "Refresh"),
Binding("d", "detail", "Detail"),
Binding("f", "filter_reload", "apply"),
Binding("q", "quit", "Quit"),
Binding("escape", "quit", "Quit"),
]
def __init__(self, dm: DeviceManager) -> None:
super().__init__()
self.dm = dm
self._devices: list[ShellyDevice] = []
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
with Vertical():
yield Static("[bold]Shelly Manager[/] — [dim]r refresh · d detail · f reload filters · q quit[/]")
yield FilterBar(id="filters")
yield DataTable(
id="table",
cursor_type="row",
zebra_stripes=True,
)
yield Footer()
def on_mount(self) -> None:
table = self.query_one("#table", DataTable)
table.add_columns("Name", "IP", "URL", "Gen", "Model", "FW", "Online", "Caps")
self.load_table()
@work(exclusive=True)
async def load_table(self) -> None:
filt = self.query_one(FilterBar).build_filter()
devices = await self.dm.storage.list_devices(filt)
self._devices = devices
table = self.query_one("#table", DataTable)
table.clear(columns=True)
table.add_columns("Name", "IP", "URL", "Gen", "Model", "FW", "Online", "Caps")
for d in devices:
caps = ", ".join(d.capabilities[:3])
table.add_row(
d.display_name,
d.ip,
d.http_url,
str(d.generation),
format_model_plain(d.model),
d.firmware[:16] if d.firmware else "",
"yes" if d.online else "no",
caps,
key=d.id,
)
def action_refresh(self) -> None:
self.load_table()
def action_filter_reload(self) -> None:
self.load_table()
def action_detail(self) -> None:
table = self.query_one("#table", DataTable)
row = table.cursor_row
if row is None or row < 0 or row >= len(self._devices):
return
dev = self._devices[row]
self.app.push_screen(DeviceDetailScreen(self.dm, dev))
def action_quit(self) -> None:
self.app.exit()
@@ -0,0 +1,84 @@
"""Single-device extended view."""
from __future__ import annotations
import json
from rich.console import Group
from rich.syntax import Syntax
from rich.text import Text
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Container
from textual.screen import Screen
from textual.widgets import Footer, Header, Static
from shelly_manager.core.device_manager import DeviceManager
from shelly_manager.core.model_names import format_model_plain
from shelly_manager.core.models import ShellyDevice
class DeviceDetailScreen(Screen[None]):
BINDINGS = [
Binding("escape", "back", "Back"),
Binding("r", "refresh", "Refresh"),
]
def __init__(self, dm: DeviceManager, device: ShellyDevice) -> None:
super().__init__()
self.dm = dm
self.device = device
def compose(self) -> ComposeResult:
yield Header()
with Container(id="detail_wrap"):
yield Static(id="detail_text")
yield Footer()
def on_mount(self) -> None:
self._render()
def action_refresh(self) -> None:
self.run_worker(self._do_refresh, exclusive=True)
async def _do_refresh(self) -> None:
fresh = await self.dm.refresh_device(self.device.id)
if fresh:
self.device = fresh
self._render()
def _render(self) -> None:
d = self.device
# Use Rich renderables (Group / Text / Syntax), not Textual markup strings: URLs like
# ``http://...`` break ``[link=...]`` parsing (``:`` after ``http``), and JSON often
# contains ``[`` / ``]`` which Textual would treat as markup.
header = Text()
header.append(d.display_name, style="bold")
header.append(
f" {d.ip} Gen{d.generation} {format_model_plain(d.model)}\nWeb UI: ",
)
header.append(d.http_url, style=f"link {d.http_url}")
header.append("\n")
tip = Text.from_markup(
"[dim]Tip: run `shelly-manager-ui` → Device page for form-based config edit + save.[/dim]\n\n"
)
settings_json = json.dumps(d.settings, indent=2)[:12000]
status_json = json.dumps(d.status, indent=2)[:12000]
settings_block = Syntax(settings_json, "json", word_wrap=True, line_numbers=False)
status_block = Syntax(status_json, "json", word_wrap=True, line_numbers=False)
body: Group = Group(
header,
tip,
Text("Settings\n", style="bold"),
settings_block,
Text("\n"),
Text("Status\n", style="bold"),
status_block,
)
self.query_one("#detail_text", Static).update(body)
def action_back(self) -> None:
self.dismiss()
@@ -0,0 +1,43 @@
"""Discovery progress log."""
from __future__ import annotations
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Vertical
from textual.screen import Screen
from textual.widgets import Footer, Header, Log
from shelly_manager.core.device_manager import DeviceManager
class DiscoveryScreen(Screen[None]):
BINDINGS = [Binding("escape", "back", "Back")]
def __init__(self, dm: DeviceManager) -> None:
super().__init__()
self.dm = dm
def compose(self) -> ComposeResult:
yield Header()
with Vertical():
yield Log(id="log", highlight=True)
yield Footer()
def on_mount(self) -> None:
self.run_worker(self._run_discovery, exclusive=True)
async def _run_discovery(self) -> None:
log = self.query_one("#log", Log)
def on_progress(phase: str, msg: str) -> None:
log.write_line(f"[{phase}] {msg}")
import aiohttp
async with aiohttp.ClientSession() as session:
await self.dm.discover_all(session=session, on_progress=on_progress)
log.write_line("[done] Discovery finished.")
def action_back(self) -> None:
self.dismiss()
@@ -0,0 +1,44 @@
"""Mass tag assignment via filters."""
from __future__ import annotations
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Vertical
from textual.screen import Screen
from textual.widgets import Button, Footer, Header, Input, Static
from shelly_manager.cli.widgets.filter_bar import FilterBar
from shelly_manager.core.device_manager import DeviceManager
class MassConfigScreen(Screen[None]):
BINDINGS = [Binding("escape", "back", "Back")]
def __init__(self, dm: DeviceManager) -> None:
super().__init__()
self.dm = dm
def compose(self) -> ComposeResult:
yield Header()
with Vertical():
yield Static("[bold]Mass tags[/] — apply comma-separated tags to filtered devices.")
yield FilterBar(id="filters")
yield Input(placeholder="tags e.g. kitchen, floor-1", id="tags")
yield Button("Apply", variant="primary", id="apply")
yield Static(id="result")
yield Footer()
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "apply":
self.run_worker(self._apply_tags, exclusive=True)
async def _apply_tags(self) -> None:
filt = self.query_one(FilterBar).build_filter()
raw = self.query_one("#tags", Input).value or ""
tags = [t.strip() for t in raw.split(",") if t.strip()]
n = await self.dm.apply_mass_tags(filt, tags)
self.query_one("#result", Static).update(f"Updated {n} device(s).")
def action_back(self) -> None:
self.dismiss()