This is just work-in-progress. Switching underlying frameworks...

I wanna go FastAPI.
This commit is contained in:
Eli Ribble 2023-05-15 14:46:58 -07:00
parent 320591277f
commit c36a32179a
6 changed files with 153 additions and 6 deletions

View File

@ -1,6 +1,7 @@
import argparse
import asyncio
import logging
import pathlib
import pnpdevice.server
@ -8,14 +9,16 @@ LOGGER = logging.getLogger(__name__)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose logging")
parser.add_argument("-c", "--config", default="/etc/pnpdevice/config.toml", type=pathlib.Path, help="The config file to use.")
parser.add_argument("-s", "--simulate", action="store_true", help="When present, simulate the state of the relays")
parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose logging")
args = parser.parse_args()
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
if args.simulate:
relays = pnpdevice.relays.RelaysFake()
relays = pnpdevice.relays.RelaysFake(args.config)
else:
relays = pnpdevice.relays.RelaysReal()
relays = pnpdevice.relays.RelaysFake(args.config)
#relays = pnpdevice.relays.RelaysReal(args.config)
pnpdevice.server.run(relays)

View File

@ -1,9 +1,110 @@
import dataclasses
import logging
import os
import pathlib
import tomllib
from typing import Dict, List, Union
import tomli_w
LOGGER = logging.getLogger(__name__)
@dataclasses.dataclass
class Pin:
chip: str
number: int
name: str
@dataclasses.dataclass
class Relay:
pin: Pin
name: str
state: bool
def _deserialize(data: List[Dict[str, str]]) -> List[Relay]:
"Deserialize a list of relays from the config."
return [Relay(
name=relay["name"],
pin=Pin(
chip=relay["chip"],
number=int(relay["pinnumber"]),
name=relay["pinname"],
)) for relay in data]
def _serialize(data: List[Relay]) -> List[Dict[str, str]]:
"Serialize a list of relays to the config."
return [{
"chip": relay.pin.chip,
"name": relay.name,
"pinnumber": str(relay.pin.number),
"pinname": relay.pin.name,
} for relay in data]
class Relays:
"Class for interacting with relays."
def __init__(self, configpath: pathlib.Path):
self.configpath = configpath
try:
with open(configpath, "rb") as f:
content = tomllib.load(f)
self.config = content
except Exception as e:
LOGGER.info("Unable to load config file: %s", e)
self.config = {
"relays": [],
}
self.relays = _deserialize(self.config["relays"])
def chips_list(self) -> List[str]:
raise NotImplementedError()
def pins_list(self) -> List[Pin]:
raise NotImplementedError()
def relay_add(self, chip: str, pinnumber: int, name: str) -> None:
self.config["relays"].append({
"chip": chip,
"name": name,
"pinnumber": pinnumber,
})
with open(self.configpath, "wb") as f:
tomli_w.dump({
"relays": _serialize(self.config["relays"]),
}, f)
LOGGER.info("Wrote config file %s", self.configpath)
def relays_list(self) -> List[Relay]:
return self.relays
def relay_on(self, name: str) -> bool:
raise NotImplementedError()
def relay_set(self, name: str, state: bool) -> None:
raise NotImplementedError()
class RelaysFake(Relays):
"Class for fake relays, useful for testing."
def chips_list(self) -> List[str]:
return ["chipA", "chipB"]
def pins_list(self) -> List[Pin]:
return [
Pin(chip="chipA", number=1, name="CON2-P1"),
Pin(chip="chipA", number=2, name="CON2-P2"),
Pin(chip="chipA", number=3, name="CON2-P10"),
Pin(chip="chipB", number=1, name="CON2-P3"),
Pin(chip="chipB", number=2, name="CON2-P7"),
]
def relays_list(self) -> List[Relay]:
raise NotImplementedError()
def relay_on(self, name: str) -> bool:
raise NotImplementedError()
def relay_set(self, name: str, state: bool) -> None:
raise NotImplementedError()
class RelaysReal(Relays):
"Class for controlling real relays."

View File

@ -10,17 +10,36 @@ def html(text: str) -> web.Response:
@aiohttp_jinja2.template("index.template.html")
async def index(request):
return {}
relays = request.app["relays"]
return {
"relays": relays.relays_list(),
}
async def status(request):
@aiohttp_jinja2.template("relay-create.template.html")
async def relay_create_get(request: Request):
"Get the form to create a new relay."
relays = request.app["relays"]
pins = relays.pins_list()
return {"pins": pins}
async def relay_create_post(request: Request, chip_and_number: str, name: str):
"Create a new relay."
chip, number = chip_and_number.partition("-")
LOGGER.info("Creating relay %s %s with name %s", chip, number, name)
return RedirectResponse(status_code=303, url="/")
async def status(request: Request):
return html("Status")
def run(relays: pnpdevice.relays.Relays):
"Run the embedded web server"
app = web.Application()
app["relays"] = relays
aiohttp_jinja2.setup(app,
loader=jinja2.FileSystemLoader("templates"))
app.on_startup.append(pnpdevice.discovery.handle)
app.add_routes([web.get("/", index)])
app.add_routes([web.get("/relay/create", relay_create_get)])
app.add_routes([web.post("/relay/create", relay_create_post)])
web.run_app(app)

View File

@ -6,6 +6,7 @@ authors = [
dependencies = [
"aiohttp",
"aiohttp_jinja2",
"tomli-w",
"zeroconf",
]
dynamic = ["version", "description"]

View File

@ -1,3 +1,14 @@
<html>
<h1>Pools'n'Pumps Device</h1>
<h1>Pools'n'Pumps Device</h1>
{% if not relays %}
<h2>No relays found.</h2>
{% else %}
<h2>Relays</h2>
<ul>
{% for relay in relays %}
<li>{{ relay.name }}</li>
{% endfor %}
</ul>
{% endif %}
<a href="/relay/create">Create Relay</a>
</html>

View File

@ -0,0 +1,12 @@
<html>
<h1>Relay Creation</h1>
<form method="POST" action="/relay/create">
<input type="text" name="name" placeholder="Pool Pump 1"></input>
<select name="chip-and-number">
{% for pin in pins %}
<option value="{{ pin.chip }}-{{ pin.number }}">{{ pin.name }}</option>
{% endfor %}
</select>
<button type="submit">Create</button>
</form>
</html>