ettore.dreucci/lagomareGates
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add support for Shelly Cloud API

Browse Source
This commit is contained in:
Ettore
2026-05-14 19:22:07 +02:00
parent adcc2b9522
commit 46ba26a86d
7 changed files with 272 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/core/database.py
View File

@@ -2,7 +2,7 @@ import os
from datetime import datetime
from typing import Optional
from sqlalchemy import Boolean, Double, String, Text, create_engine
from sqlalchemy import Boolean, Double, String, Text, create_engine, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
from core.config import DATA_DIR, DATABASE_URL
@@ -21,7 +21,9 @@ class GateDB(Base):
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String, nullable=False)
gate_type: Mapped[str] = mapped_column(String, nullable=False) # 'car' | 'pedestrian'
avconnect_macro_id: Mapped[str] = mapped_column(String, nullable=False) # AVConnect macro ID
api_provider: Mapped[str] = mapped_column(String, nullable=False, default="avconnect") # 'avconnect' | 'shelly'
avconnect_macro_id: Mapped[Optional[str]] = mapped_column(String, nullable=True) # AVConnect macro ID
shelly_device_id: Mapped[Optional[str]] = mapped_column(String, nullable=True) # Shelly Cloud device ID
status: Mapped[str] = mapped_column(String, default="enabled") # 'enabled' | 'disabled'
group_name: Mapped[Optional[str]] = mapped_column(String, nullable=True) # display group label
lat: Mapped[Optional[float]] = mapped_column(Double, nullable=True) # WGS-84 latitude
@@ -38,6 +40,14 @@ class ApiCredential(Base):
mock_avconnect: Mapped[bool] = mapped_column(Boolean, default=False)
class ShellyCredential(Base):
__tablename__ = "shelly_credentials"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
server_uri: Mapped[str] = mapped_column(String, nullable=False) # e.g. https://shelly-3.eu.shelly.cloud
auth_key_enc: Mapped[str] = mapped_column(String, nullable=False) # Fernet-encrypted auth key
class Keypass(Base):
__tablename__ = "keypasses"

20
src/core/schemas.py
View File

@@ -81,7 +81,9 @@ class GateResponse(BaseModel):
id: int
name: str
gate_type: str
avconnect_macro_id: str
api_provider: str
avconnect_macro_id: Optional[str] = None
shelly_device_id: Optional[str] = None
status: str
group_name: Optional[str] = None
lat: Optional[float] = None
@@ -102,14 +104,16 @@ class GatePublicResponse(BaseModel):
class GateCreate(BaseModel):
name: str
gate_type: str # 'car' | 'pedestrian'
avconnect_macro_id: str
api_provider: str = "avconnect" # 'avconnect' | 'shelly'
avconnect_macro_id: Optional[str] = None
shelly_device_id: Optional[str] = None
status: str = "enabled"
group_name: Optional[str] = None
lat: Optional[float] = None
lon: Optional[float] = None
# ── AVConnect Credentials ─────────────────────────────────────────────────────
# ── API Credentials ──────────────────────────────────────────────────────────
class CredentialRead(BaseModel):
id: int
@@ -121,6 +125,16 @@ class CredentialUpsert(BaseModel):
password: str
class ShellyCredentialRead(BaseModel):
id: int
server_uri: str
class ShellyCredentialUpsert(BaseModel):
server_uri: str
auth_key: str
# ── Admin users ───────────────────────────────────────────────────────────────
class AdminUserResponse(BaseModel):

52
src/routers/credentials.py
View File

@@ -5,14 +5,17 @@ from pydantic import BaseModel
from sqlalchemy.orm import Session
from core.auth import encrypt_secret
from core.database import ApiCredential, get_db
from core.database import ApiCredential, ShellyCredential, get_db
from core.dependencies import require_admin
from core.schemas import CredentialRead, CredentialUpsert
from core.schemas import CredentialRead, CredentialUpsert, ShellyCredentialRead, ShellyCredentialUpsert
from services.avconnect import AVConnectAPI
from services.shelly import ShellyCloudAPI
router = APIRouter(prefix="/api/admin/credentials", tags=["admin-credentials"])
# ── AVConnect credentials ─────────────────────────────────────────────────────
@router.get("", response_model=list[CredentialRead])
async def list_credentials(
db: Session = Depends(get_db), _: dict = Depends(require_admin)
@@ -37,7 +40,7 @@ async def upsert_credential(
if cred:
cred.username = req.username
cred.password_enc = encrypt_secret(req.password)
cred.session_id = session_id # reuse the session obtained during validation
cred.session_id = session_id
else:
cred = ApiCredential(
username=req.username,
@@ -50,7 +53,47 @@ async def upsert_credential(
return CredentialRead(id=cred.id, username=cred.username)
# ── Mock AVConnect setting ─────────────────────────────────────────────────────
# ── Shelly Cloud credentials ──────────────────────────────────────────────────
@router.get("/shelly", response_model=Optional[ShellyCredentialRead])
async def get_shelly_credential(
db: Session = Depends(get_db), _: dict = Depends(require_admin)
):
cred: Optional[ShellyCredential] = db.query(ShellyCredential).first()
if not cred:
return None
return ShellyCredentialRead(id=cred.id, server_uri=cred.server_uri)
@router.put("/shelly", response_model=ShellyCredentialRead)
async def upsert_shelly_credential(
req: ShellyCredentialUpsert,
db: Session = Depends(get_db),
_: dict = Depends(require_admin),
):
try:
ok = ShellyCloudAPI(req.server_uri, req.auth_key).validate_credentials()
except Exception as exc:
raise HTTPException(502, f"Could not reach Shelly Cloud: {exc}")
if not ok:
raise HTTPException(422, "Shelly Cloud rejected these credentials")
cred: Optional[ShellyCredential] = db.query(ShellyCredential).first()
if cred:
cred.server_uri = req.server_uri
cred.auth_key_enc = encrypt_secret(req.auth_key)
else:
cred = ShellyCredential(
server_uri=req.server_uri,
auth_key_enc=encrypt_secret(req.auth_key),
)
db.add(cred)
db.commit()
db.refresh(cred)
return ShellyCredentialRead(id=cred.id, server_uri=cred.server_uri)
# ── Mock mode setting ─────────────────────────────────────────────────────────
class MockSettingResponse(BaseModel):
enabled: bool
@@ -76,7 +119,6 @@ async def set_mock_setting(
):
cred: Optional[ApiCredential] = db.query(ApiCredential).first()
if not cred:
from fastapi import HTTPException
raise HTTPException(503, "AVConnect credentials not configured")
cred.mock_avconnect = req.enabled
db.commit()

96
src/routers/gates.py
View File

@@ -8,7 +8,7 @@ from sqlalchemy.orm import Session
from core.auth import decrypt_secret
from core.config import utcnow
from core.database import ApiCredential, GateAccessLog, GateDB, Keypass, TelegramConfig, get_db
from core.database import ApiCredential, GateAccessLog, GateDB, Keypass, ShellyCredential, TelegramConfig, get_db
from core.dependencies import require_admin, require_manager, require_keypass
from core.schemas import GateCreate, GatePublicResponse, GateResponse
from services.gates import call_open_gate
@@ -35,6 +35,49 @@ def _notify(db: Session, gate_name: str, opened_by: str, ip: str | None) -> None
).start()
def _validate_gate_create(req: GateCreate) -> None:
if req.gate_type not in ("car", "pedestrian"):
raise HTTPException(400, "gate_type must be 'car' or 'pedestrian'")
if req.api_provider not in ("avconnect", "shelly"):
raise HTTPException(400, "api_provider must be 'avconnect' or 'shelly'")
if req.api_provider == "avconnect" and not req.avconnect_macro_id:
raise HTTPException(400, "avconnect_macro_id is required for AVConnect gates")
if req.api_provider == "shelly" and not req.shelly_device_id:
raise HTTPException(400, "shelly_device_id is required for Shelly gates")
def _do_open_gate(
gate_db: GateDB,
db: Session,
mock: bool,
) -> tuple[bool, Optional[str], Optional[str]]:
"""Dispatch the gate-open call to the appropriate API provider."""
if gate_db.api_provider == "shelly":
shelly_cred: Optional[ShellyCredential] = db.query(ShellyCredential).first()
if not shelly_cred:
raise HTTPException(503, "Shelly credentials not configured")
return call_open_gate(
api_provider="shelly",
shelly_device_id=gate_db.shelly_device_id,
shelly_server_uri=shelly_cred.server_uri,
shelly_auth_key=decrypt_secret(shelly_cred.auth_key_enc),
mock=mock,
)
# AVConnect (default)
av_cred: Optional[ApiCredential] = db.query(ApiCredential).first()
if not av_cred:
raise HTTPException(503, "AVConnect credentials not configured")
return call_open_gate(
api_provider="avconnect",
avconnect_macro_id=gate_db.avconnect_macro_id,
avconnect_username=av_cred.username,
avconnect_password=decrypt_secret(av_cred.password_enc),
avconnect_session_id=av_cred.session_id,
mock=mock,
)
# ── Admin: gate CRUD ──────────────────────────────────────────────────────────
@router.get("/api/admin/gates", response_model=list[GateResponse])
@@ -50,8 +93,7 @@ async def create_gate(
db: Session = Depends(get_db),
_: dict = Depends(require_admin),
):
if req.gate_type not in ("car", "pedestrian"):
raise HTTPException(400, "gate_type must be 'car' or 'pedestrian'")
_validate_gate_create(req)
gate = GateDB(**req.model_dump())
db.add(gate)
db.commit()
@@ -66,6 +108,7 @@ async def update_gate(
db: Session = Depends(get_db),
_: dict = Depends(require_admin),
):
_validate_gate_create(req)
gate: Optional[GateDB] = db.query(GateDB).filter(GateDB.id == gate_id).first()
if not gate:
raise HTTPException(404, "Gate not found")
@@ -102,21 +145,14 @@ async def admin_open_gate(
if gate_db.status != "enabled":
raise HTTPException(409, "Gate is disabled")
cred_db: Optional[ApiCredential] = db.query(ApiCredential).first()
if not cred_db:
raise HTTPException(503, "AVConnect credentials not configured")
# Determine mock mode from AVConnect credential record (applies to all providers)
av_cred: Optional[ApiCredential] = db.query(ApiCredential).first()
mock = bool(av_cred.mock_avconnect) if av_cred else False
ip = request.client.host if request.client else None
ua = request.headers.get("User-Agent")
mock = bool(cred_db.mock_avconnect)
success, error_msg, new_sid = call_open_gate(
gate_db.avconnect_macro_id,
cred_db.username,
decrypt_secret(cred_db.password_enc),
cred_db.session_id,
mock=mock,
)
success, error_msg, new_sid = _do_open_gate(gate_db, db, mock)
db.add(GateAccessLog(
timestamp=utcnow(),
@@ -130,8 +166,8 @@ async def admin_open_gate(
error=error_msg,
))
if new_sid and new_sid != cred_db.session_id:
cred_db.session_id = new_sid
if new_sid and av_cred and new_sid != av_cred.session_id:
av_cred.session_id = new_sid
db.commit()
if not success:
@@ -169,25 +205,17 @@ async def open_gate(
if not gate_db:
raise HTTPException(404, "Gate not found or disabled")
cred_db: Optional[ApiCredential] = db.query(ApiCredential).first()
if not cred_db:
raise HTTPException(503, "AVConnect credentials not configured")
ip = request.client.host if request.client else None
ua = request.headers.get("User-Agent")
allowed = json.loads(_kp.allowed_gates) if _kp.allowed_gates else None
if allowed is not None and gate_id not in allowed:
raise HTTPException(403, "This keypass does not have access to this gate")
mock = bool(cred_db.mock_avconnect)
success, error_msg, new_sid = call_open_gate(
gate_db.avconnect_macro_id,
cred_db.username,
decrypt_secret(cred_db.password_enc),
cred_db.session_id,
mock=mock,
)
av_cred: Optional[ApiCredential] = db.query(ApiCredential).first()
mock = bool(av_cred.mock_avconnect) if av_cred else False
ip = request.client.host if request.client else None
ua = request.headers.get("User-Agent")
success, error_msg, new_sid = _do_open_gate(gate_db, db, mock)
db.add(GateAccessLog(
timestamp=utcnow(),
@@ -201,8 +229,8 @@ async def open_gate(
error=error_msg,
))
if new_sid and new_sid != cred_db.session_id:
cred_db.session_id = new_sid
if new_sid and av_cred and new_sid != av_cred.session_id:
av_cred.session_id = new_sid
db.commit()
if not success:
@@ -210,5 +238,5 @@ async def open_gate(
raise HTTPException(502, error_msg or "Gate operation failed")
logger.info("Gate opened by keypass: gate_id=%d gate=%r keypass=%r ip=%r", gate_db.id, gate_db.name, f"{_kp.description} ({_kp.code})", ip)
_notify(db, gate_db.name, f"{_kp.description} ({_kp.code})", ip)
_notify(db, gate_db.name, f"{_kp.description}", ip)
return {"success": True, "gate": gate_db.name}

30
src/services/gates.py
View File

@@ -1,21 +1,37 @@
from typing import Optional
from .avconnect import AVConnectAPI
from .shelly import ShellyCloudAPI
def call_open_gate(
macro_id: str,
username: str,
password: str,
session_id: Optional[str] = None,
api_provider: str,
avconnect_macro_id: Optional[str] = None,
avconnect_username: Optional[str] = None,
avconnect_password: Optional[str] = None,
avconnect_session_id: Optional[str] = None,
shelly_device_id: Optional[str] = None,
shelly_server_uri: Optional[str] = None,
shelly_auth_key: Optional[str] = None,
mock: bool = False,
) -> tuple[bool, Optional[str], Optional[str]]:
"""Attempt to open a gate. Returns (success, error_msg, new_session_id)."""
"""Attempt to open a gate. Returns (success, error_msg, new_avconnect_session_id)."""
if mock:
return True, None, None
if api_provider == "shelly":
try:
api = AVConnectAPI(username, password, session_id)
ok, new_sid = api.exec_gate_macro(macro_id)
assert shelly_server_uri and shelly_auth_key and shelly_device_id
ShellyCloudAPI(shelly_server_uri, shelly_auth_key).open_gate(shelly_device_id)
return True, None, None
except Exception as e:
return False, str(e), None
# Default: AVConnect
try:
assert avconnect_username and avconnect_password and avconnect_macro_id
api = AVConnectAPI(avconnect_username, avconnect_password, avconnect_session_id)
ok, new_sid = api.exec_gate_macro(avconnect_macro_id)
if not ok:
return False, "Gate did not confirm open", new_sid
return True, None, new_sid

47
src/static/admin.html
View File

@@ -122,7 +122,7 @@
<nav class="tabs">
<button class="tab-btn active" data-tab="keypasses">Keypasses</button>
<button class="tab-btn" data-tab="gates">Gates</button>
<button class="tab-btn admin-only" data-tab="credentials">AVConnect Credentials</button>
<button class="tab-btn admin-only" data-tab="credentials">API Credentials</button>
<button class="tab-btn" data-tab="stats">Statistics</button>
<button class="tab-btn admin-only" data-tab="admins">Admins</button>
<button class="tab-btn admin-only" data-tab="telegram">Notifications</button>
@@ -167,7 +167,8 @@
<th>Name</th>
<th>Group</th>
<th>Type</th>
<th>AVConnect Macro ID</th>
<th>Provider</th>
<th>Device / Macro ID</th>
<th>Status</th>
<th></th>
</tr>
@@ -179,7 +180,10 @@
<!-- ── Credentials pane ───────────────────────────────────────────── -->
<div id="tab-credentials" class="tab-pane">
<h3 style="margin-bottom:1rem">AVConnect Credentials</h3>
<h3 style="margin-bottom:1rem">API Credentials</h3>
<!-- AVConnect -->
<h4 style="margin:0 0 .75rem">AVConnect</h4>
<div class="card" style="max-width:440px">
<form id="credentials-form">
<div class="field">
@@ -195,14 +199,34 @@
<button type="submit" class="btn btn-primary">Save</button>
</form>
</div>
<!-- Shelly Cloud -->
<h4 style="margin:1.5rem 0 .75rem">Shelly Cloud</h4>
<div class="card" style="max-width:440px">
<form id="shelly-credentials-form">
<div class="field">
<label for="shelly-server-uri">Server URI</label>
<input id="shelly-server-uri" type="url" autocomplete="off"
placeholder="e.g. https://shelly-3.eu.shelly.cloud" />
</div>
<div class="field">
<label for="shelly-auth-key">Auth Key</label>
<input id="shelly-auth-key" type="password" autocomplete="new-password"
placeholder="Leave empty to keep current" />
</div>
<p id="shelly-cred-error" class="error-msg hidden"></p>
<button type="submit" class="btn btn-primary">Save</button>
</form>
</div>
<h3 style="margin:1.5rem 0 1rem">Mock Mode</h3>
<div class="card" style="max-width:440px">
<p style="color:var(--text-muted);font-size:.9rem;margin-bottom:1rem">
When enabled, gate open requests always succeed without contacting AVConnect.
When enabled, gate open requests always succeed without contacting any API.
</p>
<label style="display:flex;align-items:center;gap:.75rem;cursor:pointer;margin:0">
<input type="checkbox" id="mock-toggle" style="width:1.1rem;height:1.1rem;flex-shrink:0;cursor:pointer" />
<span style="font-weight:600">Enable mock AVConnect</span>
<span style="font-weight:600">Enable mock mode</span>
</label>
</div>
</div>
@@ -450,8 +474,19 @@
</select>
</div>
<div class="field">
<label for="gate-api-provider">API Provider</label>
<select id="gate-api-provider">
<option value="avconnect">AVConnect</option>
<option value="shelly">Shelly Cloud</option>
</select>
</div>
<div class="field" id="gate-avconnect-field">
<label for="gate-avconnect-macro-id">AVConnect Macro ID</label>
<input id="gate-avconnect-macro-id" type="text" placeholder="e.g. 42" required />
<input id="gate-avconnect-macro-id" type="text" placeholder="e.g. 42" />
</div>
<div class="field" id="gate-shelly-field" style="display:none">
<label for="gate-shelly-device-id">Shelly Device ID</label>
<input id="gate-shelly-device-id" type="text" placeholder="e.g. e0:98:06:xx:xx:xx" />
</div>
<div class="field">
<label for="gate-status">Status</label>

70
src/static/admin.js
View File

@@ -429,13 +429,16 @@ async function loadGates() {
const badge = g.status === "enabled"
? '<span class="badge badge-green">Enabled</span>'
: '<span class="badge badge-muted">Disabled</span>';
const providerLabel = g.api_provider === 'shelly' ? 'Shelly' : 'AVConnect';
const deviceId = g.api_provider === 'shelly' ? (g.shelly_device_id || '') : (g.avconnect_macro_id || '');
const tr = document.createElement("tr");
tr.innerHTML = `
<td>${g.id}</td>
<td>${esc(g.name)}</td>
<td>${g.group_name ? esc(g.group_name) : '<span style="color:var(--text-muted)">—</span>'}</td>
<td>${g.gate_type === "car" ? "🚘 Car" : "🚶 Pedestrian"}</td>
<td><code style="font-size:.85em">${esc(g.avconnect_macro_id)}</code></td>
<td>${g.group_name ? esc(g.group_name) : '<span style="color:var(--text-muted)">\u2014</span>'}</td>
<td>${g.gate_type === "car" ? "\u{1F698} Car" : "\u{1F6B6} Pedestrian"}</td>
<td><span style="font-size:.85em">${esc(providerLabel)}</span></td>
<td><code style="font-size:.85em">${esc(deviceId)}</code></td>
<td>${badge}</td>
<td><div style="text-align:right;white-space:nowrap;display:flex;gap:.5rem;justify-content:flex-end">
${g.status === 'enabled' ? `<button class="btn btn-primary" style="font-size:.8rem;padding:.35rem .9rem"
@@ -481,7 +484,11 @@ function openGateModal(gate = null) {
document.getElementById("gate-name").value = gate ? gate.name : "";
document.getElementById("gate-group-name").value = gate ? (gate.group_name || "") : "";
document.getElementById("gate-type").value = gate ? gate.gate_type : "car";
document.getElementById("gate-avconnect-macro-id").value = gate ? gate.avconnect_macro_id : "";
const provider = gate ? (gate.api_provider || "avconnect") : "avconnect";
document.getElementById("gate-api-provider").value = provider;
document.getElementById("gate-avconnect-macro-id").value = gate ? (gate.avconnect_macro_id || "") : "";
document.getElementById("gate-shelly-device-id").value = gate ? (gate.shelly_device_id || "") : "";
_updateGateProviderFields(provider);
document.getElementById("gate-status").value = gate ? gate.status : "enabled";
document.getElementById("gate-lat").value = (gate && gate.lat != null) ? gate.lat : "";
document.getElementById("gate-lon").value = (gate && gate.lon != null) ? gate.lon : "";
@@ -498,17 +505,41 @@ function openGateModal(gate = null) {
document.getElementById("gate-modal").classList.remove("hidden");
}
function _updateGateProviderFields(provider) {
const avField = document.getElementById("gate-avconnect-field");
const shellyField = document.getElementById("gate-shelly-field");
const avInput = document.getElementById("gate-avconnect-macro-id");
const shellyInput = document.getElementById("gate-shelly-device-id");
if (provider === "shelly") {
avField.style.display = "none";
shellyField.style.display = "";
avInput.required = false;
shellyInput.required = true;
} else {
avField.style.display = "";
shellyField.style.display = "none";
avInput.required = true;
shellyInput.required = false;
}
}
document.getElementById("btn-new-gate").addEventListener("click", () => openGateModal());
document.getElementById("gate-cancel").addEventListener("click", () => {
document.getElementById("gate-modal").classList.add("hidden");
});
document.getElementById("gate-api-provider").addEventListener("change", e => {
_updateGateProviderFields(e.target.value);
});
document.getElementById("gate-form").addEventListener("submit", async e => {
e.preventDefault();
const editId = document.getElementById("gate-edit-id").value;
const provider = document.getElementById("gate-api-provider").value;
const payload = {
name: document.getElementById("gate-name").value.trim(),
gate_type: document.getElementById("gate-type").value,
avconnect_macro_id: document.getElementById("gate-avconnect-macro-id").value.trim(),
api_provider: provider,
avconnect_macro_id: provider === "avconnect" ? document.getElementById("gate-avconnect-macro-id").value.trim() : null,
shelly_device_id: provider === "shelly" ? document.getElementById("gate-shelly-device-id").value.trim() : null,
status: document.getElementById("gate-status").value,
group_name: document.getElementById("gate-group-name").value.trim() || null,
lat: document.getElementById("gate-lat").value !== "" ? parseFloat(document.getElementById("gate-lat").value) : null,
@@ -539,6 +570,12 @@ async function loadCredentials() {
document.getElementById("cred-username").value = list[0].username;
}
} catch { /* no creds yet */ }
try {
const shelly = await api("GET", "/api/admin/credentials/shelly");
if (shelly) {
document.getElementById("shelly-server-uri").value = shelly.server_uri;
}
} catch { /* no shelly creds yet */ }
try {
const { enabled } = await api("GET", "/api/admin/credentials/mock");
document.getElementById("mock-toggle").checked = enabled;
@@ -569,7 +606,28 @@ document.getElementById("credentials-form").addEventListener("submit", async e =
try {
await api("PUT", "/api/admin/credentials", { username, password });
document.getElementById("cred-password").value = "";
showToast("Credentials saved");
showToast("AVConnect credentials saved");
} catch (e) {
errEl.textContent = e.message;
errEl.classList.remove("hidden");
}
});
document.getElementById("shelly-credentials-form").addEventListener("submit", async e => {
e.preventDefault();
const server_uri = document.getElementById("shelly-server-uri").value.trim();
const auth_key = document.getElementById("shelly-auth-key").value;
const errEl = document.getElementById("shelly-cred-error");
errEl.classList.add("hidden");
if (!auth_key) {
errEl.textContent = "Auth key is required.";
errEl.classList.remove("hidden");
return;
}
try {
await api("PUT", "/api/admin/credentials/shelly", { server_uri, auth_key });
document.getElementById("shelly-auth-key").value = "";
showToast("Shelly Cloud credentials saved");
} catch (e) {
errEl.textContent = e.message;
errEl.classList.remove("hidden");