diff --git a/SBCEye.py b/SBCEye.py index 860c298..6cce89b 100644 --- a/SBCEye.py +++ b/SBCEye.py @@ -3,7 +3,6 @@ SBCEye: Animate the OLED display attached to my OctoPrint server with bme280 and system data Show, log and graph the environmental, system and gpio data via a web interface -Give me a on/off button + url to control the bench lights via a GPIO pin !! DISPLAY, BME280 and GPIO functionality is CURRENTLY Raspberry PI only! - Needs to be made generic for other architectures @@ -41,27 +40,30 @@ import sys import logging import random +import schedule +import psutil from datetime import timedelta from logging.handlers import RotatingFileHandler from atexit import register from signal import signal, SIGTERM, SIGINT, SIGHUP from multiprocessing import Process, Queue -import schedule -import psutil +from pathlib import Path # Local classes from load_config import Settings from robin import Robin from httpserver import serve_http from netreader import Netreader -from pinreader import Pinreader -from bus_drivers import i2c_setup +from gpiohandler import GPIOHandler +from i2c_bus import i2c_setup +from bme_sensor import bme_setup +from oled_display import oled_setup # Re-nice to reduce blocking of other processes os.nice(10) # The setting class will also process the arguments -settings = Settings() +settings = Settings(appname=Path(sys.argv[0]).stem) # Let the console know we are starting print("Starting SBCEye") @@ -78,7 +80,7 @@ datefmt=settings.short_format, handlers=[handler]) -# Older scheduler versions can log debug to 'INFO' not 'DEBUG', change threshold. +# Older scheduler versions might log debug at wrong level, change threshold. schedule_logger = logging.getLogger('schedule') schedule_logger.setLevel(level=logging.WARN) @@ -105,26 +107,18 @@ # # Import, setup and return hardware drivers, or 'None' if setup fails -disp, bme280 = i2c_setup(settings.have_screen, settings.have_sensor) - -if disp: - disp.contrast(settings.display_contrast) - disp.invert(settings.display_invert) - disp.fill(0) # Blank asap in case we are showing garbage - disp.show() - -if settings.button_out > 0: - try: - from RPi import GPIO - except ImportError as e: - print(e) - print("ERROR: button & pin control requirements not met, features disabled") - settings.button_out = 0 +i2c, bus_lock = i2c_setup(settings) # # Local Classes, Globals -display_queue = None # will be set during +# We override the dictionary class so that every time an item is +# modified it sends a a message to the display queue. +# This allows the display to run in a seperate process while keeping +# it's local data copy updated in real-time. +# The queue is initially disabled (type: None), and assigned as a +# queue object only if the display is enabled and detected. +display_queue = None class TheData(dict): '''Override the dictionary class to also send data to the queue for the display''' def __setitem__(self, item, value): @@ -136,7 +130,7 @@ def __delitem__(self, item): display_queue.put([item], None) super().__delitem__(item) -# Use a (custom overridden) dictionary to store current readings +# Use this overridden dictionary to store current readings data = TheData({}) # Counters used for incremental data need pre-populating @@ -152,44 +146,8 @@ def __delitem__(self, item): # # Local functions -def button_control(action="toggle"): - '''Set the controlled pin to a specified state''' - if settings.button_out > 0: - ret = f'{settings.button_name} ' - pin = settings.button_out - if action.lower() in ['toggle','invert','button']: - GPIO.output(pin, not GPIO.input(pin)) - ret += 'Toggled: ' - elif action.lower() in [settings.pin_state_names[1].lower(),'on','true']: - GPIO.output(pin,True) - ret += 'Switched: ' - elif action.lower() in [settings.pin_state_names[0].lower(),'off','false']: - GPIO.output(pin,False) - ret += 'Switched: ' - elif action.lower() in ['random','easter']: - GPIO.output(pin,random.choice([True, False])) - ret += 'Randomly Switched: ' - else: - ret += ': ' - state = GPIO.input(pin) - ret += settings.pin_state_names[state] - else: - state = False - ret = 'Not supported, no output pin defined' - pins.update_pins() - return (ret, state) - -def button_interrupt(*_): - '''give a short delay, then re-read input to provide a minimum hold-down time - and suppress false triggers from other gpio operations''' - time.sleep(settings.button_hold) - if GPIO.input(settings.button_pin): - logging.info('Button pressed') - button_control() - def update_system(): - '''Get current environmental and system data, called on a schedule - ''' + '''Get current environmental and system data, called on a schedule''' data['sys-temp'] = psutil.sensors_temperatures()[cpu_thermal_device][0].current data['sys-load'] = psutil.getloadavg()[0] data["sys-freq"] = psutil.cpu_freq().current @@ -211,12 +169,12 @@ def update_system(): counter["sys-cpu-int"] = int_count def update_sensors(): - '''Get current environmental sensor data - ''' - if bme280: - data['env-temp'] = bme280.temperature - data['env-humi'] = bme280.relative_humidity - data['env-pres'] = bme280.pressure + '''Get current environmental sensor data''' + if bme: + bme.update_sensor() + data['env-temp'] = bme.temperature + data['env-humi'] = bme.humidity + data['env-pres'] = bme.pressure # Failed pressure measurements really foul up the graph, skip if data['env-pres'] == 0: data['env-pres'] = 'U' @@ -228,22 +186,22 @@ def update_data(): net.update(data) rrd.update(data) -def hourly(): +def daily(): '''Remind everybody we are alive''' myself = os.path.basename(__file__) timestamp = time.strftime(settings.long_format) uptime = timedelta(seconds=int(time.time() - psutil.boot_time())) - logging.info(f'{settings.name} :: up {uptime}') - print(f'{myself} :: {timestamp} :: {settings.name} :: up {uptime}') + logging.info(f'{settings.name} :: system uptime {uptime}') + print(f'{myself} :: {timestamp} :: {settings.name} :: system uptime {uptime}',flush=True) def handle_signal(sig, *_): '''Handle common signals''' if DISPLAY: - # clean up the screen process + # clean up the display process DISPLAY.join() if sig == SIGHUP: handle_restart() - elif sig == SIGINT and settings.debug: + elif sig == SIGINT and settings.debug_sigint: handle_restart() else: # calling sys.exit() will invoke handle_exit() @@ -252,46 +210,38 @@ def handle_signal(sig, *_): def handle_restart(): '''In-Place safe restart (re-reads config)''' logging.info('Safe Restarting') - print('Restart\n') + print('Restart\n',flush=True) rrd.write_updates() + if bus_lock: + bus_lock.release() os.execv(sys.executable, ['python'] + sys.argv) def handle_exit(): '''Ensure we write ipending data to the RRD database as we exit''' rrd.write_updates() + if bus_lock: + bus_lock.release() logging.info('Exiting') - print('Graceful Exit\n') + print('Graceful Exit\n',flush=True) # The fun starts here: if __name__ == '__main__': - # Log sensor status - if bme280: + # Environmental sensor + bme = bme_setup(i2c, settings) if i2c else None + if bme: logging.info('Environmental sensor configured and enabled') elif settings.have_sensor: - logging.warning('Environmental data configured but no sensor detected: '\ + logging.warning('Environmental data configured but no sensor available: '\ 'Environment status and logging disabled') - # Set button interrupt and output if we have a button and a pin to control - if settings.button_out > 0: - GPIO.setmode(GPIO.BCM) # Use BCM GPIO numbering - GPIO.setup(settings.button_out, GPIO.OUT) - logging.info(f'Controllable pin ({settings.button_name}) enabled') - if settings.button_pin > 0: - GPIO.setup(settings.button_pin, GPIO.IN) - # Set up the button pin interrupt - GPIO.add_event_detect(settings.button_pin, - GPIO.RISING, button_interrupt, - bouncetime = int(settings.button_hold * 2000)) - logging.info('Button enabled') - if len(settings.button_url) > 0: - logging.info(f'Web Button enabled on: /{settings.button_url}') - print(f'Controllable pin ({settings.button_name}) configured and enabled; '\ - f'(pin={settings.button_pin}, url="{settings.button_url})"') - # Display animation setup + disp = oled_setup(settings) if i2c else None if disp: + # display initialisation does a 'clear()' and 'show()' + disp.contrast(settings.display_contrast) + from animator import animate display_queue = Queue() DISPLAY = Process(target=animate, args=(settings, disp, display_queue), @@ -299,13 +249,13 @@ def handle_exit(): DISPLAY.start() else: DISPLAY = None - if settings.have_screen: + if settings.have_display: logging.warning('Display configured but did not initialise properly: '\ 'Display features disabled') print('Performing initial data update', end='') - if settings.net_map: - print(f' may take up to {settings.net_timeout}s if ping targets are down') + if settings.netlist: + print(f' (may take up to {settings.net_timeout}s if ping targets are down)') else: print() @@ -315,17 +265,17 @@ def handle_exit(): # Populate initial sensor data update_sensors() - # Network (ping) monitoring - net = Netreader((settings.net_map, settings.net_timeout), data) + # GPIO pin monitoring + gpio = GPIOHandler(settings.pinlist, settings.buttons, data, settings.identifier) - # GPIO Pin monitoring - pins = Pinreader((settings.pin_map, settings.pin_state_names), data) + # Network (ping) monitoring + net = Netreader((settings.netlist, settings.net_timeout), data) # RRD init now that the data{} structure is populated rrd = Robin(settings, data) # Start the web server, it will fork into a seperate thread and run continually - serve_http(settings, rrd, data, (button_control,)) + serve_http(settings, rrd, gpio, data) # Exit handlers (needed for rrd cache write on shutdown) signal(SIGTERM, handle_signal) @@ -334,11 +284,11 @@ def handle_exit(): register(handle_exit) # Schedule pin monitoring, database updates and logging events - if settings.log_hourly: - schedule.every().hour.at(":00").do(hourly) schedule.every(settings.data_interval).seconds.do(update_data) - if len(settings.pin_map.keys()) > 0: - schedule.every(settings.pin_interval).seconds.do(pins.update_pins) + if gpio.available: + schedule.every(settings.pin_interval).seconds.do(gpio.update) + if settings.log_daily: + schedule.every().day.at("00:00").do(daily) # We got this far... time to start the show logging.info("Init complete, starting schedules and entering service loop") diff --git a/SBCEye.service b/SBCEye.service index 66e35af..281bf63 100644 --- a/SBCEye.service +++ b/SBCEye.service @@ -1,3 +1,12 @@ +# (as root) Place this file in /etc/systemd/service/ +# run 'systemctl reload-daemon' then +# 'systemctl enable --now SBCEye.service +# +# Note: assumes that SBCEye has it's own # user account ('eye') +# with a homedir at '/home/eye', the SBCEye repo checked out as `/home/eye/SBCEye' +# and that a virtual environment exists at '/home/eye/SBCEye/env' +# with the required libraries installed (see docs) + [Unit] Description=SBCEye monitoring script After=multi-user.target @@ -7,6 +16,7 @@ Type=simple Restart=always WorkingDirectory=/home/eye/SBCEye/ ExecStart=/home/eye/SBCEye/env/bin/python /home/eye/SBCEye/SBCEye.py +SyslogIdentifier=SBCEye User=eye Group=eye diff --git a/animator.py b/animator.py index 6f8be23..01b93d0 100644 --- a/animator.py +++ b/animator.py @@ -47,7 +47,7 @@ class Animator: Handles starting the display and then displays the desired information screens according to user-defined 'frame' rate. Screens are 'slid' into place to provide a pleasing animation effect - A screensaver can be invoked to blank or invert the display as the user wishes + A screensaver can be invoked to blank the display as the user wishes ''' def __init__(self, settings, disp, data): @@ -60,7 +60,7 @@ def __init__(self, settings, disp, data): self.height = self.disp.height self.span = self.width*2 + self.margin - self.display_rotate = settings.display_rotate + # How fast self.animate_speed = settings.animate_speed # Create image canvas (with mode '1' for 1-bit color) @@ -93,14 +93,14 @@ def __init__(self, settings, disp, data): # Start saver saver_settings = (settings.saver_mode, settings.saver_on, - settings.saver_off, settings.display_invert) + settings.saver_off) self.screensaver = Saver(disp, saver_settings) self.screensaver.check() schedule.every().hour.at(":00").do(self.screensaver.check) # Notify logs etc logging.info('Display configured and enabled') - print('Display configured and enabled') + print('Display configured and enabled',flush=True) self._splash() @@ -109,15 +109,14 @@ def _clean(self): self.draw.rectangle((0,0,self.span-1,self.height-1), outline=0, fill=0) def _show(self, xpos=0): - '''Put a specific area of the canvas onto display''' - if self.display_rotate: - self.disp.image(self.image.transform((self.width,self.height), - Image.EXTENT,(xpos,0,xpos+self.width,self.height)) - .transpose(Image.ROTATE_180)) + '''Put a specific area of the canvas onto display, + or blank if the screensaver is active''' + if self.screensaver.active: + self.disp.hide() else: - self.disp.image(self.image.transform((self.width,self.height), - Image.EXTENT,(xpos,0,xpos+self.width,self.height))) - self.disp.show() + self.disp.display(self.image.transform((self.width,self.height), + Image.EXTENT,(xpos,0,xpos+self.width,self.height))) + self.disp.show() def _slideout(self): '''Slide the display view across the canvas to animate between screens''' @@ -211,7 +210,8 @@ def _frame(self): def _hourly(self): '''check screensaver and totally frivously do a spash screen once an hour''' self.screensaver.check() - self._splash() + if not self.screensaver.active: + self._splash() def animate(settings, disp, queue): @@ -237,7 +237,7 @@ def animate(settings, disp, queue): def die_with_dignity(*_): '''Exit cleanly (eg without stack trace) on a sigint/sigterm''' - print('Display animator process exiting') + print('Display animator process exiting',flush=True) sys_exit() signal(SIGTERM, die_with_dignity) @@ -265,3 +265,8 @@ def die_with_dignity(*_): data.pop(key, None) schedule.run_pending() sleep(0.25) + +if __name__ == "__main__": + from sys import exit + print('animator class for SBCEye, see inline docs') + exit() diff --git a/bme_sensor.py b/bme_sensor.py new file mode 100644 index 0000000..5d3f6f5 --- /dev/null +++ b/bme_sensor.py @@ -0,0 +1,52 @@ +'''Bus based environmental sensor initialisation + currently assumes a BME280 (temp/press/humi) sensor +''' +# pragma pylint: disable=import-outside-toplevel + +def bme_setup(i2c, settings): + '''Import and initialise BME sensor library, return the sensor object + + parameters: + i2c: i2c bus object + settings: settings oject (from load_config) + returns: + bme: Sensor module object, or None if failed + ''' + + bme = None + if settings.have_sensor: + # BME280 I2C Tepmerature Pressure and Humidity sensor + # Uses pimoroni library: + # https://github.com/pimoroni/bme280-python + # pip install pimoroni-bme280 + try: + import bme280 + except ImportError as error: + print(error) + print("ERROR: BME280 environment sensor requirements not met", flush=True) + return None + + try: + # Create the I2C BME280 sensor object and get initial readings + bme = bme280.BME280(i2c_addr=settings.sensor_addr, i2c_dev=i2c) + except Exception as error: + print(error) + print("We do not have a environmental sensor", flush=True) + return None + + try: + # Try initial reading of sensor, bus errors can cause this to fail + bme.update_sensor() + except Exception as error: + print(error) + print("Environmental sensor failed initial update, this may be due to bus errors", flush=True) + return None + + print("BME280 sensor found", flush=True) + return bme + +if __name__ == "__main__": + from sys import exit + print('bme280 sensor class for SBCEye, see inline docs') + exit() + diff --git a/bus_drivers.py b/bus_drivers.py deleted file mode 100644 index 56a4e47..0000000 --- a/bus_drivers.py +++ /dev/null @@ -1,97 +0,0 @@ -'''Bus based hardware driver initialisation - -Imports and starts the optional Bus based devices. -Gracefully fails and disables funcions as appropriate if anything goes wrong - -I'd probably do this differently if writing from scratch/refactoring, and -try to make better use of Importlib in SBCEye.py. -''' -# pragma pylint: disable=import-outside-toplevel - -import importlib - - -def i2c_setup(screen, sensor): - '''Import and start the I2C bus devices - - parameters: - screen: (bool) is screen enabled in config? - sensor: (bool) is environmental sensor (bme280) enabled in config? - - returns: - disp: Display driverr object or None if failed - bme280: Sensor module object, or None if failed - ''' - - disp = None - bme280 = None - - # Start by trying to load the correct modules - if screen or sensor: - # I2C Comms - try: - import busio - from board import SCL, SDA - except ImportError as error: - print(error) - print("ERROR: I2C bus requirements not met") - screen = sensor = False - - if screen: - # I2C 128x64 OLED Display - try: - import adafruit_ssd1306 - except ImportError as error: - print(error) - print("ERROR: ssd1306 display requirements not met") - screen = False - - if sensor: - # BME280 I2C Tepmerature Pressure and Humidity sensor - try: - import adafruit_bme280 - except ImportError as error: - print(error) - print("ERROR: BME280 environment sensor requirements not met") - sensor = False - - # Now the actual device driver objects - if screen or sensor: - try: - # Create the I2C interface object - i2c = busio.I2C(SCL, SDA) - print('We have a I2C bus') - except ValueError as error: - print(error) - print("No I2C bus, display and sensor functions will be disabled") - screen = sensor = False - - if screen: - try: - # Create the I2C display object - disp = adafruit_ssd1306.SSD1306_I2C(128, 64, i2c) - print("SSD1306 i2c display found") - except RuntimeError as error: - disp = None - print(error) - print("ERROR: SSD1306 i2c display failed to initialise, disabling") - - if not importlib.util.find_spec("PIL"): - disp = None - print("ERROR: PIL graphics module not found, disabling display") - - if sensor: - try: - # Create the I2C BME280 sensor object - bme280 = adafruit_bme280.Adafruit_BME280_I2C(i2c, address=0x76) - print("BME280 sensor found with address 0x76") - except RuntimeError as error: - try: - bme280 = adafruit_bme280.Adafruit_BME280_I2C(i2c, address=0x77) - print("BME280 sensor found with address 0x77") - except RuntimeError as failure: - print(error) - print(failure) - print("We do not have a environmental sensor") - - return disp, bme280 diff --git a/defaults.ini b/defaults.ini index 1f3731f..7a00781 100644 --- a/defaults.ini +++ b/defaults.ini @@ -1,45 +1,61 @@ # -# SBCEye Default Settings +# SBCEye Default Settings # Copy this file ('defaults.ini') to 'settings.ini' and edit # +# NOTE!: when specifying comma seperated lists, do not pad the commas with +# space characters!, eg. use '0,one,2,three'; not '0, one, 2, three' -[general] # General Settings # name: Used in logs and web, blank = hostname # long_format: Long time format for web, etc. time.strftime() # short_format: Short time format for logs, etc. time.strftime() -# log_hourly: Log a 'heartbeat'every hour? True/False` -# sensor: look for BME280 sensor? True/False -# screen: look for OLED display? True/False +# log_daily: Log a 'heartbeat'every day? True/False` +# sensor: Look for BME280 sensor? True/False +# display: Look for OLED display? True/False # pin_state_names: Localisation text for pin status (text,text) # -name = +[general] +name = My Awesome SBC long_format = %H:%M:%S, %A, %d %B, %Y short_format = %d-%m-%Y %H:%M:%S -log_hourly = True -screen = False +log_daily = True +display = False sensor = False pin_state_names = Off,On # # Internal Webserver -[web] # host: Ip address to bind server to, blank = bind to all addresses # port: Port number for web server -# sensor_name: Environmental sensor name used in web panel (eg; location) -# allow_dump: Allows RRDB database dumps, be careful using this +# sensor_name: Sensor name used in web panel (eg; location) +# allow_dump: Allows RRDB database dumps via we:, be careful using this +# since it can impact performance while running +# allow_backup: Allows RRDB database backup triggering via web, be careful using this # since it can impact performance while running -# show_control: Include a link to the web pin control 'button' # +[web] host = port = 7080 sensor_name = Room allow_dump = True -show_control = True +allow_backup = True + +# Time intervals (seconds) for the main system action schedules +# data: Interval between main reading updates +# pin: Pins are checked for state changes this frequently +# rrd: Maximum age before cached RRD database updates are written +# ping: Timout for ping responses +# - must be > 4 to distinguish 'unavailable' vs 'not responding' in logs +# - will be reduced if it exceeds the data interval (above) -0.5s +# +[intervals] +data = 10 +pin = 1 +rrd = 300 +ping = 4 -[graph] # RRD graphing options # durations: Default graph durations # wide: Total width in pixels @@ -51,93 +67,102 @@ show_control = True # half_height: Graphs matching this prefix will be drawn half-height # # default durations are mapped in rrd graph as: start='end-', 'end=now' -# for more details on how to specify the range the rrd documentation. +# for more details on how to specify the range see the rrd documentation. # https://oss.oetiker.ch/rrdtool/doc/rrdfetch.en.html#TIME%20OFFSET%20SPECIFICATION # Note: 'm' means Minutes for large values, but Months for values from 1 to 5. # +[graph] durations = 600s,1h,3h,1d,3d,1w,1m,3m,1y,3y -wide = 1200 -high = 300 +wide = 1600 +high = 400 line_color = #00A0A0 line_width = 2 area_color = #D0E0E0#FFFFFF area_depth = 0 half_height = pin,net +# List of fixed links for the main page's footer (optional) +# Give one link per line: +# link_name = URL +# Underscores in the name will be converted to spaces when displayed +# eg: +# [links] +# Homepage = https://www.example.org/ +# Webcam_Viewer = https://www.example.org/webcam/?action=stream +# Cool stuff = http://some.thing.cool/ # -# GPIO +[links] -[pins] # GPIO pins to monitor and log -# All pins are defined using BCM GPIO numbering -# https://www.raspberrypi.org/forums/viewtopic.php?t=105200 -# Try running `gpio readall` on the Pi itself for a map +# pins are specified with their gpio chip device and index. +# - in general the gpiochip0 pin index numbers correspond to +# the BCM pin numbering scheme used in older Pi libraries. +# - for other architectures you need to reference the hardware +# documentation for your Board. +# - see https://raspberrypi.stackexchange.com/a/147309 +# - an empty list disables pin monitoring +# # Pins are listed one per line: -# = BCM pin number +# Pin Name = gpiochip_device_path,line_offset_number # eg: -# Lamp = 7 -# Ventilator = 8 -# Printer = 25 +# [pins] +# Lamp = /dev/gpiochip0,7 +# Ventilator = /dev/gpiochip0,8 +# Printer = /dev/gpiochip0,25 # +[pins] -[button] -# We can control one pin via a either the web ui, and/or a physical button -# out: bcm pin number we want to control, '0' to disable -# pin: bcm pin number of button, '0' to disable button -# url: / for web control, blank to disable -# hold: Minimum press time, seconds (float), debounces and decouples the button +# Enable web output control for pins from the above list +# List the pin name and a (comma seperated) list of CIDR address blocks +# to allow control from. +# - Restrict to a specific IP address with a CIDR block of 32 +# - or to a local subnet with a CIDR block of 24 +# - IPV6 notation should also work (untested) +# - See: https://en.wikipedia.org/wiki/Classless_Inter-Domain_Routing # eg: -# out = 7 -# pin = 27 -# url = lamp +# [webpins] +# Lamp = 10.0.0.120/24 +# Ventilator = 10.0.0.120/32,10.0.0.130/32 # -out = 0 -pin = 0 -url = -hold = 0.250 +[webpins] +# Enable input (button) on/off control for pins from the above list +# Each pin to be controlled is listed along with the input pin +# chip, line offset, trigger and debounce details +# +# pin = device,offset,trigger,debounce +# The triger can either be 'rising' or 'falling' +# Debounce is in milliseconds, a good default is 66ms +# - a long debounce makes a 'press and hold' effect +# eg: +# [buttons] +# lamp = /dev/gpiochip0,27,rising,66 +# Ventilator = /dev/gpiochip0,26,rising,1000 # -# Connectivity +[buttons] -[ping] -# A list of targets to be used for network connectivity (ping) tests +# An optional list of targets to be used for network connectivity (ping) tests # Targets are listed one per line: -# = ip address +# = ip address # eg: +# [ping] # router = 192.168.0.1 # internet = 8.8.8.8 # +[ping] -# -# Data Logging and Recording Settings - -[intervals] -# Time intervals (seconds) for the main system action schedules -# pin: Pins are checked for state changes this frequently -# data: Interval between main reading updates -# rrd: Maximum age before cached RRD database updates are written -# ping: Timout for ping responses -# - must be > 4 to distinguish 'unavailable' vs 'not responding' in logs -# - will be reduced if it exceeds the data interval (above) -0.5s -# -pin = 2 -data = 10 -rrd = 300 -ping = 4 - -[log] -# Logging +# Action Logging # file_dir: Folder must be writable by the SBCEye process # file_name: .log # file_count: Maximum number of old logfiles to retain # file_size: Maximum size before logfile rolls over (Kb) # +[log] file_dir = ./data file_name = SBCEye.log file_count = 3 file_size = 1024 -[rrd] # RRD database # dir: Folder must be writable by SBCEye process # file_name: Database file name; .rrd @@ -147,43 +172,68 @@ file_size = 1024 # if that breaks the backup_count limit. (Days) # backup_time: Time of daily backup; HH:MM # +[rrd] dir = ./data file_name = SBCEye.rrd backup_count = 10 backup_age = 7 backup_time = 23:45 -# -# OLED Status dsplay options +# I2C Bus options +# only used if display or sensor enabled in [general] config +# bus_id: The I2C bus to use, integer +# sensor_addr: I2C address of the sensor +# display_addr: I2C address of the display +# lock: device,pin,pin +# optional; 'consumes' the pins on the gpiochip device. +# The pin offsets used by the device must be known +# and will vary between different devices and hardware. +# For a Raspberry PI model3 using default I2C bus 1: +# lock = /dev/gpiochip,0,1 +# +[bus] +bus_id = 1 +sensor_addr = 0x76 +display_addr = 0x3c +lock = + +# The following display, saver and animate options are only used +# if display is enabled in [general] config -[display] -# Display orientation, contrast and burn-in prevention +# OLED Status dsplay options # rotate: Is the display 'upside down'? # - generally the connections from the glass are at the bottom # contrast: This gives a limited brightness reduction (0-255) # - does not give full dimming to black -# invert: Default is light text on dark background # +[display] rotate= False contrast = 127 -invert = False -[saver] # Screen saver / burn-in reducer -# saver_mode: Possible values are 'off', 'blank' and 'invert' +# saver_mode: Possible values are 'disabled' or 'blank' # saver_on: Start time for screensaver (hour, 24hr clock) # saver_off: End time # -mode = off +[saver] +mode = disabled on = 20 off = 8 -[animate] # Display Animation # passtime: time between display refreshes (seconds) # passes: number of refreshes of a screen before moving to next # speed: rows to scroll on each animation step between screens # +[animate] passtime = 3 passes = 2 speed = 16 + +# Enable extra debug +# http: Log http requests to console (or syslog when run as a service) +# sigint: restart on SIGINT instead of exiting +# +[debug] +http = False +sigint = False diff --git a/gpiohandler.py b/gpiohandler.py new file mode 100644 index 0000000..0d2f9d5 --- /dev/null +++ b/gpiohandler.py @@ -0,0 +1,184 @@ +'''Monitor, record and log GPIO pin changes using gpiod + +provides: + GPIOReader: A class to update and log the pin statuses +''' + +import gpiod +import logging +from os import getpid +from re import search + +# Needs gpiod bindings at V2.0 or later, standard debian12/bookworm is v1.6 +# use a virtualenv and 'pip install --upgrade gpiod' as needed. +if int(search('^[0-9]+', gpiod.__version__).group(0)) < 2: + readerfail = 'gpiod bindings library version too low ({}), '\ + 'pinreader requires gpiod v2.x.x or later.'\ + .format(gpiod.__version__) +else: + readerfail = None + from pinreader import PinReader + from watch_button import buttonHandler + +''' +PinReader class (dict) +''' +INPUT = gpiod.line.Direction.INPUT +OUTPUT = gpiod.line.Direction.OUTPUT +ACTIVE = gpiod.line.Value.ACTIVE +INACTIVE = gpiod.line.Value.INACTIVE + + +class GPIOHandler: + '''Read and update GPIO pin status + + Reads the currrent (boolean) status of a set of gipo pins defined in a dictionary + Updates the relevant entries in data{} and logs state changes + + parameters: + pinlist: dictionary of pin definitions {label: (chip,index)} from config, + this will be passed directly to the PinReader init. + data: the main data{} dictionary, a key/value pair; 'pin-=value' + will be added to it and the value updated with pin state changes. + + provides: + update(): processes and updates the pin data, logs state changes + set_pin(pin,value): attempts to set the output value of a pin + - returns success/fail, used for web control + ''' + + def __init__(self, pinlist, buttons, data, consumer): + '''Setup and do initial reading''' + self.available = False + self.pinlist = pinlist + self.pins = {} + self.data = data + self._consumer = consumer + if not self.pinlist: + print('No GPIO pins configured for monitoring') + return + if readerfail: + print('GPIOD bindings could not be imported: {}\nPin monitoring disabled' + .format(readerfail)) + logging.warning('GPIO pins were specified but pinreader failed to import, see syslog') + logging.info('Pin monitoring disabled') + return + try: + self.pins = PinReader(self.pinlist) + except Exception as e: + print('GPIO pin setup failed: {}\nPin monitoring disabled.'.format(e)) + logging.warning('GPIO pins were specified but pin setup failed, see syslog') + logging.info('Pin monitoring disabled') + return + self.directions = {} # used to remember directions for change detection + self.consumers = {} # used to remember consumers for change detection + for pin in self.pins: + data[f'pin-{pin}'] = self._value_to_data(self.pins[pin].value) + self.directions[pin] = self.pins[pin].direction + self.consumers[pin] = self.pins[pin].consumer + print('Pin \'{}\': {}'.format(pin, repr(self.pins[pin])[10:-1])) + logging.info('Pin \'{}\': {}'.format(pin, repr(self.pins[pin])[10:-1])) + self.buttons = buttonHandler(buttons, self) + print('GPIO monitoring configured') + logging.info('GPIO monitoring configured') + self.available = True + + def _value_to_data(self, value): + '''Records failed reads as 'U' (unavailable) for data{} entries''' + return 'U' if value is None else int(value) + + def _update_pin(self, pin): + '''Check if pin has changed state, log changes and + update the data{} dictionary with new state''' + log = '' + direction = self.pins[pin].direction + if direction != self.directions[pin]: + self.directions[pin] = direction + log += ' direction to: \'{}\','.format(direction) + consumer = self.pins[pin].consumer + if consumer != self.consumers[pin]: + self.consumers[pin] = consumer + consumer = None if consumer is None else '\'{}\''.format(consumer) + log += ' consumer to: {},'.format(consumer) + value = self.pins[pin].value + if self._value_to_data(value) != self.data[f'pin-{pin}']: + self.data[f'pin-{pin}'] = self._value_to_data(value) + log += ' value to: {},'.format(value) + if len(log) != 0: + logging.info('Pin \'{}\' changed{}'.format(pin, log.rstrip(','))) + print('Pin \'{}\' changed{}'.format(pin, log.rstrip(','))) + + def update(self): + '''Update data{} dictionary for all pins''' + self.pins.update() + for pin in self.pins: + self._update_pin(pin) + + def setPin(self, pin, value): + '''Sets the pin to output mode and sets it's value. + Parameters: + pin: the pin name from config + value: (int) 1 = Active, 0 = Inactive + Returns: + False if the output cannot be set.''' + if value == 0: + value = INACTIVE + elif value == 1: + value = ACTIVE + else: + raise ValueError('Invalid output value: {} ({})' + .format(value, type(value))) + self.pins.update(pin) + chip = self.pins[pin].chip + line = self.pins[pin].line + newval = None + if self.pins[pin].consumer is not None: + logging.warning('Failed to set ouput on pin \'{}\', currently used by: \'{}\'' + .format(pin, self.pins[pin].consumer)) + else: + try: + with gpiod.Chip(chip).request_lines( + consumer=self._consumer, + config={line: gpiod.LineSettings( + direction = OUTPUT, + output_value = value)}, + ) as request: + newval = request.get_values()[0] + except OSError as e: + # log a warning and return 'False' if the write fails + logging.warning('Could not set output value on pin {}:{} : {}' + .format(chip, line, e)) + self.pins.update(pin) + self._update_pin(pin) + return True if newval == value else False + + def makeInput(self, pin): + '''Sets the pin to input mode and reads it's value to data{} dictionary. + Parameters: + pin: the pin name from config''' + self.pins.update(pin) + chip = self.pins[pin].chip + line = self.pins[pin].line + if self.pins[pin].consumer is not None: + logging.warning('Failed to set input mode on pin \'{}\', currently used by: \'{}\'' + .format(pin, self.pins[pin].consumer)) + else: + try: + with gpiod.Chip(chip).request_lines( + consumer=self._consumer, + config={line: gpiod.LineSettings( + direction = INPUT)}, + ) as request: + _ = request.get_values()[0] + except OSError as e: + # log a warning and return 'False' if the write fails + logging.warning('Could not set input mode on pin {}:{} : {}' + .format(chip, line, e)) + self.pins.update(pin) + self._update_pin(pin) + return + +if __name__ == "__main__": + from sys import exit + print('gpioHandler class for SBCEye, see inline docs') + exit() diff --git a/httpserver.py b/httpserver.py index 998b465..3527d72 100644 --- a/httpserver.py +++ b/httpserver.py @@ -12,12 +12,13 @@ # HTTP server import http.server from urllib.parse import urlparse, parse_qs +from ipaddress import ip_address, ip_network from threading import Thread # Logging import logging -def serve_http(settings, rrd, data, helpers): +def serve_http(settings, rrd, gpio, data): '''Spawns a http.server.HTTPServer in a separate thread on the given port''' handler = _BaseRequestHandler httpd = http.server.ThreadingHTTPServer((settings.web_host, settings.web_port), handler, False) @@ -30,30 +31,50 @@ def serve_http(settings, rrd, data, helpers): # there is probably a better way to do this, eg using a meta-class and inheritance http.settings = settings http.rrd = rrd + http.gpio = gpio http.data = data - http.button_control = helpers[0] http.icon_file = 'favicon.ico' if not os.path.exists(http.icon_file): http.icon_file = f'{sys.path[0]}/{http.icon_file}' - if rrd.rrdtool: + if rrd.rrdtool and rrd.gzip: http.db_graphable = True if settings.web_allow_dump: logging.info("RRD database is dumpable via web") http.db_dumpable = True else: http.db_dumpable = False + if settings.web_allow_backup: + logging.info("RRD database backups can be triggered via web") + http.db_backupable = True + else: + http.db_backupable = False else: - logging.warning('Commandline rrdtool not found, '\ - 'graphing and dumping functions are unavailable') + logging.warning('Commandline rrdtool or gzip not found, '\ + 'graphing, backup and dumping functions are unavailable') http.db_dumpable = False http.db_graphable = False + # Note the list of link targets + for link in settings.links: + logging.info(f"Web link '{link}' points to: {settings.links[link]}") + + # Note the controllable pins + for pin in list(settings.webpins): + if pin not in gpio.pins.keys(): + print('Cannot configure web control for \'{}\', since it is not '\ + 'in the pin list'.format(pin)) + del settings.webpins[pin] + print('Pin \'{}\' controllable via web UI from: {}'\ + .format(pin, settings.webpins[pin])) + logging.info('Pin \'{}\' controllable via web UI from: {}'\ + .format(pin, settings.webpins[pin])) + # Start the server logging.info(f'HTTP server will bind to port {str(settings.web_port)} '\ f'on host {settings.web_host}') httpd.server_bind() address = f"http://{httpd.server_name}:{httpd.server_port}" - print(f"Webserver starting on : {address}") + print(f"Webserver starting on : {address}",flush=True) httpd.server_activate() def serve_forever(httpd): @@ -63,19 +84,34 @@ def serve_forever(httpd): logging.info("Http Server closing down") thread = Thread(target=serve_forever, args=(httpd, )) - thread.setDaemon(True ) + thread.daemon = True thread.start() - class _BaseRequestHandler(http.server.BaseHTTPRequestHandler): '''Handles each individual request in a new thread''' - def _set_headers(self): + def log_message(self, format, *args): + # This function effectively suppresses the http server log output + if http.settings.debug_http: + print(f'HTTP request:: {self.client_address[0]} : {args[0]} '\ + f'({args[1]})',flush=True) + return + + + def _common_headers(self): self.send_response(200) + self.send_header('Cache-Control', 'no-cache, no-store, must-revalidate') + self.send_header('Pragma', 'no-cache') + self.send_header('Expires', '0') + + def _redirect(self): + self._common_headers() + self.send_header('refresh', '0; url=./') + self.end_headers() + + def _set_headers(self): + self._common_headers() self.send_header("Content-type", "text/html") - self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") - self.send_header("Pragma", "no-cache") - self.send_header("Expires", "0") self.end_headers() def _set_png_headers(self): @@ -190,18 +226,18 @@ def _give_sys(self): def _give_net(self): # Network Connectivity ret = '' - netlist = {} + targetlist = {} for key in http.data.keys(): if key[0:4] == 'net-': - netlist[key] = key[4:] - if len(http.data.keys() & netlist.keys()) > 0: + targetlist[key] = key[4:] + if len(http.data.keys() & targetlist.keys()) > 0: ret += 'Ping\n' - for item,name in netlist.items(): - ret += f'{name}:' + for item,name in targetlist.items(): + ret += f'{name}:' if http.data[item] == 'U': - ret += 'Fail\n' + ret += 'Fail\n' else: - ret += f'{http.data[item]:.1f}'\ + ret += f'{http.data[item]:.1f}'\ ''\ ' ms'\ '\n' @@ -217,9 +253,29 @@ def _give_pins(self): pinlist[key] = key[4:] if len(http.data.keys() & pinlist.keys()) > 0: ret += 'GPIO\n' - for item,name in pinlist.items(): - ret += f'{name}:'\ - f'{http.settings.pin_state_names[http.data[item]]}\n' + for item, name in pinlist.items(): + title = '{}:\n chip: {}\n line: {}\n direction: {}\n consumer: {}'.format( + name, http.gpio.pins[name].chip, http.gpio.pins[name].line, + http.gpio.pins[name].direction, http.gpio.pins[name].consumer) + ret += f'{name}:' + direction = '({})'.format(http.gpio.pins[name].direction[:-3]) + consumer = '{}'.format(http.gpio.pins[name].consumer) + if http.data[item] == 'U': + ret += '{}'.format(consumer) + direction = '' + else: + em = 'font-weight: bold' if http.data[item] == 1 else '' + if name in http.settings.webpins.keys(): + link = 'href="./{}" title="Pin Control" '\ + 'style="text-decoration: underline; {}"'.format(name, em) + ret += '{}'\ + .format(link, http.settings.pin_state_names[http.data[item]]) + else: + ret += '{}'\ + .format(em, http.settings.pin_state_names[http.data[item]]) + ret += ''\ + '{}\n'.format(direction) return ret def _give_graphlinks(self, skip=""): @@ -229,33 +285,34 @@ def _give_graphlinks(self, skip=""): if (len(http.settings.graph_durations) > 0) and http.db_graphable: if len(skip) == 0: ret += 'Graphs\n' - ret += '\n' + ret += '\n' for duration in http.settings.graph_durations: if duration != skip: - ret += f' '\ f'{duration} \n' else: - ret += f' {duration} \n' + ret += f'{duration} \n' if len(skip) > 0: - ret += ' :  Home\n' + ret += '\n'\ + 'Home\n' ret += '\n' return ret def _give_links(self): - # Link to the log and pin contol pages - ret = f'''{self._give_graphlinks()} - - - Log\n''' - if http.settings.web_show_control and (http.settings.button_pin > 0): - ret += f'  '\ - f'{http.settings.button_name}\n' - ret += '\n' + # Links to the graph pages + ret = f'{self._give_graphlinks()}' + # Configured links and log page + for link in http.settings.links: + ret += f''\ + f''\ + f'{link}\n' + ret += f'\n'\ + f''\ + f'Action Log\n' return ret - def _give_log(self, lines=25): + def _give_log(self, lines=32): # Combine and give last (lines) lines of log parsed_lines = parse_qs(urlparse(self.path).query).get('lines', None) if isinstance(parsed_lines, list): @@ -265,8 +322,8 @@ def _give_log(self, lines=25): try: lines = int(lines) except ValueError: - lines = int(100) - lines = max(1, min(lines, 250000)) + lines = int(32) + lines = max(1, lines) # Use a shell one-liner used to extract the last {lines} of data from the logs # There is doubtless a more 'python' way to do this, but it is fast, cheap and works.. log_command = \ @@ -278,10 +335,10 @@ def _give_log(self, lines=25): 
\n{log}
Latest {lines} lines shown\n \n -
25 : -  250 : -  2500 : -  Home
\n''' +
32 : + 320 : + 3200
\n + \n''' return ret def _give_graphs(self, start, end, stamp): @@ -318,6 +375,42 @@ def _give_dump_portal(self): ''' + def _give_pin_portal(self, pin, control): + ret = '

{} Pin Control

\n'.format(http.settings.name) + ret += '
{}: '.format(pin) + if http.gpio.pins[pin].value is None: + ret += 'used
\n'.format(http.gpio.pins[pin].consumer) + ret += '
consumed by: \'' + ret += '{}\'
\n'\ + .format(http.gpio.pins[pin].consumer) + else: + ret += '{}\n'\ + .format(http.settings.pin_state_names[http.gpio.pins[pin].value]) + ret += '
mode: {}
\n'\ + .format(http.gpio.pins[pin].direction) + if http.gpio.pins[pin].value is None: + ret += '
'\ + 'Pins used (consumed) by other processes cannot be controlled
' + elif control and http.gpio.pins[pin].direction == 'input': + for state in (0, 1): + ret += '\n'\ + .format(http.settings.pin_state_names[state]) + elif control: + newstate = 1 if http.gpio.pins[pin].value == 0 else 0 + ret += '\n'\ + .format(http.settings.pin_state_names[newstate]) + ret += '\n' + else: + ret += '
'\ + 'Client is not authorised to control pin
' + ret += '\n' + return ret + def _write_dedented(self, html): # Strip leading whitespace and write response = re.sub(r'^\s*','', html, flags=re.MULTILINE) @@ -382,38 +475,11 @@ def do_GET(self): self._set_icon_headers() with open(http.icon_file,'rb') as favicon: self.wfile.write(favicon.read()) - elif ((urlparse(self.path).path == '/' + http.settings.button_url) - and (len(http.settings.button_url) > 0) - and (http.settings.button_out > 0)): - # Web button control - parsed = parse_qs(urlparse(self.path).query).get('state', ['status']) - action = parsed[0] - if action != 'status': - logging.info(f'Web button triggered by: {self.client_address[0]}'\ - f' with action: {action}') - status, state = http.button_control(action) - self._set_headers() - response = self._give_head(f" :: {http.settings.button_name}") - response += f'

{status}

\n' - invert_state = http.settings.pin_state_names[not state] - response += f'''\n''' - response += '
\n'\ - 'Home
\n' - response += self._give_timestamp() - response += '\n' - response += self._give_foot() - self._write_dedented(response) elif (urlparse(self.path).path == '/dump_gz') and http.db_dumpable: # Raw dump download start = time.time() logging.info(f"RRD database dump requested by {self.client_address[0]}") - response = http.rrd.dump() + response = http.rrd.dump(reason=f'Web ({self.client_address[0]})') self._set_download_headers(len(response), f'{http.settings.name}-rrd-{time.strftime("%Y%m%d-%H%M%S")}.xml.gz') self.wfile.write(response) @@ -425,26 +491,70 @@ def do_GET(self): response += self._give_dump_portal() response += self._give_foot() self._write_dedented(response) + elif (urlparse(self.path).path == '/backup') and http.db_backupable: + # trigger a backup and notify + logging.info(f"RRD database backup triggered by {self.client_address[0]}") + self.send_response(302) + self.send_header('Location','log') + self.end_headers() + http.rrd.backup() elif urlparse(self.path).path == '/log': self._set_headers() - response = self._give_head() + response = self._give_head(" :: Logfile Viewer") response += f'

{http.settings.name} Log

\n' response += self._give_log() response += self._give_timestamp() response += self._give_foot(refresh=60, scroll=True) self._write_dedented(response) + elif urlparse(self.path).path[1:] in http.settings.webpins.keys(): + pin = urlparse(self.path).path[1:] + parsed_action = urlparse(self.path).query + action = parsed_action.casefold() + control = False + if http.gpio.pins[pin].value is not None: + for cidr in http.settings.webpins[pin]: + if ip_address(self.client_address[0]) in ip_network(cidr,strict=False): + control = True + if not control and action != '': + self.send_error(403, 'Forbidden', + 'Your IP address is not permitted to control this pin') + print('Denied access to \'/{}\' from client at IP: {}'\ + .format(pin, self.client_address[0])) + return + if action == http.settings.pin_state_names[0].casefold(): + http.gpio.setPin(pin, 0) + logging.info('Pin \'{}\' set output: {} via web ({})' + .format(pin, http.settings.pin_state_names[0], self.client_address[0])) + self._redirect() + elif action == http.settings.pin_state_names[1].casefold(): + http.gpio.setPin(pin, 1) + logging.info('Pin \'{}\' set output: {} via web ({})' + .format(pin, http.settings.pin_state_names[1], self.client_address[0])) + self._redirect() + elif action == 'input': + http.gpio.makeInput(pin) + logging.info('Pin \'{}\' set to input mode via web ({})' + .format(pin, self.client_address[0])) + self._redirect() + elif action != '': + self.send_error(418, 'I\'m a {}, '\ + 'I do not know how to \'{}\''\ + .format(pin, parsed_action)) + else: + self._set_headers() + response = self._give_head(" :: Pin Control :: {}".format(pin)) + response += self._give_pin_portal(pin, control) + response += self._give_timestamp() + response += self._give_foot(refresh=60) + self._write_dedented(response) elif urlparse(self.path).path == '/': # Main Page - cam = parse_qs(urlparse(self.path).query).get('cam', None) exclude = parse_qs(urlparse(self.path).query).get('exclude', '') exclude = [item for sublist in exclude for item in sublist.split(',')] self._set_headers() response = self._give_head() if not "deco" in exclude: response += f'

{http.settings.name}

\n' - if cam and http.settings.cam_url: - response += f'Webcam\n' response += '\n' if not "env" in exclude: response += self._give_env() @@ -468,3 +578,9 @@ def do_GET(self): def do_HEAD(self): '''returns headers''' self._set_headers() + +if __name__ == "__main__": + from sys import exit + print('HTTPserver class for SBCEye, see inline docs') + exit() + diff --git a/i2c_bus.py b/i2c_bus.py new file mode 100644 index 0000000..7e487f6 --- /dev/null +++ b/i2c_bus.py @@ -0,0 +1,73 @@ +'''I2C hardware driver initialisation + +Needed by the optional Bus based devices. +Gracefully fails and disables services as appropriate if anything goes wrong +''' +# pragma pylint: disable=import-outside-toplevel + +def i2c_pin_consume(settings): + '''Consume the specified pins with a gpiod request to prevent + other gpio processes from modifying them''' + if len(settings.bus_lock) != 3: + return None + try: + import gpiod + except ImportError: + print('WARNING: Cannot lock I2C pins, gpiod library unavailable') + return None + chip = settings.bus_lock[0] + lines = tuple(settings.bus_lock[1:]) + if not gpiod.is_gpiochip_device(chip): + print('WARNING: Cannot lock I2C pins, \'{}\' is not a gpiodchip device'\ + .format(chip)) + return None + try: + lock = gpiod.request_lines(chip, + consumer='{}'.format(settings.identifier), + config={lines:None}) + except Exception as e: + print('WARNING: Cannot lock I2C pins, request failed:\n{}'\ + .format(e)) + return None + print('I2C bus pins locked (\'{}\': {})'.format(chip, lines)) + return lock + + +def i2c_setup(settings): + '''Import and start the I2C bus device + + parameters: + settings: settings (from load_config) + returns: + i2c: Bus driver object, or None if failed + ''' + + # Load the correct modules, be graceful if that fails + i2c = None + lock = None + if settings.have_display or settings.have_sensor: + # I2C Comms + # Uses standard SMBUS lib (currently smbus2) + try: + from smbus2 import SMBus + except ImportError as error: + print(error) + print("ERROR: I2C bus requirements (smbus2) not met", flush=True) + return None, None + # Lock the bus pins if necesscary + lock = i2c_pin_consume(settings) + # Now the I2C device driver object + try: + i2c = SMBus(settings.bus_id) + except ValueError as error: + print(error) + print("No I2C bus, display and sensor functions will be disabled", flush=True) + return None, None + print('I2C bus found', flush=True) + return i2c, lock + +if __name__ == "__main__": + from sys import exit + print('i2c bus object class for SBCEye, see inline docs') + exit() + diff --git a/load_config.py b/load_config.py index 274adfb..401823e 100644 --- a/load_config.py +++ b/load_config.py @@ -20,12 +20,23 @@ class Settings: No Methods Attributes: + appname: a short text identifier for the app Basically, look at the class and config.ini file, I'm not going to list and describe everything a second time here ;-) ''' - def __init__(self): + def __init__(self,appname=None): + def hexint(instring): + ''' Helper function so we can use '0xNN' hex values for some inputs''' + try: + outint = int(instring) + except ValueError: + outint = int(instring, 16) + return outint + + if sys.path[0] == '': + sys.path[0] = '.' self.my_version = check_output(["git", "describe", "--tags", "--always", "--dirty"], cwd=sys.path[0]).decode('ascii').strip() @@ -71,20 +82,24 @@ def __init__(self): print('\nERROR: Cannot find a configuration file, exiting') sys.exit() - config = configparser.RawConfigParser() config.optionxform = str config.read(config_file) + if appname: + self.identifier = '{}-{}'.format(appname, os.getpid()) + else: + self.identifier = os.getpid() + # Set attributes from .ini file general = config["general"] self.name = general.get("name") self.long_format = general.get("long_format") self.short_format = general.get("short_format") - self.log_hourly = general.getboolean("log_hourly") + self.log_daily = general.getboolean("log_daily") self.have_sensor = general.getboolean("sensor") - self.have_screen = general.getboolean("screen") + self.have_display = general.getboolean("display") self.pin_state_names = tuple(general.get("pin_state_names").split(',')) if self.name == "": self.name = f'{os.uname().nodename}' @@ -93,8 +108,9 @@ def __init__(self): self.web_host = web.get("host") self.web_port = web.getint("port") self.web_sensor_name = web.get("sensor_name") + self.web_show_cam = web.getboolean("show_cam") self.web_allow_dump = web.getboolean("allow_dump") - self.web_show_control = web.getboolean("show_control") + self.web_allow_backup = web.getboolean("allow_backup") graph = config["graph"] self.graph_durations = graph.get("durations").split(',') @@ -106,26 +122,34 @@ def __init__(self): self.graph_area_depth = graph.get("area_depth") self.graph_half_height = graph.get("half_height").split(',') - self.pin_map = {} - for pin in config["pins"]: - self.pin_map[pin] = config.getint("pins",pin) - - self.net_map = {} - for host in config["ping"]: - self.net_map[host] = config.get("ping",host) - - button = config["button"] - self.button_out = button.getint("out") - self.button_pin = button.getint("pin") - self.button_url = button.get("url") - self.button_hold = button.getfloat("hold") - if self.button_out == 0: - self.button_name = 'Undefined' - else: - self.button_name = f'gpio-{self.button_out}' - for name, pin in self.pin_map.items(): - if pin == self.button_out: - self.button_name = name + self.links= {} + links = config["links"] + for name in links: + real = name.replace('_',' ') + self.links[real] = links.get(name) + + self.pinlist = {} + pins = config["pins"] + for pin in pins: + self.pinlist[pin] = pins.get(pin).split(',') + self.pinlist[pin][1] = int(self.pinlist[pin][1]) + + self.webpins = {} + webpins = config["webpins"] + for pin in webpins: + self.webpins[pin] = webpins.get(pin).split(',') + + self.buttons = {} + buttons = config["buttons"] + for pin in buttons: + self.buttons[pin] = buttons.get(pin).split(',') + self.buttons[pin][1] = int(self.buttons[pin][1]) + self.buttons[pin][3] = int(self.buttons[pin][3]) + + self.netlist = {} + ping = config["ping"] + for host in ping: + self.netlist[host] = ping.get(host) intervals = config["intervals"] self.pin_interval = intervals.getint("pin") @@ -148,10 +172,15 @@ def __init__(self): self.rrd_backup_age = int(abs(rrd.getfloat("backup_age")) * 86400) self.rrd_backup_time = rrd.get("backup_time") + bus = config["bus"] + self.bus_id = hexint(bus.get("bus_id")) + self.sensor_addr = hexint(bus.get("sensor_addr")) + self.display_addr = hexint(bus.get("display_addr")) + self.bus_lock = bus.get("lock").split(',') + display = config["display"] self.display_rotate = display.getboolean("rotate") self.display_contrast = display.getint("contrast") - self.display_invert = display.getboolean("invert") saver = config["saver"] self.saver_mode = saver.get("mode") @@ -163,19 +192,14 @@ def __init__(self): self.animate_passes = animate.getint("passes") self.animate_speed = animate.getint("speed") - self.cam_url = None - if "webcam" in config: - cam = config["webcam"] - self.cam_url = cam.get("url") - self.cam_width = cam.getint("width", 50) - - # Optional [DEBUG] section can be enabled - # If this section is present it changes the operation of - # SIGINT (eg Ctrl-c) to restart the service, instead of exiting - # Currently has no other configurable items - if "debug" in config: - self.debug = True - else: - self.debug = False + debug = config["debug"] + self.debug_http = debug.getboolean("http") + self.debug_sigint = debug.getboolean("sigint") print("Settings loaded from configuration file successfully") + +if __name__ == "__main__": + from sys import exit + print('setting loader class for SBCEye, see inline docs') + exit() + diff --git a/netreader.py b/netreader.py index 705d63c..0e39661 100644 --- a/netreader.py +++ b/netreader.py @@ -17,7 +17,7 @@ class Netreader: parameters: settings: (tuple) consisting of: - map: (dict) UI name and IP address/name + list: (dict) UI name and IP address/name timeout: (int) Timout in seconds data: the main data{} dictionary, a key/value pair; 'net-=value' will be added to it and the vaue updated with pin state changes. @@ -27,14 +27,15 @@ class Netreader: ''' def __init__(self, settings, data): '''Setup and do initial reading''' - (self.map, self.timeout) = settings + (self.list, self.timeout) = settings self.states = {} - if not self.map: + if not self.list: print('No network addresses configured for monitoring') return - for name,_ in self.map.items(): + for name, target in self.list.items(): self.states[name] = "init" data[f'net-{name}'] = 'U' + print(f'Ping target \'{name}\': {target}') self.update(data) print('Network monitoring configured and logging enabled') logging.info('Network monitoring configured and logging enabled') @@ -48,7 +49,7 @@ def _ping_runner(self, target, data): data: the main data{} dictionary no return ''' - address = self.map[target] + address = self.list[target] key = f'net-{target}' (data[key], status) = ping_target(address,self.timeout) if status: @@ -65,7 +66,7 @@ def _ping_runner(self, target, data): def update(self, data): '''Test each target in parallel via threads''' threadlist = [] - for target,_ in self.map.items(): + for target,_ in self.list.items(): gatherer = threading.Thread(target=self._ping_runner, args=[target, data]) gatherer.start() threadlist.append(gatherer) diff --git a/oled_display.py b/oled_display.py new file mode 100644 index 0000000..c179b56 --- /dev/null +++ b/oled_display.py @@ -0,0 +1,54 @@ +'''Bus based environmental sensor initialisation +''' +# pragma pylint: disable=import-outside-toplevel + +import importlib.util + +def oled_setup(settings): + '''Import and initialise ssd1306 library, return the display object + uses the 'luma' library, which in turn requires smbus2 + + parameters: + settings: settings oject (from load_config) + returns: + disp: Display module object, or None if failed + ''' + + disp = None + if settings.have_display: + # I2C OLED Display + # Uses the ssd1306 driver and libs from the luma libraries + # https://github.com/rm-hull/luma.core + # https://github.com/rm-hull/luma.oled + # pip install luma.core luma.oled + try: + from luma.oled.device import ssd1306 + except ImportError as error: + print(error) + print("ERROR: ssd1306 display requirements not met", flush=True) + return None + + # for luma display rotation must be specified at init, + # (value from 0 to 3, rotating 90 degrees each step) + rotate = 2 if settings.display_rotate else 0 + try: + # Create the display object + disp = ssd1306(rotate=rotate) + print("SSD1306 i2c display found", flush=True) + except Exception as error: + print(error) + print("ERROR: SSD1306 i2c display failed to initialise, disabling", flush=True) + return None + + # check in advance that the full luma library is available, and not + # just the display driver. This is used by the animator + if not importlib.util.find_spec("luma"): + print("ERROR: Luma library not found, disabling display", flush=True) + return None + return disp + +if __name__ == "__main__": + from sys import exit + print('OLED display class for SBCEye, see inline docs') + exit() + diff --git a/pinreader.py b/pinreader.py index c400d8a..5125270 100644 --- a/pinreader.py +++ b/pinreader.py @@ -1,78 +1,106 @@ -'''Really simple and direct reading of BCM GPIO pins +# Needs gpiod bindings at V2.0 or later, standard debian12/bookworm is v1.6 +# use a virtualenv and 'pip install --upgrade gpiod' as needed. +import gpiod +from os import getpid -provides: - Pinreader: A class to update and log the pin statuses - get_pin(pin): reads a bcm gpio pin and returns it's raw value -''' +INPUT = gpiod.line.Direction.INPUT +ACTIVE = gpiod.line.Value.ACTIVE -import os -import logging - -GPIO_ROOT = '/sys/class/gpio' -export_handle = f'{GPIO_ROOT}/export' -unexport_handle = f'{GPIO_ROOT}/unexport' - -class Pinreader: - '''Read and update pin status +class PinReader(dict): + ''' + PinReader class (dict) + A class derived from a dict() that can record and update the status of a list of pins - Reads the currrent (boolean) status of a set of gipo pins defined in a dictionary - Updates the relevant entries in data{} and logs state changes + ''' + class _instance: + ''' + class that holds individual pin instance data and exposes an update() method for the pin + ''' + def __init__(self, chip, line): + ''' Creates an pin instance object + Takes the GPIO chip device specifier (path, or identifier with gpiod v2+) + and line offset ''' + self._pid = getpid() # record PID of process that called init() + if not gpiod.is_gpiochip_device(chip): + raise ValueError('\'{}\' is not a valid GPIO device'.format(chip)) + self.chip = chip + try: + self._chip = gpiod.Chip(chip) + except PermissionError as e: + raise PermissionError('Cannot access \'{}\': {}' .format(chip, e)) + if line < 0 or line >= self._chip.get_info().num_lines: + raise ValueError('Requested line ({}) outside range for \'{}\' ({} lines)' + .format(line, chip, self._chip.get_info().num_lines)) + self.line = line + info = self._chip.get_line_info(self.line) + self.name = info.name + self.get() - parameters: - settings: (tuple) consisting of: - map: (dict) pin names and BCM GPIO number - state_names: (tuple) localised names for pin states (text,text) - data: the main data{} dictionary, a key/value pair; 'pin-=value' - will be added to it and the vaue updated with pin state changes. + def __repr__(self): + ''' returns a formatted representation of the pin data ''' + consumer = None if self.consumer is None else '\'{}\''.format(self.consumer) + return '_instance(chip=\'{}\' line={} consumer={} direction=\'{}\' value={})'\ + .format(self.chip, self.line, consumer, self.direction, self.value) - provides: - update_pins(): processes and updates the pins - ''' + def __str__(self): + ''' returns a formatted string giving direction, consumer and value ''' + consumer = None if self.consumer is None else '\'{}\''.format(self.consumer) + return 'direction: {}, consumer: {}, value: {}'\ + .format(self.direction, consumer, self.value) - def __init__(self, settings, data): - '''Setup and do initial reading''' - (self.map, self.state_names) = settings - self.data = data - if not self.map: - print('No GPIO pins configured for monitoring') - return - for pin_name, pin_number in self.map.items(): - data[f'pin-{pin_name}'] = get_pin(pin_number) - logging.info(f'{pin_name}: {self.state_names[data[f"pin-{pin_name}"]]}') - print('GPIO monitoring configured and logging enabled') - logging.info('GPIO monitoring configured and logging enabled') + def _value(self): + ''' reads the value of the pin ''' + try: + with self._chip.request_lines(consumer='pinreader-{}'.format(self._pid), + config={self.line: None}) as request: + val = request.get_values()[0] + return 1 if val == ACTIVE else 0 + except OSError: + # silently return None if the read fails + # - there are possible race conditions if this pin is simultaneously + # accessed by another program (or instance of this class..) etc. + return None - def update_pins(self): - '''Check if any pins have changed state, and log if so - updates the main data{} dictionary with new state - no parameters, no return''' - for name, pin in self.map.items(): - this_pin_state = get_pin(pin) - if this_pin_state != self.data[f"pin-{name}"]: - # Pin has changed state, store new state and log - self.data[f'pin-{name}'] = this_pin_state - logging.info(f'{name}: {self.state_names[this_pin_state]}') + def get(self): + ''' reads and updates the current status of the pin, + the value is only read if the pin is unused ''' + line = self._chip.get_line_info(self.line) + self.direction = 'input' if line.direction == INPUT else 'output' + if line.used: + self.consumer = line.consumer + self.value = None + else: + self.consumer = None + self.value = self._value() + return self.value -def get_pin(pin): - '''Read pin state, return an integer + ''' class init() ''' + def __init__(self, pinlist, tolerant=False): + ''' Main class init(): + Takes a pin list: a dictionary{} of pin labels, each item being a tuple() + of gpio chip path, and line offset. + The 'tolerant' flag prevents exceptions if an invalid pin is specified ''' + super().__init__({}) + self.chips = [] + for label in pinlist: + try: + super().__setitem__(label, self._instance(*pinlist[label])) + except ValueError as e: + if not tolerant: + raise ValueError('failed to set up pin \'{}\': {}' + .format(label, e)) + else: + if pinlist[label][0] not in self.chips: + self.chips.append(pinlist[label][0]) - parameters: - pin: (int) the BCM gpio pin number + def update(self, items=None): + ''' Updates all pins (deault), a pin, or a list of pins ''' + items = [items] if type(items) == str else items + items = super().keys() if items is None else items + for line in items: + super().__getitem__(line).get() - returns: - pin_state: (int) 0=low, 1=high - ''' - gpio_handle = f'{GPIO_ROOT}/gpio{str(pin)}' - exported = os.path.isdir(gpio_handle) - if not exported: - export = os.open(export_handle, os.O_WRONLY) - os.write(export, bytes(str(pin), 'ascii')) - os.close(export) - value = os.open(f'{gpio_handle}/value', os.O_RDONLY) - ret = int(os.read(value,1)) - os.close(value) - if not exported: - unexport = os.open(unexport_handle, os.O_WRONLY) - os.write(unexport, bytes(str(pin), 'ascii')) - os.close(unexport) - return int(ret) +if __name__ == "__main__": + from sys import exit + print('PinReader{} class for SBCEye, see inline docs') + exit() diff --git a/robin.py b/robin.py index f01bd33..0a296a0 100644 --- a/robin.py +++ b/robin.py @@ -6,7 +6,6 @@ import time from pathlib import Path import logging -import gzip import subprocess import os from shutil import which @@ -37,7 +36,6 @@ def __init__(self, s, data): self.graph_args["area_depth"] = s.graph_area_depth self.half_height = s.graph_half_height - # Sensor and system sources with limits (min,max) self.data_sources = { 'env-temp': ('-40','80'), @@ -57,7 +55,7 @@ def __init__(self, s, data): # Graphs and parameters self.graph_map = { 'env-temp': (f'{s.web_sensor_name} Temperature, \u00B0Centigrade', - None, None, '%3.0lf\u00B0', '%3.1lf\u00B0C'), + None, None, '%3.1lf\u00B0', '%3.1lf\u00B0C'), 'env-humi': (f'{s.web_sensor_name} Humidity, % percent', None, None, '%3.0lf', '%3.0lf%%'), 'env-pres': (f'{s.web_sensor_name} Pressure, millibars', @@ -83,16 +81,16 @@ def __init__(self, s, data): None, None, '%5.0lf', '%5.0lf /s', '--units-exponent','0'), } # connectivity - for host in s.net_map.keys(): + for host in s.netlist.keys(): self.data_sources[f'net-{host}'] = ('0','U') self.graph_map[f'net-{host}'] = (f'{host} Ping, milliseconds', '25', '0' ,'%3.0lf', '%3.1lf ms', '--alt-autoscale', '--units-exponent','0') # pins - for name in s.pin_map.keys(): + for name in s.pinlist.keys(): self.data_sources[f'pin-{name}'] = ('0','1') self.graph_map[f'pin-{name}'] = (f'{name} Pin State, '\ - f'0 = {s.pin_state_names[0]}, 1 = {s.pin_state_names[1]}', + f'0 = {s.pin_state_names[0]}, 1 = {s.pin_state_names[1]}, None = pin n/a', '1', '0' ,'%3.1lf', '%3.0lf', '--alt-autoscale', '--units-exponent','0') # set the list of active and storable sources @@ -136,7 +134,7 @@ def __init__(self, s, data): mini = self.data_sources[source][0] maxi = self.data_sources[source][1] ds_list.append(f'DS:{source}:GAUGE:60:{mini}:{maxi}') - print(f" data source: {source} ({mini},{maxi})") + print(f" added data source: {source} ({mini},{maxi})") args = [str(self.db_file)] if source_file.is_file(): print(f'Importing from previous {source_file}') @@ -161,17 +159,21 @@ def __init__(self, s, data): if not source in existing_sources: mini = self.data_sources[source][0] maxi = self.data_sources[source][1] - print(f"Adding: {source} ({mini},{maxi}) to {self.db_file}") + print(f"Added data source: {source} ({mini},{maxi}) to {self.db_file}") rrdtool.tune( str(self.db_file), f"DS:{source}:GAUGE:60:{mini}:{maxi}") - # Disable dumping if rrdtool not in path + # Disable backup and dumping if rrdtool or gzip not in path self.rrdtool = which("rrdtool") - if self.rrdtool: + self.gzip = which("gzip") + if self.rrdtool and self.gzip: print(f'Commandline rrdtool: {self.rrdtool}') + print(f'Commandline gzip: {self.gzip}') else: - print('No commandline rrdtool available, ' + 'graphing and dumping disabled') + print('No commandline rrdtool and/or gzip available: backups disabled') + logging.warning('rrdtool or gzip not found: Disabling database backups') + self.backup_count = 0 # Use a home-brew local cache self.cache = [] @@ -179,29 +181,38 @@ def __init__(self, s, data): self.cache_age = s.rrd_interval # Notify - print('RRD database and cache configured and enabled') + print('RRD database and cache configured and enabled',flush=True) logging.info(f'RRD database is: {str(self.db_file)}') - def _backup(self): + def backup(self): '''Backup and rotate old backups''' if self.backup_count > 0: # Copy to a timestamped file self.write_updates() - suffix = time.strftime("%Y-%m-%d.%H:%M:%S.gz") - if not db_lock.acquire(blocking=True, timeout=600): - print('Error: Backup failed, could not acquire db lock within 600s') - return + suffix = time.strftime("%Y-%m-%d.%H:%M:%S.xml.gz") start = time.time() - with open(f'{self.db_file}', 'rb') as dbfile: - with gzip.GzipFile( - f'{str(self.backup_path)}/{self.backup_name}.{suffix}', - mode = 'wb', compresslevel = 6) as zipfile: - zipfile.write(dbfile.read()) - db_lock.release() - #logging.info(f'Database backup saved as: {self.backup_name}.{suffix}') - print(f'Database backup saved as: {self.backup_name}.{suffix} '\ - f'(took: {(time.time() - start):.2f}s)') + try: + backupfile = open(f'{str(self.backup_path)}/{self.backup_name}.{suffix}', 'wb') + except Exception as e: + logging.error(f'Database backup file write failed: {self.backup_name}.{suffix}\n{e}') + print(f'Database backup file write failed: {self.backup_name}.{suffix}\n{e}',flush=True) + return + else: + with backupfile: + backupfile.write(self.dump(reason='Backup')) + size = os.stat(f'{str(self.backup_path)}/{self.backup_name}.{suffix}').st_size + if size == 0: + logging.error(f'Database backup failed: empty datafile returned') + print(f'Database backup failed: empty datafile returned',flush=True) + try: + os.remove(f'{str(self.backup_path)}/{self.backup_name}.{suffix}') + except: + pass # ignore a failure here + return + logging.info(f'Database backup saved as: {self.backup_name}.{suffix} '\ + f'(size: {size}, took: {(time.time() - start):.2f}s)') + print(f'Database backup saved as: {self.backup_name}.{suffix}',flush=True) # Process old backups now = time.time() @@ -220,34 +231,33 @@ def _backup(self): else: os.remove(f'{self.backup_path}/{name}') #logging.info(f'Removed stale backup: {name}') - print(f'Removed stale backup: {name}') + print(f'Removed stale backup: {name}',flush=True) def start_backups(self): '''Add the backup schedule job''' # Start the backup schedule, using threads since it can run for some time if self.backup_count > 0: - schedule.every().day.at(self.backup_time).do(run_threaded, self._backup) + schedule.every().day.at(self.backup_time).do(run_threaded, self.backup) + - def dump(self): + def dump(self, reason=''): '''provide a gzipped dump of database''' dump_local.zipped = bytearray() - if self.rrdtool: + if self.rrdtool and self.gzip: self.write_updates() - print('Dump requested') + print('Dump requested: {}'.format(reason),flush=True) if not db_lock.acquire(blocking=True, timeout=60): - print('Error: Dumping failed, could not acquire db lock within 60s') + print('Error: Dumping failed, could not acquire db lock within 60s',flush=True) return dump_local.zipped dump_local.start = time.time() - dump = subprocess.check_output([self.rrdtool, 'dump', str(self.db_file)]) + with subprocess.Popen([self.rrdtool, 'dump', str(self.db_file)], \ + stdout=subprocess.PIPE) as dump_local.raw: + dump_local.zipped = subprocess.check_output(('gzip'), stdin=dump_local.raw.stdout) db_lock.release() - print(f'Dump is: {len(dump)} bytes raw and '\ - f'took {(time.time() - dump_local.start):.2f}s') - dump_local.start = time.time() - dump_local.zipped = gzip.compress(dump, compresslevel=6) - print(f'Dump compressed to {len(dump_local.zipped)} bytes '\ - f'in {(time.time() - dump_local.start):.2f}s') + print(f'Dump is: {len(dump_local.zipped)} bytes compressed, and '\ + f'took {(time.time() - dump_local.start):.2f}s',flush=True) else: - print('Dump requested but denied because commandline "rrdtool" unavailable') + print('Dump requested but denied because commandline "rrdtool" or "gzip" unavailable',flush=True) return dump_local.zipped def update(self, data): @@ -265,11 +275,11 @@ def write_updates(self): if len(self.cache) > 0: if not db_lock.acquire(blocking=True, timeout=self.cache_age): print('Error: Data Write failed, could not acquire database '\ - f'lock within write period ({self.cache_age}s)') + f'lock within write period ({self.cache_age}s)',flush=True) return # check if cache was emptied in another thread while waiting for lock if len(self.cache) > 0: - # print(f'DB WRITE:len={len(self.cache)}') + # print(f'DB WRITE:len={len(self.cache)}',flush=True) try: rrdtool.update( str(self.db_file), @@ -279,7 +289,7 @@ def write_updates(self): self.cache = [] except rrdtool.OperationalError as rrd_error: print("RRDTool update error:") - print(rrd_error) + print(rrd_error,flush=True) db_lock.release() self.last_write = time.time() @@ -331,12 +341,12 @@ def draw_graph(self, start, end, duration, graph): print(f'Graph generation failed:\n{graph_error}') print(f'cmd: {graph_error.cmd}') print(f'output: {graph_error.output}') - print(f'stdout: {graph_error.stderr}') + print(f'stdout: {graph_error.stderr}',flush=True) if len(graph_local.response) == 0: - print(f'Error: png file generation failed for : {graph} : {start}>>{end}') + print(f'Error: png file generation failed for : {graph} : {start}>>{end}',flush=True) else: - print(f'Error: No graph available for type: {graph}') + print(f'Error: No graph available for type: {graph}',flush=True) return graph_local.response def run_threaded(job_func): @@ -344,3 +354,9 @@ def run_threaded(job_func): ''' job_thread = Thread(target=job_func) job_thread.start() + +if __name__ == "__main__": + from sys import exit + print('robin (rrd handler) class for SBCEye, see inline docs') + exit() + diff --git a/saver.py b/saver.py index a2b62b9..ebbb659 100644 --- a/saver.py +++ b/saver.py @@ -1,45 +1,41 @@ -'''Implements a screen saver/inverter for the SBCEye project +'''Implements a screen saver for the SBCEye project ''' import time +import logging class Saver: '''Saver class: Turns the display on/off between specified times - Can also invert the display as a form of burn-in protection modes: - 'off': Screensaver disabled + 'disabled': Screensaver disabled 'blank': Turn screen off - 'invert': Invert the screen parameters: disp: display driver object settings: (tuple) consisting of: - mode: (str) One of 'off', 'blank', 'invert' + mode: (str) One of 'disabled', 'blank' start: (int) Start time, hour, 0-23 end: (int) End time, hour, 0-23 - invert: (bool) Base invert state for the display - Optional, defaults to False + + provides: + check(): Checks state and applies saver as needed + called from a schedule loop by main code ''' active = False # Current state def __init__(self, disp, settings): - self.disp = disp - (self.mode, start, end, *invert) = settings - self.invert = False - if len(invert) == 1: - self.invert = invert[0] - if self.mode != 'off': - print(f'Saver will {self.mode} display between: '\ - f'{start}:00 and {end}:00') + (self.mode, start, end) = settings + if self.mode != 'disabled': if (start == end)\ - or start not in range(0,23)\ - or end not in range(0,23): - print('start/end times identical or out of range; disabling') - self.mode = 'off' + or start not in range(0,23)\ + or end not in range(0,23): + logging.warning('Saver start/end times identical or out of range; disabling') + print('Saver start/end times identical or out of range; disabling',flush=True) + self.mode = 'disabled' elif start < end: self.saver_map = [False]*24 for i in range(start, end): @@ -48,6 +44,10 @@ def __init__(self, disp, settings): self.saver_map = [True]*24 for i in range(end, start): self.saver_map[i] = False + logging.info(f'Saver will {self.mode} display between: '\ + f'{start:02d}:00 and {end:02d}:00') + print(f'Saver will {self.mode} display between: '\ + f'{start:02d}:00 and {end:02d}:00',flush=True) self.check() @@ -55,23 +55,23 @@ def _apply_state(self, state): '''Apply the desired state to the display''' if state: self.active = True - print('Saver activated') - if self.mode == 'invert': - self.disp.invert(not self.invert) - elif self.mode == 'blank': - self.disp.poweroff() + if self.mode == 'blank': + print('Saver activated',flush=True) else: self.active = False - print('Saver deactivated') - if self.mode == 'invert': - self.disp.invert(self.invert) - elif self.mode == 'blank': - self.disp.poweron() + if self.mode == 'blank': + print('Saver deactivated',flush=True) def check(self): '''Check the current state vs the time, and apply changes as needed.''' - if self.mode != 'off': + if self.mode != 'disabled': hour = time.localtime()[3] if self.active != self.saver_map[hour]: self._apply_state(self.saver_map[hour]) + +if __name__ == "__main__": + from sys import exit + print('OLED screensaver class for SBCEye, see inline docs') + exit() + diff --git a/watch_button.py b/watch_button.py new file mode 100644 index 0000000..ba46717 --- /dev/null +++ b/watch_button.py @@ -0,0 +1,124 @@ +# Needs gpiod bindings at V2.0 or later, standard debian12/bookworm is v1.6 +# use a virtualenv and 'pip install --upgrade gpiod' as needed. +import gpiod +import logging +from datetime import timedelta +from threading import Thread +from os import getpid + +INPUT = gpiod.line.Direction.INPUT +RISING_EDGE = gpiod.edge_event.EdgeEvent.Type.RISING_EDGE +FALLING_EDGE = gpiod.edge_event.EdgeEvent.Type.FALLING_EDGE + +def _get_device(chip, line): + ''' Common gpiochip+line setup ''' + # Test whether chip is a gpiochip + if not gpiod.is_gpiochip_device(chip): + raise ValueError('Button setup: Not a gpiochip device: \'{}\''.format(chip)) + # Test we can use gpiochip + try: + device = gpiod.Chip(chip) + except Exception as e: + raise ValueError('Button setup: Failed to setup gpiochip (\'{}\') device:\n{}' + .format(chip, e)) + return device + + +# Input +class buttonHandler: + class _button: + ''' + Sets the input pin up to monitor for changes: + - pin will be used (consumed) by the process + Takes: + chip (str) + line (int) + consumer (string) + debounce (int) in ms + Provides: + event(): + blocks and waits for events + returns a string with 'rising' or 'falling' otherwise + ''' + def __init__(self, chip, line, consumer, debounce): + self.chip = chip + self.line = line + bias = gpiod.line.Bias.AS_IS + edge = gpiod.line.Edge.BOTH + clock = gpiod.line.Clock.MONOTONIC + debounce = timedelta(milliseconds=debounce) + # Get and test the device + device = _get_device(self.chip, self.line) + if device.get_line_info(line).used: + raise ValueError('Button setup: Cannot acquire gpiochip \'{}\''\ + 'line {}; currently used by: \'{}\'' + .format(chip, line, device.get_line_info(line).consumer)) + # Create a request object for the input + self.request = device.request_lines( + consumer=consumer, + config={line: gpiod.LineSettings( + direction=INPUT, + bias=bias, + edge_detection=edge, + event_clock=clock, + debounce_period=debounce)}) + + def event(self): + ''' (wait indefinately for and) return the first event in the queue ''' + self.request.wait_edge_events(timeout=None) + event = self.request.read_edge_events(max_events=1)[0] + if event.line_offset == self.line: + if event.event_type == RISING_EDGE: + return 'rising' + elif event.event_type == FALLING_EDGE: + return 'falling' + + def __init__(self, buttons, gpio): + ''' Creates button objects for all the specified pins + and spawns threads to monitor them and flip the pin + when the button is pressed ''' + self.gpio = gpio + self.watched = {} + self._threads = {} + for button in buttons: + if button not in gpio.pins.keys(): + print('Cannot configure button for undefined pin \'{}\'.' + .format(button)) + continue + self.watched[button] = self._button(chip=buttons[button][0], + line=buttons[button][1], + consumer=gpio._consumer, + debounce=buttons[button][3]) + self._threads[button] = Thread(target=self._serve_input, + args=(button, buttons[button][2])) + self._threads[button].daemon = True + self._threads[button].start() + print('Pin \'{}\' button configured on \'{}\':{} ({})' + .format(button, buttons[button][0], + buttons[button][1], buttons[button][2])) + logging.info('Pin \'{}\' button configured on \'{}\':{} ({})' + .format(button, buttons[button][0], + buttons[button][1], buttons[button][2])) + + def _flip(self, pin): + ''' A simple function to invert the output ''' + current = self.gpio.pins[pin].get() + if current == 0: + self.gpio.setPin(pin, 1) + elif current == 1: + self.gpio.setPin(pin, 0) + + def _serve_input(self, pin, edge): + '''service loop serving the input pin events (run in a thread)''' + while True: + event = self.watched[pin].event() + if event == edge: + self._flip(pin) + print('Button toggle for pin \'{}\''.format(pin), flush=True) + logging.info('Button toggle for pin \'{}\''.format(pin)) + +if __name__ == "__main__": + from sys import exit + print('ButtonWatcher class for SBCEye, see inline docs') + exit() +