<commit_before># Copyright 2017 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""LED driver for the VoiceHat."""

import itertools
import threading
import time
import RPi.GPIO as GPIO


class LED:
    """Starts a background thread to show patterns with the LED.

    Simple usage:
        my_led = LED(channel = 25)
        my_led.start()
        my_led.set_state(LED.BEACON)
        my_led.stop()
    """

    OFF = 0
    ON = 1
    BLINK = 2
    BLINK_3 = 3
    BEACON = 4
    BEACON_DARK = 5
    DECAY = 6
    PULSE_SLOW = 7
    PULSE_QUICK = 8

    def __init__(self, channel):
        self.animator = threading.Thread(target=self._animate, daemon=True)
        self.channel = channel
        self.iterator = None
        self.running = False
        self.state = None
        self.sleep = 0
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(channel, GPIO.OUT)
        self.pwm = GPIO.PWM(channel, 100)
        self.lock = threading.Lock()

    def __del__(self):
        self.stop()
        GPIO.cleanup(self.channel)

    def start(self):
        """Start the LED driver."""
        with self.lock:  # pylint: disable=E1129
            if not self.running:
                self.running = True
                self.pwm.start(0)  # off by default
                self.animator.start()

    def stop(self):
        """Stop the LED driver and sets the LED to off."""
        with self.lock:  # pylint: disable=E1129
            if self.running:
                self.running = False
                self.animator.join()
                self.pwm.stop()

    def set_state(self, state):
        """Set the LED driver's new state.

        Note the LED driver must be started for this to have any effect.
        """
        with self.lock:  # pylint: disable=E1129
            self.state = state

    def _animate(self):
        while True:
            state = None
            running = False
            with self.lock:  # pylint: disable=E1129
                state = self.state
                self.state = None
                running = self.running
            if not running:
                return
            if state:
                if not self._parse_state(state):
                    raise ValueError('unsupported state: %d' % state)
            if self.iterator:
                self.pwm.ChangeDutyCycle(next(self.iterator))
                time.sleep(self.sleep)
            else:
                # We can also wait for a state change here with a Condition.
                time.sleep(1)

    def _parse_state(self, state):
        self.iterator = None
        self.sleep = 0.0
        handled = False

        if state == self.OFF:
            self.pwm.ChangeDutyCycle(0)
            handled = True<commit_msg>Narrow check for state change in led driver<commit_after># Copyright 2017 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""LED driver for the VoiceHat."""

import itertools
import threading
import time
import RPi.GPIO as GPIO


class LED:
    """Starts a background thread to show patterns with the LED.

    Simple usage:
        my_led = LED(channel = 25)
        my_led.start()
        my_led.set_state(LED.BEACON)
        my_led.stop()
    """

    OFF = 0
    ON = 1
    BLINK = 2
    BLINK_3 = 3
    BEACON = 4
    BEACON_DARK = 5
    DECAY = 6
    PULSE_SLOW = 7
    PULSE_QUICK = 8

    def __init__(self, channel):
        self.animator = threading.Thread(target=self._animate, daemon=True)
        self.channel = channel
        self.iterator = None
        self.running = False
        self.state = None
        self.sleep = 0
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(channel, GPIO.OUT)
        self.pwm = GPIO.PWM(channel, 100)
        self.lock = threading.Lock()

    def __del__(self):
        self.stop()
        GPIO.cleanup(self.channel)

    def start(self):
        """Start the LED driver."""
        with self.lock:  # pylint: disable=E1129
            if not self.running:
                self.running = True
                self.pwm.start(0)  # off by default
                self.animator.start()

    def stop(self):
        """Stop the LED driver and sets the LED to off."""
        with self.lock:  # pylint: disable=E1129
            if self.running:
                self.running = False
                self.animator.join()
                self.pwm.stop()

    def set_state(self, state):
        """Set the LED driver's new state.

        Note the LED driver must be started for this to have any effect.
        """
        with self.lock:  # pylint: disable=E1129
            self.state = state

    def _animate(self):
        while True:
            state = None
            running = False
            with self.lock:  # pylint: disable=E1129
                state = self.state
                self.state = None
                running = self.running
            if not running:
                return
            if state is not None:
                if not self._parse_state(state):
                    raise ValueError('unsupported state: %d' % state)
            if self.iterator:
                self.pwm.ChangeDutyCycle(next(self.iterator))
                time.sleep(self.sleep)
            else:
                # We can also wait for a state change here with a Condition.
                time.sleep(1)

    def _parse_state(self, state):
        self.iterator = None
        self.sleep = 0.0
        handled = False

        if state == self.OFF:
            self.pwm.ChangeDutyCycle(0)
            handled = True
{'fraction_non_alphanumeric': 0.08499539736115373, 'fraction_numerical': 0.013807916538815588, 'mean_word_length': 1.7847073900042716, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5737177', 'n_tokens_mistral': 1928, 'n_tokens_neox': 1807, 'n_words': 607}