<commit_before>"""A generic system11 driver overlay.

This is based on the Snux platform to generically support all kinds of System11 platforms.
"""
from typing import Any, Optional, Set, Tuple, Dict

from mpf.core.machine import MachineController
from mpf.core.platform import DriverPlatform, DriverConfig, SwitchSettings, DriverSettings, RepulseSettings, \
    SwitchPlatform, SwitchConfig

from mpf.platforms.interfaces.driver_platform_interface import DriverPlatformInterface, PulseSettings, HoldSettings

from mpf.core.delays import DelayManager

MYPY = False
if MYPY:   # pragma: no cover
    class SwitchDriverPlatform(DriverPlatform, SwitchPlatform):     # noqa
        pass


# pylint: disable-msg=too-many-instance-attributes
class System11OverlayPlatform(DriverPlatform, SwitchPlatform):

    """Overlay platform to drive system11 machines using a WPC controller."""

    __slots__ = ["delay", "platform", "system11_config", "a_side_queue", "c_side_queue",
                 "a_side_done_time", "c_side_done_time", "drivers_holding_a_side", "drivers_holding_c_side",
                 "a_side_enabled", "c_side_enabled", "ac_relay_in_transition", "prefer_a_side", "drivers"]

    def __init__(self, machine: MachineController) -> None:
        """Initialise the board."""
        super().__init__(machine)

        self.delay = DelayManager(machine)

        self.platform = None            # type: Optional[SwitchDriverPlatform]

        self.system11_config = None     # type: Any

        self.a_side_queue = \
            set()   # type: Set[Tuple[DriverPlatformInterface, Optional[PulseSettings], Optional[HoldSettings]]]
        self.c_side_queue = \
            set()   # type: Set[Tuple[DriverPlatformInterface, Optional[PulseSettings], Optional[HoldSettings]]]

        self.a_side_done_time = 0
        self.c_side_done_time = 0
        self.drivers_holding_a_side = set()     # type: Set[DriverPlatformInterface]
        self.drivers_holding_c_side = set()     # type: Set[DriverPlatformInterface]
        self.a_side_enabled = True
        self.c_side_enabled = False
        self.drivers = {}               # type: Dict[str, DriverPlatformInterface]

        self.ac_relay_in_transition = False
        # Specify whether the AC relay should favour the A or C side when at rest.
        # Typically during a game the 'C' side should be preferred, since that is
        # normally where the flashers are which need a quick response without having to wait on the relay.
        # At game over though, it should prefer the 'A' side so that the relay isn't permanently energised.
        self.prefer_a_side = True

    def stop(self):
        """Stop the overlay. Nothing to do here because stop is also called on parent platform."""

    @property
    def a_side_busy(self):
        """Return if A side cannot be switches off right away."""
        return self.drivers_holding_a_side or self.a_side_done_time > self.machine.clock.get_time() or self.a_side_queue

    @property
    def c_side_active(self):
        """Return if C side cannot be switches off right away."""
        return self.drivers_holding_c_side or self.c_side_done_time > self.machine.clock.get_time()

    @property
    def c_side_busy(self):
        """Return if C side cannot be switches off right away."""
        return self.drivers_holding_c_side or self.c_side_done_time > self.machine.clock.get_time() or self.c_side_queue

    @property
    def a_side_active(self):
        """Return if A side cannot be switches off right away."""
        return self.drivers_holding_a_side or self.a_side_done_time > self.machine.clock.get_time()

    def _null_log_handler(self, *args, **kwargs):
        pass

    async def initialize(self):
        """Automatically called by the Platform class after all the core modules are loaded."""
        # load coil platform
        self.platform = self.machine.get_platform_sections(
            "platform", getattr(self.machine.config.get('system11', {}), 'platform', None))

        # we have to wait for coils to be initialized
        self.machine.events.add_handler("init_phase_1", self._initialize)

    def _initialize(self, **kwargs):
        del kwargs
        self._validate_config()

        self.configure_logging('Platform.System11', self.system11_config['console_log'],
                               self.system11_config['file_log'])

        self.log.debug("Configuring A/C Select Relay for driver %s",
                       self.system11_config['ac_relay_driver'].name)

        self.system11_config['ac_relay_driver'].get_and_verify_hold_power(1.0)

        self.log.debug("Configuring A/C Select Relay transition delay for "
                       "%sms", self.system11_config['ac_relay_delay_ms'])

        self.machine.events.add_handler(self.system11_config['prefer_a_side_event'], self._prefer_a_side)
        self.log.info("Configuring System11 driver to prefer A side on event %s",
                      self.system11_config['prefer_a_side_event'])

        self.machine.events.add_handler(self.system11_config['prefer_c_side_event'], self._prefer_c_side)
        self.log.info("Configuring System11 driver to prefer C side on event %s",
                      self.system11_config['prefer_c_side_event'])

    def _prefer_a_side(self, **kwargs):
        del kwargs
        self.prefer_a_side = True
        self._enable_a_side()

    def _prefer_c_side(self, **kwargs):
        del kwargs
        self.prefer_a_side = False
        self._enable_c_side()

    def _validate_config(self):
        self.system11_config = self.machine.config_validator.validate_config(
            'system11', self.machine.config.get('system11', {}))

    def tick(self):
        """System11 main loop.

        Called based on the timer_tick event.
        """
        if self.prefer_a_side:
            if self.a_side_queue:
                self._service_a_side()
            elif self.c_side_queue:
                self._service_c_side()
            elif self.c_side_enabled and not self.c_side_active:
                self._enable_a_side()
        else:
            if self.c_side_queue:
                self._service_c_side()
            elif self.a_side_queue:
                self._service_a_side()
            elif self.a_side_enabled and not self.a_side_active:
                self._enable_c_side()

    def configure_switch(self, number: str, config: SwitchConfig, platform_config: dict):
        """Configure switch on system11 overlay."""
        return self.platform.configure_switch(number, config, platform_config)

    async def get_hw_switch_states(self):
        """Get initial hardware state."""
        return await self.platform.get_hw_switch_states()

    def configure_driver(self, config: DriverConfig, number: str, platform_settings: dict):
        """Configure a driver on the system11 overlay.

        Args:
        ----
            config: Driver config dict
            number: Number of the driver.
            platform_settings: Platform specific config.
        """
        assert self.platform is not None
        orig_number = number

        if number and (number.lower().endswith('a') or number.lower().endswith('c')):
<commit_msg>validate switches for system11 overlays<commit_after>"""A generic system11 driver overlay.

This is based on the Snux platform to generically support all kinds of System11 platforms.
"""
from typing import Any, Optional, Set, Tuple, Dict

from mpf.core.machine import MachineController
from mpf.core.platform import DriverPlatform, DriverConfig, SwitchSettings, DriverSettings, RepulseSettings, \
    SwitchPlatform, SwitchConfig

from mpf.platforms.interfaces.driver_platform_interface import DriverPlatformInterface, PulseSettings, HoldSettings

from mpf.core.delays import DelayManager

MYPY = False
if MYPY:   # pragma: no cover
    class SwitchDriverPlatform(DriverPlatform, SwitchPlatform):     # noqa
        pass


# pylint: disable-msg=too-many-instance-attributes
class System11OverlayPlatform(DriverPlatform, SwitchPlatform):

    """Overlay platform to drive system11 machines using a WPC controller."""

    __slots__ = ["delay", "platform", "system11_config", "a_side_queue", "c_side_queue",
                 "a_side_done_time", "c_side_done_time", "drivers_holding_a_side", "drivers_holding_c_side",
                 "a_side_enabled", "c_side_enabled", "ac_relay_in_transition", "prefer_a_side", "drivers"]

    def __init__(self, machine: MachineController) -> None:
        """Initialise the board."""
        super().__init__(machine)

        self.delay = DelayManager(machine)

        self.platform = None            # type: Optional[SwitchDriverPlatform]

        self.system11_config = None     # type: Any

        self.a_side_queue = \
            set()   # type: Set[Tuple[DriverPlatformInterface, Optional[PulseSettings], Optional[HoldSettings]]]
        self.c_side_queue = \
            set()   # type: Set[Tuple[DriverPlatformInterface, Optional[PulseSettings], Optional[HoldSettings]]]

        self.a_side_done_time = 0
        self.c_side_done_time = 0
        self.drivers_holding_a_side = set()     # type: Set[DriverPlatformInterface]
        self.drivers_holding_c_side = set()     # type: Set[DriverPlatformInterface]
        self.a_side_enabled = True
        self.c_side_enabled = False
        self.drivers = {}               # type: Dict[str, DriverPlatformInterface]

        self.ac_relay_in_transition = False
        # Specify whether the AC relay should favour the A or C side when at rest.
        # Typically during a game the 'C' side should be preferred, since that is
        # normally where the flashers are which need a quick response without having to wait on the relay.
        # At game over though, it should prefer the 'A' side so that the relay isn't permanently energised.
        self.prefer_a_side = True

    def stop(self):
        """Stop the overlay. Nothing to do here because stop is also called on parent platform."""

    @property
    def a_side_busy(self):
        """Return if A side cannot be switches off right away."""
        return self.drivers_holding_a_side or self.a_side_done_time > self.machine.clock.get_time() or self.a_side_queue

    @property
    def c_side_active(self):
        """Return if C side cannot be switches off right away."""
        return self.drivers_holding_c_side or self.c_side_done_time > self.machine.clock.get_time()

    @property
    def c_side_busy(self):
        """Return if C side cannot be switches off right away."""
        return self.drivers_holding_c_side or self.c_side_done_time > self.machine.clock.get_time() or self.c_side_queue

    @property
    def a_side_active(self):
        """Return if A side cannot be switches off right away."""
        return self.drivers_holding_a_side or self.a_side_done_time > self.machine.clock.get_time()

    def _null_log_handler(self, *args, **kwargs):
        pass

    async def initialize(self):
        """Automatically called by the Platform class after all the core modules are loaded."""
        # load coil platform
        self.platform = self.machine.get_platform_sections(
            "platform", getattr(self.machine.config.get('system11', {}), 'platform', None))

        # we have to wait for coils to be initialized
        self.machine.events.add_handler("init_phase_1", self._initialize)

    def _initialize(self, **kwargs):
        del kwargs
        self._validate_config()

        self.configure_logging('Platform.System11', self.system11_config['console_log'],
                               self.system11_config['file_log'])

        self.log.debug("Configuring A/C Select Relay for driver %s",
                       self.system11_config['ac_relay_driver'].name)

        self.system11_config['ac_relay_driver'].get_and_verify_hold_power(1.0)

        self.log.debug("Configuring A/C Select Relay transition delay for "
                       "%sms", self.system11_config['ac_relay_delay_ms'])

        self.machine.events.add_handler(self.system11_config['prefer_a_side_event'], self._prefer_a_side)
        self.log.info("Configuring System11 driver to prefer A side on event %s",
                      self.system11_config['prefer_a_side_event'])

        self.machine.events.add_handler(self.system11_config['prefer_c_side_event'], self._prefer_c_side)
        self.log.info("Configuring System11 driver to prefer C side on event %s",
                      self.system11_config['prefer_c_side_event'])

    def _prefer_a_side(self, **kwargs):
        del kwargs
        self.prefer_a_side = True
        self._enable_a_side()

    def _prefer_c_side(self, **kwargs):
        del kwargs
        self.prefer_a_side = False
        self._enable_c_side()

    def _validate_config(self):
        self.system11_config = self.machine.config_validator.validate_config(
            'system11', self.machine.config.get('system11', {}))

    def tick(self):
        """System11 main loop.

        Called based on the timer_tick event.
        """
        if self.prefer_a_side:
            if self.a_side_queue:
                self._service_a_side()
            elif self.c_side_queue:
                self._service_c_side()
            elif self.c_side_enabled and not self.c_side_active:
                self._enable_a_side()
        else:
            if self.c_side_queue:
                self._service_c_side()
            elif self.a_side_queue:
                self._service_a_side()
            elif self.a_side_enabled and not self.a_side_active:
                self._enable_c_side()

    def configure_switch(self, number: str, config: SwitchConfig, platform_config: dict):
        """Configure switch on system11 overlay."""
        return self.platform.configure_switch(number, config, platform_config)

    def validate_switch_section(self, switch, config: dict) -> dict:
        """Validate switch config for overlayed platform."""
        return self.platform.validate_switch_section(switch, config)

    async def get_hw_switch_states(self):
        """Get initial hardware state."""
        return await self.platform.get_hw_switch_states()

    def configure_driver(self, config: DriverConfig, number: str, platform_settings: dict):
        """Configure a driver on the system11 overlay.

        Args:
        ----
            config: Driver config dict
            number: Number of the driver.
            platform_settings: Platform specific config.
        """
        assert self.platform is not None
        orig_number = number

        if number and (number.lower().endswith('a') or number.lower().endswith('c')):

{'fraction_non_alphanumeric': 0.09105824446267433, 'fraction_numerical': 0.0076565490839485915, 'mean_word_length': 2.8436678928008408, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 14, 'https://': 0, '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': '5799380', 'n_tokens_mistral': 4287, 'n_tokens_neox': 4089, 'n_words': 1081}