[DSF-Python] Script to monitor SBC and log results through DSF

oozeBot

I couldn't find much info about dsf-python outside of the few included examples, so I thought I'd post the beginning of our SBC watchdog that will be scheduled through cron to run every few minutes.

The following code checks the SBC to see if it is experiencing (or has experienced) under-voltage, throttling, or reached temp limits. If so, it logs an error in DSF if the event is currently being experienced, or a warning if it has happened since last reboot.

We have migrated to Raspberry Pi OS Bullseye, which moved the location of vcgencmd. For older OS, you will likely need to adjust the path to '/opt/vc/bin/vcgencmd'.

There are two lines of DEBUG in SBCHealthCheck that can be used to test.

Also, this was literally my first crack with writing in Python, so don't critique me too hard.

#!/usr/bin/env python3

import subprocess

from dsf.connections import CommandConnection
from dsf.commands.basecommands import MessageType
from dsf.commands.basecommands import LogLevel

#0b01010000000000000000
#  01110000000000000010
#  ||||            ||||_ Under-voltage detected
#  ||||            |||_ Arm frequency capped
#  ||||            ||_ Currently throttled
#  ||||            |_ Soft temperature limit active
#  ||||_ Under-voltage has occurred since last reboot
#  |||_ Arm frequency capped has occurred
#  ||_ Throttling has occurred
#  |_ Soft temperature limit has occurred

MESSAGES = {
  0:  'SBC under-voltage detected!',
  1:  'SBC ARM frequency capped!',
  2:  'SBC currently throttled!',
  3:  'SBC soft temperature limit active',
  16: 'SBC under-voltage has occurred',
  17: 'SBC throttling has occurred',
  18: 'SBC ARM frequency capped has occurred',
  19: 'SBC soft temperature limit has occurred'
}

########################################################################################################

def run(command, print_output=False):
  from subprocess import Popen, PIPE, STDOUT
  from io import StringIO
  popen = Popen(command, stdout=PIPE, stderr=STDOUT, universal_newlines=True, shell=True)
  out = StringIO()
  for line in popen.stdout:
    if print_output:
      print(line, end='')
    else:
      out.write(line)
    popen.stdout.close()
    return_code = popen.wait()
    if not return_code == 0:
      raise RuntimeError('The process call "{}" returned with code {}. The return code is not 0, thus an error occurred.'.format(list(command), return_code))
    stdout_string = out.getvalue()
    out.close()
    return stdout_string

########################################################################################################

def SBCHealthCheck():
  res = run( "/usr/bin/vcgencmd get_throttled" )
  #res = "throttled=0x1"     #DEBUG - throws an error
  #res = "throttled=0x50000" #DEBUG - throws a warning
  throttled = bin(int(res.split('=')[1], 0))
  for position, message in MESSAGES.items():
    if len(throttled) > position and throttled[0 - position - 1] == '1':
      command_connection = CommandConnection()
      command_connection.connect()
      if position < 10:
        command_connection.write_message(MessageType.Error, message, True, LogLevel.Info)
      else:
        command_connection.write_message(MessageType.Warning, message, True, LogLevel.Info)
      command_connection.close()
      break

########################################################################################################

if __name__ == "__main__":
  SBCHealthCheck()

mfs12

@oozeBot, if you get more code which you consider good as an example i would be happy to add it to the example folders. Especially if it solves some new use case.