Merge branch 'library_additions'

pull/51/head
Marc 2021-08-16 21:52:24 +01:00
commit 6a32e5b9b9
No known key found for this signature in database
GPG Key ID: 0657563F705ACAAE
3 changed files with 51 additions and 18 deletions

View File

@ -1,4 +1,5 @@
import json
from typing import Tuple
def json_to_bytes(message) -> bytes:
@ -14,3 +15,14 @@ def json_to_bytes(message) -> bytes:
d = json.dumps(message)
return d.encode('utf-8')
def get_version() -> Tuple[str, str]:
"""
Get the system firmware version.
:return: tuple containing version and build number
"""
with open('/etc/pineapple/version', 'r') as f:
vers = [l.strip() for l in f.readlines()]
return vers[0], vers[1]

View File

@ -1,21 +1,36 @@
from logging import Logger
from typing import Optional, Tuple, List, Union
import subprocess
import os
def uci_get(property_name) -> str:
def uci_set(key: str, value) -> bool:
"""
Get the value of a specified UCI property.
:param property_name: The UCI property to look up.
:return: The property value if found, empty string if not found.
"""
return subprocess.run(['uci', 'get', property_name], capture_output=True, check=True).stdout
def uci_set(property_name, value) -> bool:
"""
Set the value of a specified UCI property to a value.
:param property_name: The UCI property to set.
:param value: The value to set the property to
Set a UCI property to value
:param key: The UCI key to target
:param value: The value to set the UCI key to
:return: True if successful, false if not.
"""
uci_string = property_name + '=' + value
return subprocess.run(['uci', 'set', uci_string], check=True).returncode == 0
argument = f"{key}='{value}'"
out = subprocess.run(["uci", "set", argument])
if out.returncode != 0:
return False
out = subprocess.run(["uci", "commit"])
if out.returncode != 0:
return False
return True
def uci_get(key: str) -> str:
"""
Get a UCI value from it's key
:param key: The UCI key to target
:return: str
"""
out = subprocess.run(["uci", "get", key])
return out.stdout

View File

@ -32,15 +32,21 @@ class Module:
# A dictionary mapping an action to a function.
self._action_handlers: Dict[str, Callable[[Request], Union[Any, Tuple[bool, Any]]]] = {}
self._running: bool = False # set to False to stop the module loop
# Set to False to stop the module loop
self._running: bool = False
# api requests will be received over this socket
# Create a storage folder for the module by default
self._storage_folder_path = f'/root/.{name}/'
if not os.path.exists(self._storage_folder_path):
os.mkdir(self._storage_folder_path)
# API requests will be received over this socket
self._module_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self._module_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._module_socket_path = f'/tmp/modules/{name}.sock' # apth to the socket
self._buffer_size = 10485760
# if the socket already exists attempt to delete it.
# If the socket already exists attempt to delete it.
try:
os.unlink(self._module_socket_path)
except OSError: