Don't require a full pin scan for the root page.

The full pin scan is slow, and it's slowing me down. Instead, just load
what we know we need for the configured relays unless we are adding a
new relay.
This commit is contained in:
Eli Ribble 2023-05-26 09:44:17 -07:00
parent dd637c2eaa
commit b4027baf68
2 changed files with 61 additions and 49 deletions

View File

@ -87,10 +87,10 @@ class Manager:
self.configpath = configpath self.configpath = configpath
self.has_fakes = has_fakes self.has_fakes = has_fakes
# Handles to various chips # Handles to various chips, mapped by chip number
self._chips = [] self._chips = {}
# Info on various pins # Info on various pins, mapped by (chip number, line_number)
self._pins = [] self._pins = {}
# Cached info on the relays # Cached info on the relays
self._relays = [] self._relays = []
# Connection to single-board computer GPIO # Connection to single-board computer GPIO
@ -104,7 +104,7 @@ class Manager:
def chips(self) -> List[Chip]: def chips(self) -> List[Chip]:
if not self._chips: if not self._chips:
self._load_chips() self._load_chips()
return self._chips return self._chips.values()
def connect(self) -> None: def connect(self) -> None:
self.sbc = rgpio.sbc( self.sbc = rgpio.sbc(
@ -116,16 +116,18 @@ class Manager:
self.sbc = None self.sbc = None
def get_chip_by_number(self, number: int) -> Chip: def get_chip_by_number(self, number: int) -> Chip:
for chip in self.chips: try:
if chip.number == number: return self._chips[number]
return chip except KeyError:
raise Exception(f"Can't find chip {number}") self._load_chip(number)
return self._chips[number]
def get_pin_by_number(self, chip: Chip, line_number: int) -> Pin: def get_pin_by_number(self, chip: Chip, line_number: int) -> Pin:
for pin in self.pins: try:
if pin.chip == chip and pin.line_number == line_number: return self._pins[(chip.number, line_number)]
return pin except KeyError:
raise Exception(f"Can't find pin {chip.name}-{line_number}") self._load_pin(chip, line_number)
return self._pins[(chip.number, line_number)]
def get_relay_by_pin_id(self, pin_id: str) -> Relay: def get_relay_by_pin_id(self, pin_id: str) -> Relay:
for relay in self.relays: for relay in self.relays:
@ -133,11 +135,15 @@ class Manager:
return relay return relay
return None return None
def load_all_pins(self) -> None:
self._load_chips()
self._load_pins()
@property @property
def pins(self) -> List[Pin]: def pins(self) -> List[Pin]:
if not self._pins: if not self._pins:
self._load_pins() self._load_pins()
return self._pins return self._pins.values()
@property @property
def relays(self) -> Iterable[Relay]: def relays(self) -> Iterable[Relay]:
@ -177,50 +183,55 @@ class Manager:
def shutdown(self) -> None: def shutdown(self) -> None:
_write_config(self.configpath, self.config) _write_config(self.configpath, self.config)
def _load_chip(self, number: int) -> None:
try:
handle = self.sbc.gpiochip_open(number)
except rgpio.error:
return
if handle < 0:
return
okay, number_of_lines, name, label = self.sbc.gpio_get_chip_info(handle)
LOGGER.info("Chip info: %s %s %s %s", okay, number_of_lines, name, label)
if okay != 0:
LOGGER.warn("Chip %s not okay.", name)
return
self._chips[number] = Chip(
handle=handle,
label=label,
name=name,
number=number,
number_of_lines=number_of_lines,
)
# Talk to the remote pins daemon and get information about the chips # Talk to the remote pins daemon and get information about the chips
def _load_chips(self) -> None: def _load_chips(self) -> None:
for i in range(MAX_CHIPS): for i in range(MAX_CHIPS):
try: self._load_chip(i)
handle = self.sbc.gpiochip_open(i)
except rgpio.error: def _load_pin(self, chip: Chip, line_number: int) -> None:
continue okay, offset, flags, name, user = self.sbc.gpio_get_line_info(chip.handle, line_number)
if handle < 0: LOGGER.info("Got line info: %s %s %s %s %s", okay, offset, flags, name, user)
continue assert offset == line_number
okay, number_of_lines, name, label = self.sbc.gpio_get_chip_info(handle) if okay != 0:
LOGGER.info("Chip info: %s %s %s %s", okay, number_of_lines, name, label) LOGGER.warn("Line %s is not okay", name)
if okay != 0: return
LOGGER.warn("Chip %s not okay.", name) if not name:
continue LOGGER.warn("Ignoring line %d because it has no name.", line_number)
self._chips.append(Chip( return
handle=handle, self._pins[(chip.number, line_number)] = Pin(
label=label, chip=chip,
name=name, flags=flags,
number=i, line_number=line_number,
number_of_lines=number_of_lines, name=name,
)) user=user,
)
# Talk to the remote pins daemon and get information about the pins # Talk to the remote pins daemon and get information about the pins
def _load_pins(self) -> None: def _load_pins(self) -> None:
LOGGER.info("Loading pins") LOGGER.info("Loading pins")
for chip in self.chips: for chip in self.chips:
for line in range(chip.number_of_lines): for line in range(chip.number_of_lines):
okay, offset, flags, name, user = self.sbc.gpio_get_line_info(chip.handle, line) self._load_pin(chip, line)
LOGGER.info("Got line info: %s %s %s %s %s", okay, offset, flags, name, user)
assert offset == line
if okay != 0:
LOGGER.warn("Line %s is not okay", name)
continue
if not name:
LOGGER.warn("Ignoring line %d because it has no name.", line)
continue
self._pins.append(Pin(
chip=chip,
flags=flags,
line_number=line,
name=name,
user=user,
))
def _read_config(configpath: pathlib.Path) -> None: def _read_config(configpath: pathlib.Path) -> None:
with open(configpath, "rb") as f: with open(configpath, "rb") as f:

View File

@ -36,6 +36,7 @@ def index(request: Request):
@app.get("/relay/create") @app.get("/relay/create")
def relay_create_get(request: Request): def relay_create_get(request: Request):
"Get the form to create a new relay." "Get the form to create a new relay."
relays.load_all_pins()
pins = relays.pins pins = relays.pins
return templates.TemplateResponse("relay-create.template.html", { return templates.TemplateResponse("relay-create.template.html", {
"request": request, "request": request,