<commit_before>import sys

from functools import partial

import asyncio
import grpc

import mpf.platforms.visual_pinball_engine.visual_pinball_engine
from mpf.platforms.visual_pinball_engine import platform_pb2_grpc
from mpf.platforms.visual_pinball_engine.coils_pb2 import ConfigureHardwareRuleRequest
from mpf.platforms.visual_pinball_engine.get_plaform_details_pb2 import GetPlatformDetailsResponse
from mpf.platforms.visual_pinball_engine.switch_pb2 import SwitchChanges

from mpf.tests.MpfTestCase import MpfTestCase


class VpeSimulation(platform_pb2_grpc.HardwarePlatformServicer):

    def __init__(self, switches):
        self.switches = switches
        self.change_queue = None
        self.lights = {}
        self.coils = {}
        self.rules = {}

    def init_async(self):
        self.change_queue = asyncio.Queue()

    async def GetPlatformDetails(self, request, context):
        return GetPlatformDetailsResponse(
            known_switches_with_initial_state=self.switches,
            known_lights=["light-0", "light-1"],
            known_coils=["0", "1", "2"])

    def set_switch(self, switch, state):
        self.switches[switch] = state
        self.change_queue.put_nowait((switch, state))

    def GetSwitchChanges(self, request, context):
        class AsyncIterator:
            async def __anext__(self_inner):
                changed_switch = await self.change_queue.get()
                return SwitchChanges(switch_number=changed_switch[0], switch_state=changed_switch[1])

        return AsyncIterator()

    async def LightFade(self, request, context):
        for fade in request.fades:
            if request.common_fade_ms > 0:
                self.lights[fade.light_number] = (asyncio.get_event_loop().time() +
                                                  (request.common_fade_ms / 1000.0), fade.target_brightness)
            else:
                self.lights[fade.light_number] = fade.target_brightness

    async def CoilPulse(self, request, context):
        self.coils[request.coil_number] = "pulsed-{}-{}".format(request.pulse_ms, request.pulse_power)

    async def CoilEnable(self, request, context):
        self.coils[request.coil_number] = "enabled-{}-{}-{}".format(request.pulse_ms, request.pulse_power, request.hold_power)

    async def CoilDisable(self, request, context):
        self.coils[request.coil_number] = "disabled"

    async def ConfigureHardwareRule(self, request, context):
        self.rules["{}-{}".format(request.coil_number, request.switch_number)] = request

    async def RemoveHardwareRule(self, request, context):
        del self.rules["{}-{}".format(request.coil_number, request.switch_number)]



class GrpcAsyncTestChannel(grpc.Channel):

    def __init__(self, simulator):
        super().__init__()
        self.simulator = simulator

    def _get_method(self, method):
        parts = method.split("/")
        assert len(parts) == 3
        return getattr(self.simulator, parts[2])

    def subscribe(self, callback, try_to_connect=False):
        pass

    def unsubscribe(self, callback):
        pass

    def _async_call(self, method, request):
        return method(request=request, context=[])

    def _async_stream_call(self, method, request):
        coroutine = method(request=request, context=[])
        class Stream:
            def read(self):
                return coroutine.__anext__()

        return Stream()

    def unary_unary(self, method, request_serializer=None, response_deserializer=None):
        method = self._get_method(method)
        return partial(self._async_call, method)

    def unary_stream(self, method, request_serializer=None, response_deserializer=None):
        method = self._get_method(method)
        return partial(self._async_stream_call, method)

    def stream_unary(self, method, request_serializer=None, response_deserializer=None):
        print(method)
        return method()

    def stream_stream(self, method, request_serializer=None, response_deserializer=None):
        pass

    def close(self):
        pass

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


class TestVPE(MpfTestCase):

    def get_config_file(self):
        return 'config.yaml'

    def get_machine_path(self):
        return 'tests/machine_files/vpe/'

    async def _connect_to_mock_server(self, host, port):
        channel = GrpcAsyncTestChannel(self.simulator)
        return platform_pb2_grpc.HardwarePlatformStub(channel)

    def _mock_loop(self):
        self.simulator.init_async()

    def setUp(self):
        if sys.version_info < (3, 6):
            self.skipTest("Test requires Python 3.6+")
            return

        self.simulator = VpeSimulation({"0": True, "3": False, "6": False})
        mpf.platforms.visual_pinball_engine.visual_pinball_engine.VisualPinballEnginePlatform.connect = self._connect_to_mock_server
        super().setUp()

    def get_platform(self):
        return False

    def test_vpe(self):
        self.assertSwitchState("s_sling", True)
        self.assertSwitchState("s_flipper", False)
        self.assertSwitchState("s_test", False)
        self.simulator.set_switch("6", True)
        self.advance_time_and_run(.1)
        self.assertSwitchState("s_test", True)<commit_msg>move import behind version check to prevent test failure in python 3.5<commit_after>import sys

from functools import partial

import asyncio
import grpc

from mpf.platforms.visual_pinball_engine import platform_pb2_grpc
from mpf.platforms.visual_pinball_engine.coils_pb2 import ConfigureHardwareRuleRequest
from mpf.platforms.visual_pinball_engine.get_plaform_details_pb2 import GetPlatformDetailsResponse
from mpf.platforms.visual_pinball_engine.switch_pb2 import SwitchChanges

from mpf.tests.MpfTestCase import MpfTestCase


class VpeSimulation(platform_pb2_grpc.HardwarePlatformServicer):

    def __init__(self, switches):
        self.switches = switches
        self.change_queue = None
        self.lights = {}
        self.coils = {}
        self.rules = {}

    def init_async(self):
        self.change_queue = asyncio.Queue()

    async def GetPlatformDetails(self, request, context):
        return GetPlatformDetailsResponse(
            known_switches_with_initial_state=self.switches,
            known_lights=["light-0", "light-1"],
            known_coils=["0", "1", "2"])

    def set_switch(self, switch, state):
        self.switches[switch] = state
        self.change_queue.put_nowait((switch, state))

    def GetSwitchChanges(self, request, context):
        class AsyncIterator:
            async def __anext__(self_inner):
                changed_switch = await self.change_queue.get()
                return SwitchChanges(switch_number=changed_switch[0], switch_state=changed_switch[1])

        return AsyncIterator()

    async def LightFade(self, request, context):
        for fade in request.fades:
            if request.common_fade_ms > 0:
                self.lights[fade.light_number] = (asyncio.get_event_loop().time() +
                                                  (request.common_fade_ms / 1000.0), fade.target_brightness)
            else:
                self.lights[fade.light_number] = fade.target_brightness

    async def CoilPulse(self, request, context):
        self.coils[request.coil_number] = "pulsed-{}-{}".format(request.pulse_ms, request.pulse_power)

    async def CoilEnable(self, request, context):
        self.coils[request.coil_number] = "enabled-{}-{}-{}".format(request.pulse_ms, request.pulse_power, request.hold_power)

    async def CoilDisable(self, request, context):
        self.coils[request.coil_number] = "disabled"

    async def ConfigureHardwareRule(self, request, context):
        self.rules["{}-{}".format(request.coil_number, request.switch_number)] = request

    async def RemoveHardwareRule(self, request, context):
        del self.rules["{}-{}".format(request.coil_number, request.switch_number)]



class GrpcAsyncTestChannel(grpc.Channel):

    def __init__(self, simulator):
        super().__init__()
        self.simulator = simulator

    def _get_method(self, method):
        parts = method.split("/")
        assert len(parts) == 3
        return getattr(self.simulator, parts[2])

    def subscribe(self, callback, try_to_connect=False):
        pass

    def unsubscribe(self, callback):
        pass

    def _async_call(self, method, request):
        return method(request=request, context=[])

    def _async_stream_call(self, method, request):
        coroutine = method(request=request, context=[])
        class Stream:
            def read(self):
                return coroutine.__anext__()

        return Stream()

    def unary_unary(self, method, request_serializer=None, response_deserializer=None):
        method = self._get_method(method)
        return partial(self._async_call, method)

    def unary_stream(self, method, request_serializer=None, response_deserializer=None):
        method = self._get_method(method)
        return partial(self._async_stream_call, method)

    def stream_unary(self, method, request_serializer=None, response_deserializer=None):
        print(method)
        return method()

    def stream_stream(self, method, request_serializer=None, response_deserializer=None):
        pass

    def close(self):
        pass

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


class TestVPE(MpfTestCase):

    def get_config_file(self):
        return 'config.yaml'

    def get_machine_path(self):
        return 'tests/machine_files/vpe/'

    async def _connect_to_mock_server(self, host, port):
        channel = GrpcAsyncTestChannel(self.simulator)
        return platform_pb2_grpc.HardwarePlatformStub(channel)

    def _mock_loop(self):
        self.simulator.init_async()

    def setUp(self):
        if sys.version_info < (3, 6):
            self.skipTest("Test requires Python 3.6+")
            return

        self.simulator = VpeSimulation({"0": True, "3": False, "6": False})
        import mpf.platforms.visual_pinball_engine.visual_pinball_engine
        mpf.platforms.visual_pinball_engine.visual_pinball_engine.VisualPinballEnginePlatform.connect = self._connect_to_mock_server
        super().setUp()

    def get_platform(self):
        return False

    def test_vpe(self):
        self.assertSwitchState("s_sling", True)
        self.assertSwitchState("s_flipper", False)
        self.assertSwitchState("s_test", False)
        self.simulator.set_switch("6", True)
        self.advance_time_and_run(.1)
        self.assertSwitchState("s_test", True)
{'fraction_non_alphanumeric': 0.100496394118198, 'fraction_numerical': 0.0058068745902407045, 'mean_word_length': 3.1339527680991095, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 5, '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': '2872455', 'n_tokens_mistral': 3264, 'n_tokens_neox': 3123, 'n_words': 614}