led-blink-seq

#!/usr/bin/env python3

import itertools as it, operator as op, functools as ft
from collections import deque
import os, sys, pathlib, time, logging, contextlib


class BlinkError(Exception): pass

class BlinkConfig:

	leds_root = '/sys/class/leds'
	gpio_root = '/sys/class/gpio'
	duration_factor = 1.0 # multiplies values written to "duration"
	bit_repr = 1300, 150, 700
	exit_state, exit_wait = 0, False

	def __init__(self, **opts):
		for k, v in opts.items(): setattr(self, k, v)


class LEDControlBase:

	def __init__(self, led, conf): pass
	def __enter__(self): pass
	def __exit__(self, *err): pass

class LEDControl(LEDControlBase):

	led_sysfs = led_gpio = None

	def __init__(self, led, conf):
		if isinstance(led, str): self.led_sysfs = pathlib.Path(conf.leds_root) / led
		else: self.led_gpio, self.gpio_root = led, pathlib.Path(conf.gpio_root)
		self.led_sysfs_k, self.led_exit_state = conf.duration_factor, conf.exit_state
		self.led_exit_wait = conf.exit_wait or self.led_exit_state

	def file_access_wrap(self, func, checks=12, timeout=1.0):
		for n in range(checks, -1, -1):
			try: return func()
			except OSError: pass
			if checks <= 0: break
			if n: time.sleep(timeout / checks)
		else: raise OSError('Access failed', func, timeout)

	def sysfs_node(self, n, mode='wb'):
		buff = 0 if 'b' in mode else -1
		return self.file_access_wrap(ft.partial(open, n, mode, buffering=buff))
	def sysfs_write(self, dst, data, end=b'\n'):
		return self.file_access_wrap(ft.partial(dst.write, data + (end or b'')))

	def __enter__(self):
		if self.led_sysfs:
			self.led_trigger = self.sysfs_node(self.led_sysfs / 'trigger')
			self.sysfs_write(self.led_trigger, b'transient')
			self.led_set, self.led_time, self.led_run = (
				self.sysfs_node(self.led_sysfs / k) for k in 'state duration activate'.split() )
			self.sysfs_write(self.led_set, b'1')
		else:
			p = self.gpio_root / f'gpio{self.led_gpio}'
			if not p.exists():
				with self.sysfs_node(self.gpio_root / 'export') as dst:
					self.sysfs_write(dst, str(self.led_gpio).encode())
			with self.sysfs_node(p / 'direction') as dst: self.sysfs_write(dst, b'low')
			self.led_gpio = self.sysfs_node(p / 'value')
		return self

	def __exit__(self, *err):
		if self.led_sysfs:
			self.led_run.close()
			self.led_time.close()
			self.led_set.close()
			if self.led_exit_wait: self.wait_for_state(0)
			self.sysfs_write(self.led_trigger, b'none')
			self.led_trigger.close()
			if self.led_exit_state: self.set_brightness()
		else:
			self.sysfs_write( self.led_gpio,
				str(int(bool(self.led_exit_state))).encode() )
			self.led_gpio.close()

	def wait_for_state(self, state, checks=24, timeout=1.0):
		for n in range(checks, -1, -1):
			with self.sysfs_node(self.led_sysfs / 'state', 'r') as src:
				if bool(int(self.file_access_wrap(src.read).strip())) == bool(state): break
			if n: time.sleep(timeout / checks)

	def set_brightness(self, value=None):
		if value is None:
			with self.sysfs_node(self.led_sysfs / 'max_brightness', 'r') as src:
				value = int(self.file_access_wrap(src.read).strip())
		with self.sysfs_node(self.led_sysfs / 'brightness') as dst:
			self.sysfs_write(dst, str(value).encode())

	def state(self, on, ms):
		if self.led_sysfs and on:
			self.sysfs_write(self.led_time, str(int(ms * self.led_sysfs_k)).encode())
			self.sysfs_write(self.led_run, b'1')
		if self.led_gpio:
			self.sysfs_write(self.led_gpio, str(int(bool(on))).encode())
		time.sleep(ms / 1000)
		if self.led_gpio and on: self.sysfs_write(self.led_gpio, b'0')


def blink_seq(led, seq, dry_run=False, conf=None, log=None):
	'''Blinks `seq` on `led` (str for sysfs name, int for gpio pin).
		`conf` should be an instance of BlinkConfig.
		`dry_run` runs the `seq` ignoring any looping options just to check for correctness.
		`log` is only used for debugging.
		`seq` step syntax:
			{ '+' | '-' }{ ms:int | s:float 's' } |
				{ 'bit-repr' | 'r'['epeat'] }:{args} | {n[/bits][-dec]} | '[' | ']' | '<'
		`seq` example: +1s r:5 [ -100 +100 ] -1.5s 237 -5s <
	'''
	if not conf: conf = BlinkConfig()
	if isinstance(seq, str): seq = seq.split()

	ledctl_ctx = (LEDControl if not dry_run else LEDControlBase)(led, conf)
	with ledctl_ctx as ledctl:
		init = True

		while True:
			if init:
				bit_1, bit_0, bit_delay = conf.bit_repr
				seq_deque, subseq = deque(seq), list()
				init, repeat = False, 0
			try: step = seq_deque.popleft()
			except IndexError: break

			if isinstance(step, (int, float)) and step <= 0:
				raise BlinkError(seq, step)
			if not step: continue
			step = str(step)
			if log: log.debug(f'Step: {step!r}')
			# if log: log.debug(f'  subseq: {subseq}')

			### Options and meta-steps

			if ':' in step:
				op, args = step.split(':', 1)
				if op == 'bit-repr': bit_1, bit_0, bit_delay = map(int, args.split(','))
				if op in ['r', 'repeat']: repeat = int(args)
				else: raise BlinkError(seq, op, step)
				continue

			if step[0] == '<':
				if not dry_run: init = True
				continue

			if step in '[]':
				if step == '[':
					if subseq: raise BlinkError(seq, step, subseq)
					subseq.append(None)
					continue
				elif not subseq: raise BlinkError(seq, step, subseq)
				else: subseq.append(None)
			elif subseq: subseq.append(step)

			### Actual LED actions

			if step[0] in '+-':
				on, ms = int(step[0] == '+'), step[1:]
				ms = int(ms if not ms.endswith('s') else (float(ms[:-1])*1000))
				if log: log.debug(f'- state: {on} [{ms}ms]')
				if not dry_run: ledctl.state(on, ms)

			elif step[0].isdigit():
				step_v, bits = step.split('/', 1) if '/' in step else (step, None)
				bits, dec = bits.split('-', 1) if bits and '-' in bits else (bits, 0)
				step_v = int(step_v)
				bits = int(bits) if bits else step_v.bit_length()
				if dec: step_v -= int(dec)
				if log: log.debug(f'- value: {step_v} = {step_v:b} [{bits}b]')
				if step_v:
					v = step_v
					for n in range(bits):
						bit, v = v & 1, v >> 1
						ms = bit_1 if bit else bit_0
						if log: log.debug(f'-- {step_v:b}[{n}] = {bit} [{ms}ms]')
						if not dry_run:
							ledctl.state(1, ms)
							time.sleep(bit_delay / 1000)

			### (Limited) repeat for last step or subseq

			if repeat:
				repeat_step = None
				if subseq:
					if subseq[-1] is None:
						repeat_step = subseq[1:-1]
						subseq.clear()
				else: repeat_step = [step]
				if repeat_step:
					if log: log.debug(f'Repeat: {repeat} x {repeat_step}')
					for n in range(repeat-1):
						seq_deque.extendleft(reversed(repeat_step))
					repeat = 0


def main(args=None, conf=None):
	if not conf: conf = BlinkConfig()

	import argparse
	parser = argparse.ArgumentParser(
		description='Blink passed sequence using linux led/gpio sysfs interface.',
		epilog='Simple example: led-blink-seq --debug'
			' -l led0 "+1s r:5 [ -100 +100 ] -1.5s 237 -5s <"')

	parser.add_argument('seq', nargs='+', help='Sequence of actions to run.')

	parser.add_argument('-f', '--fork', action='store_true',
		help='Fork and exit in main pid after start, i.e. daemonize.')
	parser.add_argument('-l', '--led', metavar='name',
		help=f'Name of the sysfs led under {conf.leds_root} to use.')
	parser.add_argument('-c', '--conf', metavar='json-object',
		help='Overrides for configuration keys, as a single json object.'
			' See BlinkConfig class in the script for the list of used keys/values.'
			' Example: {bit_repr: [2000, 200, 1400], interval: [2.0, 10.0]}')
	parser.add_argument('-g', '--gpio-pin', type=int, metavar='n',
		help='GPIO pin number to switch for LED control.')
	parser.add_argument('-e', '--exit-state', type=int, metavar='1/0',
		help='Whether to leave LED enabled or disabled on exit.')

	parser.add_argument('-n', '--dry-run',
		action='store_true', help='Check if supplied sequence has any errors and exit.')
	parser.add_argument('-d', '--debug',
		action='store_true', help='Verbose operation mode.')
	opts = parser.parse_args(sys.argv[1:] if args is None else args)

	if not bool(opts.led) ^ bool(opts.gpio_pin):
		parser.error('One of --led or --gpio-pin must be specified (but not both).')

	logging.basicConfig(level=logging.DEBUG if opts.debug else logging.WARNING)
	log = logging.getLogger('main')

	if opts.conf:
		import json
		for k, v in json.loads(opts.conf).items():
			if k.startswith('_'): continue
			setattr(conf, k, v)

	conf.exit_state = bool(opts.exit_state)
	seq = ' '.join(opts.seq).split()
	led_spec = opts.led or int(opts.gpio_pin)
	blink_seq(led_spec, seq, dry_run=True, conf=conf, log=opts.dry_run and log) # syntax check

	if not opts.dry_run:
		if opts.fork and os.fork(): return
		blink_seq(led_spec, seq, conf=conf, log=log)


if __name__ == '__main__':
	import signal
	for sig in 'int term'.upper().split():
		signal.signal(getattr(signal, f'SIG{sig}'), lambda sig,frm: sys.exit(0))
	sys.exit(main())