mirror of
https://github.com/usetrmnl/byos_fastapi.git
synced 2026-04-29 13:44:09 -07:00
81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
import logging
|
|
from typing import List, Sequence, Tuple
|
|
|
|
import httpx
|
|
|
|
from .base import ChartPlugin
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Node-RED API endpoints for local web stats
|
|
PAGEVIEWS_URL = "http://192.168.1.100:1880/api/site/pageviews"
|
|
VISITORS_URL = "http://192.168.1.100:1880/api/site/visitors"
|
|
|
|
# Node-RED API endpoint for total power consumption
|
|
TOTAL_POWER_URL = "http://192.168.1.100:1880/api/power/all"
|
|
|
|
|
|
async def _fetch_series(url: str) -> List[Tuple[str, int]]:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.get(url)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
entries = payload.get('data', [])
|
|
parsed: List[Tuple[str, int]] = []
|
|
for entry in entries:
|
|
name = str(entry.get('name', '')).zfill(2)
|
|
try:
|
|
value = int(entry.get('value', 0))
|
|
except (TypeError, ValueError):
|
|
value = 0
|
|
parsed.append((name, value))
|
|
return parsed
|
|
|
|
|
|
class PageviewsPlugin(ChartPlugin):
|
|
"""Render hourly pageviews chart."""
|
|
|
|
DISPLAY_NAME = "Web Stats - Pageviews"
|
|
SERIES_LABEL = "Pageviews"
|
|
BASENAME = "webstats_pageviews"
|
|
OUTPUT_SUBDIR = "webstats"
|
|
REGISTRY_ORDER = 60
|
|
REFRESH_INTERVAL = 1800
|
|
CHART_STYLE = "area"
|
|
CAPTION_TEXT = "Estimated pageviews - last 24h"
|
|
|
|
async def _fetch_series(self) -> Sequence[Tuple[str, int]]:
|
|
return await _fetch_series(PAGEVIEWS_URL)
|
|
|
|
|
|
class VisitorsPlugin(ChartPlugin):
|
|
"""Render hourly visitors chart."""
|
|
|
|
DISPLAY_NAME = "Web Stats - Visitors"
|
|
SERIES_LABEL = "Visitors"
|
|
BASENAME = "webstats_visitors"
|
|
OUTPUT_SUBDIR = "webstats"
|
|
REGISTRY_ORDER = 70
|
|
REFRESH_INTERVAL = 1800
|
|
CHART_STYLE = "area"
|
|
CAPTION_TEXT = "Estimated unique visitors - last 24h"
|
|
|
|
async def _fetch_series(self) -> Sequence[Tuple[str, int]]:
|
|
return await _fetch_series(VISITORS_URL)
|
|
|
|
|
|
class TotalPowerPlugin(ChartPlugin):
|
|
"""Render the total power chart from the local power feed."""
|
|
|
|
DISPLAY_NAME = "Power - Total"
|
|
SERIES_LABEL = "Total Power (W)"
|
|
BASENAME = "power_total"
|
|
OUTPUT_SUBDIR = "power"
|
|
REGISTRY_ORDER = 80
|
|
REFRESH_INTERVAL = 1800
|
|
CHART_STYLE = "area"
|
|
CAPTION_TEXT = "All smart outlets combined - last 24h"
|
|
|
|
async def _fetch_series(self) -> Sequence[Tuple[str, int]]:
|
|
return await _fetch_series(TOTAL_POWER_URL)
|