<commit_before>import typing
from opentrons.hardware_control import ThreadManager

from . import Command, CompletedCommand, complete_command
from . base_executor import CommandExecutor
from ..errors import UnsupportedCommandException
from ..models import CommandDataType, CommandName


COMMAND_HANDLER = typing.Callable[
    [ThreadManager, CommandDataType], typing.Awaitable]


async def home_all_motors(hardware, *args):
    await hardware.home()


async def toggle_lights(hardware: ThreadManager, *args):
    light_state = hardware.get_lights()
    await hardware.set_lights(rails=not light_state.get('rails', False))


class HardwareExecutor(CommandExecutor):
    """A command executor that executes direct hardware commands"""

    COMMAND_TO_FUNC: typing.Dict[CommandName, COMMAND_HANDLER] = {
        CommandName.home_all_motors: home_all_motors,
        CommandName.toggle_lights: toggle_lights
    }

    def __init__(self,
                 hardware: ThreadManager,
                 command_filter: typing.Optional[typing.Set[CommandName]]):
        """
        Constructor

        :param hardware: hardware api access
        :param command_filter: a set of supported commands. All commands will
            be supported if None
        """
        self._hardware = hardware
        self._command_filter = command_filter

    async def execute(self, command: Command) -> CompletedCommand:
        func = self.get_handler(command.content.name)
        if func:
            await func(self._hardware, command.content.data)
        else:
            raise UnsupportedCommandException(
                f"Command '{command.content.name}' is not supported."
            )
        return complete_command(command, "executed")

    def get_handler(self, command_name: CommandName) \
            -> typing.Optional[COMMAND_HANDLER]:
        """Get the handler for the command type"""
        if self._command_filter is not None:
            if command_name not in self._command_filter:
                return None
        return self.COMMAND_TO_FUNC.get(command_name)


class DefaultHardwareExecutor(HardwareExecutor):
    """The default command executor"""
    def __init__(self, hardware: ThreadManager):
        super().__init__(hardware, {CommandName.home_all_motors,
                                    CommandName.home_pipette,
                                    CommandName.toggle_lights})<commit_msg>fix(robot-server): fix a merge race condition. (#6030)<commit_after>import typing
from opentrons.hardware_control import ThreadManager

from . import Command, CompletedCommand
from . base_executor import CommandExecutor
from .command import CommandResult
from ..errors import UnsupportedCommandException
from ..models import CommandDataType, CommandName
from robot_server.util import duration


COMMAND_HANDLER = typing.Callable[
    [ThreadManager, CommandDataType], typing.Awaitable]


async def home_all_motors(hardware, *args):
    await hardware.home()


async def toggle_lights(hardware: ThreadManager, *args):
    light_state = hardware.get_lights()
    await hardware.set_lights(rails=not light_state.get('rails', False))


class HardwareExecutor(CommandExecutor):
    """A command executor that executes direct hardware commands"""

    COMMAND_TO_FUNC: typing.Dict[CommandName, COMMAND_HANDLER] = {
        CommandName.home_all_motors: home_all_motors,
        CommandName.toggle_lights: toggle_lights
    }

    def __init__(self,
                 hardware: ThreadManager,
                 command_filter: typing.Optional[typing.Set[CommandName]]):
        """
        Constructor

        :param hardware: hardware api access
        :param command_filter: a set of supported commands. All commands will
            be supported if None
        """
        self._hardware = hardware
        self._command_filter = command_filter

    async def execute(self, command: Command) -> CompletedCommand:
        func = self.get_handler(command.content.name)
        if func:
            with duration() as timed:
                await func(self._hardware, command.content.data)
        else:
            raise UnsupportedCommandException(
                f"Command '{command.content.name}' is not supported."
            )
        return CompletedCommand(
            content=command.content,
            meta=command.meta,
            result=CommandResult(started_at=timed.start,
                                 completed_at=timed.end)
        )

    def get_handler(self, command_name: CommandName) \
            -> typing.Optional[COMMAND_HANDLER]:
        """Get the handler for the command type"""
        if self._command_filter is not None:
            if command_name not in self._command_filter:
                return None
        return self.COMMAND_TO_FUNC.get(command_name)


class DefaultHardwareExecutor(HardwareExecutor):
    """The default command executor"""
    def __init__(self, hardware: ThreadManager):
        super().__init__(hardware, {CommandName.home_all_motors,
                                    CommandName.home_pipette,
                                    CommandName.toggle_lights})
{'fraction_non_alphanumeric': 0.07812195597116696, 'fraction_numerical': 0.0007792713812585232, 'mean_word_length': 2.7447118891320206, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1894112', 'n_tokens_mistral': 1340, 'n_tokens_neox': 1306, 'n_words': 333}