<commit_before>"""
RPIO extends RPi.GPIO with interrupt handling. Importing this also sets
the default mode to GPIO.BCM (the same numbering system the kernel uses,
as opposed to the pin ids (GPIO.BOARD)).

You can use RPIO the same way as RPi.GPIO (eg. RPIO.setmode(...),
RPIO.input(...)), as well as access the new interrupt handling methods.
The following example shows how to react on events on 3 pins by using
interrupts, each with different edge detections:

    import logging
    log_format = '%(levelname)s | %(asctime)-15s | %(message)s'
    logging.basicConfig(format=log_format, level=logging.DEBUG)
    import RPIO

    def do_something(gpio_id, value):
        logging.info("New value for GPIO %s: %s" % (gpio_id, value))

    RPIO.add_interrupt_callback(17, do_something, edge='rising')
    RPIO.add_interrupt_callback(18, do_something, edge='falling')
    RPIO.add_interrupt_callback(19, do_something, edge='both')
    RPIO.wait_for_interrupts()

If you want to receive a callback inside a Thread (which won't block anything
else on the system), set `threaded_callback` to True when adding an interrupt-
callback. Here is an example:

    RPIO.add_interrupt_callback(17, do_something, edge='rising',
            threaded_callback=True)

Make sure to double-check the value returned from the interrupt, since it
is not necessarily corresponding to the edge (eg. 0 may come in as value,
even if edge="rising").

To remove all callbacks from a certain gpio pin, use
`RPIO.del_interrupt_callback(gpio_id)`. To stop the `wait_for_interrupts()`
loop you can call `RPIO.stop_waiting_for_interrupts()`.

Author: <NAME> <<EMAIL>>
License: MIT
URL: https://github.com/metachris/raspberrypi-utils
"""
import select
import os.path

from logging import info, warn, error
from threading import Thread
from functools import partial

from GPIO import *
from GPIO import cleanup as _cleanup_orig

VERSION = "0.1.8"

# BCM numbering mode by default
setmode(BCM)

# Interrupt callback maps
_map_fileno_to_file = {}
_map_fileno_to_gpioid = {}
_map_gpioid_to_fileno = {}
_map_gpioid_to_callbacks = {}

# Keep track of created kernel interfaces for later cleanup
_gpio_kernel_interfaces_created = []

# Whether to continue the epoll loop or quit at next chance. You
# can manually set this to False to stop `wait_for_interrupts()`.
_is_waiting_for_interrupts = False

# Internals
_SYS_GPIO_ROOT = '/sys/class/gpio/'
_epoll = select.epoll()


def _threaded_callback(callback, *args):
    """ Internal wrapper to start a callback in threaded mode """
    Thread(target=callback, args=args).start()


def add_interrupt_callback(gpio_id, callback, edge='both',
        threaded_callback=False):
    """
    Add a callback to be executed when the value on 'gpio_id' changes to the
    edge specified via the 'edge' parameter (default='both').

    If threaded_callback is True, the callback will be started inside a Thread.
    """
    info("Adding callback for GPIO %s" % gpio_id)
    if not edge in ["falling", "rising", "both", "none"]:
        raise AttributeError("'%s' is not a valid edge.")

    # Prepare the callback (wrap in Thread if needed)
    cb = callback if not threaded_callback else \
            partial(_threaded_callback, callback)

    # Check if /sys/class/gpio/gpioN interface exists; else create it
    path_gpio = "%sgpio%s/" % (_SYS_GPIO_ROOT, gpio_id)
    if not os.path.exists(path_gpio):
        with open(_SYS_GPIO_ROOT + "export", "w") as f:
            f.write("%s" % gpio_id)
        _gpio_kernel_interfaces_created.append(gpio_id)
        info("- kernel interface created for GPIO %s" % gpio_id)

    # If initial callback for this GPIO then set everything up. Else make sure
    # the edge detection is the same and add this to the callback list.
    if gpio_id in _map_gpioid_to_callbacks:
        with open(path_gpio + "edge", "r") as f:
            e = f.read().strip()
            if e != edge:
                raise AttributeError(("Cannot add callback for gpio %s:"
                        " edge detection '%s' not compatible with existing"
                        " edge detection '%s'.") % (gpio_id, edge, e))

        # Check whether edge is the same, else throw Exception
        info("- kernel interface already configured for GPIO %s" % gpio_id)
        _map_gpioid_to_callbacks[gpio_id].append(cb)

    else:
        # Configure gpio as input
        with open(path_gpio + "direction", "w") as f:
            f.write("in")

        # Configure gpio edge detection
        with open(path_gpio + "edge", "w") as f:
            f.write(edge)

        info("- kernel interface configured for GPIO %s" % gpio_id)

        # Open the gpio value stream and read the initial value
        f = open(path_gpio + "value", 'r')
        val_initial = f.read().strip()
        info("- inital gpio value: %s" % val_initial)
        f.seek(0)

        # Add callback info to the mapping dictionaries
        _map_fileno_to_file[f.fileno()] = f
        _map_fileno_to_gpioid[f.fileno()] = gpio_id
        _map_gpioid_to_fileno[gpio_id] = f.fileno()
        _map_gpioid_to_callbacks[gpio_id] = [cb]

        # Add to epoll
        _epoll.register(f.fileno(), select.EPOLLPRI | select.EPOLLERR)


def del_interrupt_callback(gpio_id):
    """ Delete all interrupt callbacks from a certain gpio """
    fileno = _map_gpioid_to_fileno[gpio_id]<commit_msg>RPIO: debug log messages instead of info<commit_after>"""
RPIO extends RPi.GPIO with interrupt handling. Importing this also sets
the default mode to GPIO.BCM (the same numbering system the kernel uses,
as opposed to the pin ids (GPIO.BOARD)).

You can use RPIO the same way as RPi.GPIO (eg. RPIO.setmode(...),
RPIO.input(...)), as well as access the new interrupt handling methods.
The following example shows how to react on events on 3 pins by using
interrupts, each with different edge detections:

    import logging
    log_format = '%(levelname)s | %(asctime)-15s | %(message)s'
    logging.basicConfig(format=log_format, level=logging.DEBUG)
    import RPIO

    def do_something(gpio_id, value):
        logging.info("New value for GPIO %s: %s" % (gpio_id, value))

    RPIO.add_interrupt_callback(17, do_something, edge='rising')
    RPIO.add_interrupt_callback(18, do_something, edge='falling')
    RPIO.add_interrupt_callback(19, do_something, edge='both')
    RPIO.wait_for_interrupts()

If you want to receive a callback inside a Thread (which won't block anything
else on the system), set `threaded_callback` to True when adding an interrupt-
callback. Here is an example:

    RPIO.add_interrupt_callback(17, do_something, edge='rising',
            threaded_callback=True)

Make sure to double-check the value returned from the interrupt, since it
is not necessarily corresponding to the edge (eg. 0 may come in as value,
even if edge="rising").

To remove all callbacks from a certain gpio pin, use
`RPIO.del_interrupt_callback(gpio_id)`. To stop the `wait_for_interrupts()`
loop you can call `RPIO.stop_waiting_for_interrupts()`.

Author: <NAME> <<EMAIL>>
License: MIT
URL: https://github.com/metachris/raspberrypi-utils
"""
import select
import os.path

from logging import debug, info, warn, error
from threading import Thread
from functools import partial

from GPIO import *
from GPIO import cleanup as _cleanup_orig

VERSION = "0.1.8"

# BCM numbering mode by default
setmode(BCM)

# Interrupt callback maps
_map_fileno_to_file = {}
_map_fileno_to_gpioid = {}
_map_gpioid_to_fileno = {}
_map_gpioid_to_callbacks = {}

# Keep track of created kernel interfaces for later cleanup
_gpio_kernel_interfaces_created = []

# Whether to continue the epoll loop or quit at next chance. You
# can manually set this to False to stop `wait_for_interrupts()`.
_is_waiting_for_interrupts = False

# Internals
_SYS_GPIO_ROOT = '/sys/class/gpio/'
_epoll = select.epoll()


def _threaded_callback(callback, *args):
    """ Internal wrapper to start a callback in threaded mode """
    Thread(target=callback, args=args).start()


def add_interrupt_callback(gpio_id, callback, edge='both',
        threaded_callback=False):
    """
    Add a callback to be executed when the value on 'gpio_id' changes to the
    edge specified via the 'edge' parameter (default='both').

    If threaded_callback is True, the callback will be started inside a Thread.
    """
    debug("Adding callback for GPIO %s" % gpio_id)
    if not edge in ["falling", "rising", "both", "none"]:
        raise AttributeError("'%s' is not a valid edge.")

    # Prepare the callback (wrap in Thread if needed)
    cb = callback if not threaded_callback else \
            partial(_threaded_callback, callback)

    # Check if /sys/class/gpio/gpioN interface exists; else create it
    path_gpio = "%sgpio%s/" % (_SYS_GPIO_ROOT, gpio_id)
    if not os.path.exists(path_gpio):
        with open(_SYS_GPIO_ROOT + "export", "w") as f:
            f.write("%s" % gpio_id)
        _gpio_kernel_interfaces_created.append(gpio_id)
        debug("- kernel interface created for GPIO %s" % gpio_id)

    # If initial callback for this GPIO then set everything up. Else make sure
    # the edge detection is the same and add this to the callback list.
    if gpio_id in _map_gpioid_to_callbacks:
        with open(path_gpio + "edge", "r") as f:
            e = f.read().strip()
            if e != edge:
                raise AttributeError(("Cannot add callback for gpio %s:"
                        " edge detection '%s' not compatible with existing"
                        " edge detection '%s'.") % (gpio_id, edge, e))

        # Check whether edge is the same, else throw Exception
        debug("- kernel interface already configured for GPIO %s" % gpio_id)
        _map_gpioid_to_callbacks[gpio_id].append(cb)

    else:
        # Configure gpio as input
        with open(path_gpio + "direction", "w") as f:
            f.write("in")

        # Configure gpio edge detection
        with open(path_gpio + "edge", "w") as f:
            f.write(edge)

        debug("- kernel interface configured for GPIO %s" % gpio_id)

        # Open the gpio value stream and read the initial value
        f = open(path_gpio + "value", 'r')
        val_initial = f.read().strip()
        debug("- inital gpio value: %s" % val_initial)
        f.seek(0)

        # Add callback info to the mapping dictionaries
        _map_fileno_to_file[f.fileno()] = f
        _map_fileno_to_gpioid[f.fileno()] = gpio_id
        _map_gpioid_to_fileno[gpio_id] = f.fileno()
        _map_gpioid_to_callbacks[gpio_id] = [cb]

        # Add to epoll
        _epoll.register(f.fileno(), select.EPOLLPRI | select.EPOLLERR)


def del_interrupt_callback(gpio_id):
    """ Delete all interrupt callbacks from a certain gpio """
    fileno = _map_gpioid_to_fileno[gpio_id]
{'fraction_non_alphanumeric': 0.09665151655690567, 'fraction_numerical': 0.0029681847695019014, 'mean_word_length': 3.354604200323102, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5150026', 'n_tokens_mistral': 3342, 'n_tokens_neox': 3267, 'n_words': 1128}