<commit_before>"""Contains the Light class."""
import asyncio

from functools import partial

from typing import Set, Dict, List, Tuple, Any

from mpf.core.delays import DelayManager

from mpf.core.platform import LightsPlatform, LightConfig, LightConfigColors

from mpf.core.device_monitor import DeviceMonitor
from mpf.core.machine import MachineController
from mpf.core.rgb_color import RGBColor, ColorException
from mpf.core.system_wide_device import SystemWideDevice
from mpf.devices.device_mixins import DevicePositionMixin

MYPY = False
if MYPY:
    from mpf.platforms.interfaces.light_platform_interface import LightPlatformInterface    # pylint: disable-msg=cyclic-import,unused-import; # noqa


class LightStackEntry:

    """Data class for a light stack entry."""

    __slots__ = ["priority", "start_time", "start_color", "dest_time", "dest_color", "key"]

    # pylint: disable-msg=too-many-arguments
    def __init__(self, priority, key, start_time, start_color, dest_time, dest_color):
        """Initialize light stack entry."""
        self.priority = priority
        self.key = key
        self.start_time = start_time
        self.start_color = start_color
        self.dest_time = dest_time
        self.dest_color = dest_color

    def __gt__(self, other):
        """Compare two stack entries."""
        return self.priority > other.priority or (self.priority == other.priority and self.key > other.key)

    def __repr__(self):
        """Return string representation."""
        return "<LightStackEntry {}: {} ({}) -> {} ({}) Priority: {}>".format(
            self.key, self.start_color, self.start_time, self.dest_color, self.dest_time, self.priority)


@DeviceMonitor(_color="color", _do_not_overwrite_setter=True)
class Light(SystemWideDevice, DevicePositionMixin):

    """A light in a pinball machine."""

    config_section = 'lights'
    collection = 'lights'
    class_label = 'light'

    __slots__ = ["hw_drivers", "platforms", "delay", "default_fade_ms", "_color_correction_profile", "stack",
                 "_off_color", "_drivers_loaded", "_last_fade_target"]

    def __init__(self, machine, name):
        """Initialise light."""
        self.hw_drivers = {}        # type: Dict[str, List[LightPlatformInterface]]
        self.platforms = set()      # type: Set[LightsPlatform]
        super().__init__(machine, name)
        self.machine.light_controller.initialise_light_subsystem()
        self.delay = DelayManager(self.machine)
        self._drivers_loaded = asyncio.Future()

        self.default_fade_ms = None
        self._off_color = RGBColor("off")

        self._color_correction_profile = None
        self._last_fade_target = None

        self.stack = list()     # type: List[LightStackEntry]
        """A list of dicts which represents different commands that have come
        in to set this light to a certain color (and/or fade). Each entry in the
        list contains the following key/value pairs:

        priority:
            The relative priority of this color command. Higher numbers
            take precedent, and the highest priority entry will be the command
            that's currently active. In the event of a tie, whichever entry was
            added last wins (based on 'start_time' below).
        start_time:
            The clock time when this command was added. Primarily used
            to calculate fades, but also used as a tie-breaker for multiple
            entries with the same priority.
        start_color:
            RGBColor() of the color of this light when this command came in.
        dest_time:
            Clock time that represents when a fade (from start_color to
            dest_color) will be done. If this is 0, that means there is no
            fade. When a fade is complete, this value is reset to 0.
        dest_color:
            RGBColor() of the destination this light is fading to. If
            a command comes in with no fade, then this will be the same as the
            'color' below.
        key:
            An arbitrary unique identifier to keep multiple entries in the
            stack separate. If a new color command comes in with a key that
            already exists for an entry in the stack, that entry will be
            replaced by the new entry. The key is also used to remove entries
            from the stack (e.g. when shows or modes end and they want to
            remove their commands from the light).
        """

    @classmethod
    def device_class_init(cls, machine: MachineController):
        """Register handler for duplicate light number checks."""
        machine.events.add_handler("init_phase_4",
                                   cls._check_duplicate_light_numbers,
                                   machine=machine)

    def get_hw_numbers(self):
        """Return a list of all hardware driver numbers."""
        numbers = []
        for _, drivers in sorted(self.hw_drivers.items()):
            for driver in sorted(drivers, key=lambda x: x.number):
                numbers.append(driver.number)

        return numbers

    @staticmethod
    def _check_duplicate_light_numbers(machine: MachineController, **kwargs):
        del kwargs
        check_set = set()
        for light in machine.lights.values():
            for drivers in light.hw_drivers.values():
                for driver in drivers:
                    key = (light.config['platform'], driver.number, type(driver))
                    if key in check_set:
                        raise AssertionError(
                            "Duplicate light number {} {} for light {}".format(
                                type(driver), driver.number, light))

                    check_set.add(key)

    def _map_channels_to_colors(self, channel_list) -> dict:<commit_msg>Change get_hw_numbers sort to use value and not key<commit_after>"""Contains the Light class."""
import asyncio

from functools import partial

from typing import Set, Dict, List, Tuple, Any

from mpf.core.delays import DelayManager

from mpf.core.platform import LightsPlatform, LightConfig, LightConfigColors

from mpf.core.device_monitor import DeviceMonitor
from mpf.core.machine import MachineController
from mpf.core.rgb_color import RGBColor, ColorException
from mpf.core.system_wide_device import SystemWideDevice
from mpf.devices.device_mixins import DevicePositionMixin

MYPY = False
if MYPY:
    from mpf.platforms.interfaces.light_platform_interface import LightPlatformInterface    # pylint: disable-msg=cyclic-import,unused-import; # noqa


class LightStackEntry:

    """Data class for a light stack entry."""

    __slots__ = ["priority", "start_time", "start_color", "dest_time", "dest_color", "key"]

    # pylint: disable-msg=too-many-arguments
    def __init__(self, priority, key, start_time, start_color, dest_time, dest_color):
        """Initialize light stack entry."""
        self.priority = priority
        self.key = key
        self.start_time = start_time
        self.start_color = start_color
        self.dest_time = dest_time
        self.dest_color = dest_color

    def __gt__(self, other):
        """Compare two stack entries."""
        return self.priority > other.priority or (self.priority == other.priority and self.key > other.key)

    def __repr__(self):
        """Return string representation."""
        return "<LightStackEntry {}: {} ({}) -> {} ({}) Priority: {}>".format(
            self.key, self.start_color, self.start_time, self.dest_color, self.dest_time, self.priority)


@DeviceMonitor(_color="color", _do_not_overwrite_setter=True)
class Light(SystemWideDevice, DevicePositionMixin):

    """A light in a pinball machine."""

    config_section = 'lights'
    collection = 'lights'
    class_label = 'light'

    __slots__ = ["hw_drivers", "platforms", "delay", "default_fade_ms", "_color_correction_profile", "stack",
                 "_off_color", "_drivers_loaded", "_last_fade_target"]

    def __init__(self, machine, name):
        """Initialise light."""
        self.hw_drivers = {}        # type: Dict[str, List[LightPlatformInterface]]
        self.platforms = set()      # type: Set[LightsPlatform]
        super().__init__(machine, name)
        self.machine.light_controller.initialise_light_subsystem()
        self.delay = DelayManager(self.machine)
        self._drivers_loaded = asyncio.Future()

        self.default_fade_ms = None
        self._off_color = RGBColor("off")

        self._color_correction_profile = None
        self._last_fade_target = None

        self.stack = list()     # type: List[LightStackEntry]
        """A list of dicts which represents different commands that have come
        in to set this light to a certain color (and/or fade). Each entry in the
        list contains the following key/value pairs:

        priority:
            The relative priority of this color command. Higher numbers
            take precedent, and the highest priority entry will be the command
            that's currently active. In the event of a tie, whichever entry was
            added last wins (based on 'start_time' below).
        start_time:
            The clock time when this command was added. Primarily used
            to calculate fades, but also used as a tie-breaker for multiple
            entries with the same priority.
        start_color:
            RGBColor() of the color of this light when this command came in.
        dest_time:
            Clock time that represents when a fade (from start_color to
            dest_color) will be done. If this is 0, that means there is no
            fade. When a fade is complete, this value is reset to 0.
        dest_color:
            RGBColor() of the destination this light is fading to. If
            a command comes in with no fade, then this will be the same as the
            'color' below.
        key:
            An arbitrary unique identifier to keep multiple entries in the
            stack separate. If a new color command comes in with a key that
            already exists for an entry in the stack, that entry will be
            replaced by the new entry. The key is also used to remove entries
            from the stack (e.g. when shows or modes end and they want to
            remove their commands from the light).
        """

    @classmethod
    def device_class_init(cls, machine: MachineController):
        """Register handler for duplicate light number checks."""
        machine.events.add_handler("init_phase_4",
                                   cls._check_duplicate_light_numbers,
                                   machine=machine)

    def get_hw_numbers(self):
        """Return a list of all hardware driver numbers."""
        numbers = []
        for _, drivers in sorted(self.hw_drivers.items(), key=lambda x: x[1]):
            for driver in sorted(drivers, key=lambda x: x.number):
                numbers.append(driver.number)

        return numbers

    @staticmethod
    def _check_duplicate_light_numbers(machine: MachineController, **kwargs):
        del kwargs
        check_set = set()
        for light in machine.lights.values():
            for drivers in light.hw_drivers.values():
                for driver in drivers:
                    key = (light.config['platform'], driver.number, type(driver))
                    if key in check_set:
                        raise AssertionError(
                            "Duplicate light number {} {} for light {}".format(
                                type(driver), driver.number, light))

                    check_set.add(key)

    def _map_channels_to_colors(self, channel_list) -> dict:
{'fraction_non_alphanumeric': 0.08306188925081433, 'fraction_numerical': 0.0006000342876735813, 'mean_word_length': 2.4785330948121644, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1417248', 'n_tokens_mistral': 3115, 'n_tokens_neox': 3002, 'n_words': 1054}