Source code for alarmdecoder.util

Provides utility classes for the `AlarmDecoder`_ (AD2) devices.

.. _AlarmDecoder:

.. moduleauthor:: Scott Petersen <>

import time
import threading

[docs]class NoDeviceError(Exception): """ No devices found. """ pass
[docs]class CommError(Exception): """ There was an error communicating with the device. """ pass
[docs]class TimeoutError(Exception): """ There was a timeout while trying to communicate with the device. """ pass
[docs]class InvalidMessageError(Exception): """ The format of the panel message was invalid. """ pass
[docs]class Firmware(object): """ Represents firmware for the `AlarmDecoder`_ devices. """ # Constants STAGE_START = 0 STAGE_WAITING = 1 STAGE_BOOT = 2 STAGE_LOAD = 3 STAGE_UPLOADING = 4 STAGE_DONE = 5 # FIXME: Rewrite this monstrosity. @staticmethod
[docs] def upload(dev, filename, progress_callback=None): """ Uploads firmware to an `AlarmDecoder`_ device. :param filename: firmware filename :type filename: string :param progress_callback: callback function used to report progress :type progress_callback: function :raises: :py:class:`~alarmdecoder.util.NoDeviceError`, :py:class:`~alarmdecoder.util.TimeoutError` """ def do_upload(): """ Perform the actual firmware upload to the device. """ with open(filename) as upload_file: for line in upload_file: line = line.rstrip() if line[0] == ':': dev.write(line + "\r") dev.read_line(timeout=10.0) if progress_callback is not None: progress_callback(Firmware.STAGE_UPLOADING) time.sleep(0.0) def read_until(pattern, timeout=0.0): """ Read characters until a specific pattern is found or the timeout is hit. """ def timeout_event(): """Handles the read timeout event.""" timeout_event.reading = False timeout_event.reading = True timer = None if timeout > 0: timer = threading.Timer(timeout, timeout_event) timer.start() position = 0 while timeout_event.reading: try: char = if char is not None and char != '': if char == pattern[position]: position = position + 1 if position == len(pattern): break else: position = 0 except Exception: pass if timer: if timer.is_alive(): timer.cancel() else: raise TimeoutError('Timeout while waiting for line terminator.') def stage_callback(stage): """Callback to update progress for the specified stage.""" if progress_callback is not None: progress_callback(stage) if dev is None: raise NoDeviceError('No device specified for firmware upload.') stage_callback(Firmware.STAGE_START) if dev.is_reader_alive(): # Close the reader thread and wait for it to die, otherwise # it interferes with our reading. dev.stop_reader() while dev._read_thread.is_alive(): stage_callback(Firmware.STAGE_WAITING) time.sleep(1) time.sleep(2) # Reboot the device and wait for the boot loader. stage_callback(Firmware.STAGE_BOOT) dev.write("=") read_until('......', timeout=15.0) # Get ourselves into the boot loader and wait for indication # that it's ready for the firmware upload. stage_callback(Firmware.STAGE_LOAD) dev.write("=") read_until('!load', timeout=15.0) # And finally do the upload. do_upload() stage_callback(Firmware.STAGE_DONE)