Input axis stripchart for python-gobject and Gtk3

axischart.py

#!/usr/bin/env python
# Input axis stripchart for (python-gobject & Gtk3) or (PyGtk, evolving)
# CC0 1.0 2016-01-26 Andrew Chadwick, all rights waived
# https://creativecommons.org/publicdomain/zero/1.0/

import sys
import os
from getopt import getopt
from collections import deque


APP_NAME = "Axischart"
NUM_SAMPLES = 500
HMARGIN = 20
VMARGIN = 10
CHARTHEIGHT = 40
USE_PYGTK = os.environ.get("USE_PYGTK", "") not in ["0", ""]


import logging
logger = logging.getLogger(APP_NAME)


if USE_PYGTK:
    import gtk as Gtk
    from gtk import gdk as Gdk
else:
    import gi
    gi.require_version('Gtk', '3.0')
    from gi.repository import Gtk
    from gi.repository import Gdk
import cairo


class Tester (object):

    def __init__(self):
        super(Tester, self).__init__()
        w = Gtk.Window()
        w.set_title(APP_NAME)
        w.connect("destroy", self._destroy_cb)
        d = Gtk.DrawingArea()
        if USE_PYGTK:
            last_axis = Gdk.AXIS_LAST
        else:
            last_axis = Gdk.AxisUse.LAST
        num_axes = int(last_axis)
        d.set_size_request(NUM_SAMPLES + 2*HMARGIN,
                           VMARGIN + ((CHARTHEIGHT+VMARGIN) * num_axes))
        if USE_PYGTK:
            d.connect("expose-event", self._expose_cb)
        else:
            d.connect("draw", self._draw_cb)
        self._axis_ranges = {}  # {axis_num: [min_seen, max_seen]}
        self._axis_values = {}  # {axis_num: deque([val0, ..., valN])}

        e = Gtk.EventBox()
        e.connect("button-press-event", self._event_cb)
        e.connect("motion-notify-event", self._event_cb)
        e.connect("proximity-in-event", self._event_cb)
        e.connect("proximity-out-event", self._event_cb)

        e.add(d)
        w.add(e)

        if USE_PYGTK:
            maskmod = Gdk
        else:
            maskmod = Gdk.EventMask
        mask = ( maskmod.POINTER_MOTION_MASK
            | maskmod.PROXIMITY_OUT_MASK
            | maskmod.PROXIMITY_IN_MASK
            | maskmod.BUTTON_PRESS_MASK
            | maskmod.BUTTON_RELEASE_MASK
            | maskmod.BUTTON1_MOTION_MASK
            | maskmod.ENTER_NOTIFY_MASK 
        )
        e.add_events(mask)
        w.add_events(mask)
        d.add_events(mask)

        self.win = w
        self._devices = set()

    def run(self):
        self.win.show_all()
        Gtk.main()

    def _destroy_cb(self, win):
        Gtk.main_quit()

    def _expose_cb(self, da, event):
        win = da.get_window()
        cr = win.cairo_create()
        self._draw_cb(da, cr)

    def _draw_cb(self, da, cr):
        cr.set_source_rgb(1, 1, 1)
        cr.paint()
        axis_ids = set(self._axis_ranges.keys())
        axis_ids.intersection_update(set(self._axis_values.keys()))
        for i in sorted(axis_ids):
            cr.set_line_width(1)
            cr.set_line_cap(cairo.LINE_CAP_SQUARE)
            y0 = (i*(VMARGIN+CHARTHEIGHT)) + CHARTHEIGHT/2.0
            vmin, vmax = self._axis_ranges[i]
            assert vmax-vmin > 0
            values = self._axis_values[i]
            for x, v in enumerate(values):
                x += HMARGIN
                if v is None:
                    cr.set_source_rgb(1,0,0)
                    cr.rectangle(x, y0, 1, 1)
                    cr.fill()
                else:
                    cr.set_source_rgb(0,0,0)
                    y = y0 + (CHARTHEIGHT*v) / (-2.0*(vmax-vmin))
                    cr.move_to(x+0.5, y0+0.5)
                    cr.line_to(x+0.5, y+0.5)
                    cr.stroke()
            cr.set_source_rgb(1,0,0)
            cr.move_to(HMARGIN, y0)
            use = Gdk.AxisUse(i)
            cr.show_text(use.value_nick)

    def _record_axis_value(self, i, value):
        default_range = (0.0, 1.0)
        if value is not None:
            vmin, vmax = self._axis_ranges.get(i, default_range)
            self._axis_ranges[i] = (min(vmin, value), max(vmax, value))
        elif i not in self._axis_ranges:
            self._axis_ranges[i] = default_range
        values = self._axis_values.get(i, None)
        if values is None:
            values = deque(maxlen=NUM_SAMPLES)
            self._axis_values[i] = values
        values.append(value)

    def _event_cb(self, widget, event):
        if USE_PYGTK:
            dev = event.device
        else:
            dev = event.get_source_device()
        #print dir(dev)
        #print dir(event)
        if dev not in self._devices:
            src = dev.get_source()
            logger.info(
                "New device class=%r name=%r (src=%r)",
                dev.__class__.__name__,
                dev.get_name(),
                src.value_nick,
            )
            self._devices.add(dev)
        if USE_PYGTK:
            last_axis = Gdk.AXIS_LAST
            axis_use_ignore = Gdk.AXIS_IGNORE
        else:
            last_axis = Gdk.AxisUse.LAST
            axis_use_ignore = Gdk.AxisUse.IGNORE
        for i in range(int(last_axis)):
            use = Gdk.AxisUse(i)
            if use == axis_use_ignore:
                continue
            value = event.get_axis(use)
            if str(value) == 'nan':
                value = None
            self._record_axis_value(i, value)
            #if use == Gdk.AxisUse.PRESSURE and value is not None:
            #    logger.info("Pressure: %0.3f", value)
        widget.queue_draw()

if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    if USE_PYGTK:
        logger.info("Using PyGTK and GTK2")
    else:
        logger.info("Using PyGI and GTK3")
    tester = Tester()
    tester.run()