#!/usr/bin/env python3
#
#   Copyright 2020 - The Android Open Source Project
#
#   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.

import logging
import bluetooth_packets_python3 as bt_packets
import logging

from bluetooth_packets_python3 import hci_packets
from bluetooth_packets_python3.hci_packets import EventCode
from bluetooth_packets_python3 import l2cap_packets
from bluetooth_packets_python3.l2cap_packets import CommandCode, LeCommandCode
from bluetooth_packets_python3.l2cap_packets import ConfigurationResponseResult
from bluetooth_packets_python3.l2cap_packets import ConnectionResponseResult
from bluetooth_packets_python3.l2cap_packets import InformationRequestInfoType
from bluetooth_packets_python3.l2cap_packets import LeCreditBasedConnectionResponseResult


class HciMatchers(object):

    @staticmethod
    def CommandComplete(opcode):
        return lambda msg: HciMatchers._is_matching_command_complete(msg.payload, opcode)

    @staticmethod
    def ExtractMatchingCommandComplete(packet_bytes, opcode=None):
        return HciMatchers._extract_matching_command_complete(packet_bytes, opcode)

    @staticmethod
    def _is_matching_command_complete(packet_bytes, opcode=None):
        return HciMatchers._extract_matching_command_complete(packet_bytes, opcode) is not None

    @staticmethod
    def _extract_matching_command_complete(packet_bytes, opcode=None):
        event = HciMatchers._extract_matching_event(packet_bytes, EventCode.COMMAND_COMPLETE)
        if event is None:
            return None
        complete = hci_packets.CommandCompleteView(event)
        if opcode is None or complete is None:
            return complete
        else:
            if complete.GetCommandOpCode() != opcode:
                return None
            else:
                return complete

    @staticmethod
    def CommandStatus(opcode=None):
        return lambda msg: HciMatchers._is_matching_command_status(msg.payload, opcode)

    @staticmethod
    def ExtractMatchingCommandStatus(packet_bytes, opcode=None):
        return HciMatchers._extract_matching_command_complete(packet_bytes, opcode)

    @staticmethod
    def _is_matching_command_status(packet_bytes, opcode=None):
        return HciMatchers._extract_matching_command_status(packet_bytes, opcode) is not None

    @staticmethod
    def _extract_matching_command_status(packet_bytes, opcode=None):
        event = HciMatchers._extract_matching_event(packet_bytes, EventCode.COMMAND_STATUS)
        if event is None:
            return None
        complete = hci_packets.CommandStatusView(event)
        if opcode is None or complete is None:
            return complete
        else:
            if complete.GetCommandOpCode() != opcode:
                return None
            else:
                return complete

    @staticmethod
    def EventWithCode(event_code):
        return lambda msg: HciMatchers._is_matching_event(msg.payload, event_code)

    @staticmethod
    def ExtractEventWithCode(packet_bytes, event_code):
        return HciMatchers._extract_matching_event(packet_bytes, event_code)

    @staticmethod
    def _is_matching_event(packet_bytes, event_code):
        return HciMatchers._extract_matching_event(packet_bytes, event_code) is not None

    @staticmethod
    def _extract_matching_event(packet_bytes, event_code):
        event = hci_packets.EventView(bt_packets.PacketViewLittleEndian(list(packet_bytes)))
        if event is None:
            return None
        if event_code is not None and event.GetEventCode() != event_code:
            return None
        return event

    @staticmethod
    def LeEventWithCode(subevent_code):
        return lambda msg: HciMatchers._extract_matching_le_event(msg.payload, subevent_code) is not None

    @staticmethod
    def ExtractLeEventWithCode(packet_bytes, subevent_code):
        return HciMatchers._extract_matching_le_event(packet_bytes, subevent_code)

    @staticmethod
    def _extract_matching_le_event(packet_bytes, subevent_code):
        inner_event = HciMatchers._extract_matching_event(packet_bytes, hci_packets.EventCode.LE_META_EVENT)
        if inner_event is None:
            return None
        event = hci_packets.LeMetaEventView(inner_event)
        if event.GetSubeventCode() != subevent_code:
            return None
        return event

    @staticmethod
    def LeConnectionComplete():
        return lambda msg: HciMatchers._extract_le_connection_complete(msg.payload) is not None

    @staticmethod
    def ExtractLeConnectionComplete(packet_bytes):
        return HciMatchers._extract_le_connection_complete(packet_bytes)

    @staticmethod
    def _extract_le_connection_complete(packet_bytes):
        inner_event = HciMatchers._extract_matching_le_event(packet_bytes, hci_packets.SubeventCode.CONNECTION_COMPLETE)
        if inner_event is not None:
            return hci_packets.LeConnectionCompleteView(inner_event)

        inner_event = HciMatchers._extract_matching_le_event(packet_bytes,
                                                             hci_packets.SubeventCode.ENHANCED_CONNECTION_COMPLETE)
        if inner_event is not None:
            return hci_packets.LeEnhancedConnectionCompleteView(inner_event)

        return None

    @staticmethod
    def LogEventCode():
        return lambda event: logging.info("Received event: %x" % hci_packets.EventView(bt_packets.PacketViewLittleEndian(list(event.payload))).GetEventCode())

    @staticmethod
    def LinkKeyRequest():
        return lambda event: HciMatchers.EventWithCode(EventCode.LINK_KEY_REQUEST)

    @staticmethod
    def IoCapabilityRequest():
        return lambda event: HciMatchers.EventWithCode(EventCode.IO_CAPABILITY_REQUEST)

    @staticmethod
    def IoCapabilityResponse():
        return lambda event: HciMatchers.EventWithCode(EventCode.IO_CAPABILITY_RESPONSE)

    @staticmethod
    def UserPasskeyNotification():
        return lambda event: HciMatchers.EventWithCode(EventCode.USER_PASSKEY_NOTIFICATION)

    @staticmethod
    def UserPasskeyRequest():
        return lambda event: HciMatchers.EventWithCode(EventCode.USER_PASSKEY_REQUEST)

    @staticmethod
    def UserConfirmationRequest():
        return lambda event: HciMatchers.EventWithCode(EventCode.USER_CONFIRMATION_REQUEST)

    @staticmethod
    def RemoteHostSupportedFeaturesNotification():
        return lambda event: HciMatchers.EventWithCode(EventCode.REMOTE_HOST_SUPPORTED_FEATURES_NOTIFICATION)

    @staticmethod
    def LinkKeyNotification():
        return lambda event: HciMatchers.EventWithCode(EventCode.LINK_KEY_NOTIFICATION)

    @staticmethod
    def SimplePairingComplete():
        return lambda event: HciMatchers.EventWithCode(EventCode.SIMPLE_PAIRING_COMPLETE)

    @staticmethod
    def Disconnect():
        return lambda event: HciMatchers.EventWithCode(EventCode.DISCONNECT)

    @staticmethod
    def DisconnectionComplete():
        return lambda event: HciMatchers.EventWithCode(EventCode.DISCONNECTION_COMPLETE)

    @staticmethod
    def RemoteOobDataRequest():
        return lambda event: HciMatchers.EventWithCode(EventCode.REMOTE_OOB_DATA_REQUEST)

    @staticmethod
    def PinCodeRequest():
        return lambda event: HciMatchers.EventWithCode(EventCode.PIN_CODE_REQUEST)

    @staticmethod
    def LoopbackOf(packet):
        return HciMatchers.Exactly(hci_packets.LoopbackCommandBuilder(packet))

    @staticmethod
    def Exactly(packet):
        data = bytes(packet.Serialize())
        return lambda event: data == event.payload


class AdvertisingMatchers(object):

    @staticmethod
    def AdvertisingCallbackMsg(type, advertiser_id=None, status=None, data=None):
        return lambda event: True if event.message_type == type and (advertiser_id == None or advertiser_id == event.advertiser_id) \
            and (status == None or status == event.status) and (data == None or data == event.data) else False

    @staticmethod
    def AddressMsg(type, advertiser_id=None, address=None):
        return lambda event: True if event.message_type == type and (advertiser_id == None or advertiser_id == event.advertiser_id) \
            and (address == None or address == event.address) else False


class ScanningMatchers(object):

    @staticmethod
    def ScanningCallbackMsg(type, status=None, data=None):
        return lambda event: True if event.message_type == type and (status == None or status == event.status) \
            and (data == None or data == event.data) else False


class NeighborMatchers(object):

    @staticmethod
    def InquiryResult(address):
        return lambda msg: NeighborMatchers._is_matching_inquiry_result(msg.packet, address)

    @staticmethod
    def _is_matching_inquiry_result(packet, address):
        hci_event = HciMatchers.ExtractEventWithCode(packet, EventCode.INQUIRY_RESULT)
        if hci_event is None:
            return False
        inquiry_view = hci_packets.InquiryResultView(hci_event)
        if inquiry_view is None:
            return False
        results = inquiry_view.GetInquiryResults()
        return any((address == result.bd_addr for result in results))

    @staticmethod
    def InquiryResultwithRssi(address):
        return lambda msg: NeighborMatchers._is_matching_inquiry_result_with_rssi(msg.packet, address)

    @staticmethod
    def _is_matching_inquiry_result_with_rssi(packet, address):
        hci_event = HciMatchers.ExtractEventWithCode(packet, EventCode.INQUIRY_RESULT_WITH_RSSI)
        if hci_event is None:
            return False
        inquiry_view = hci_packets.InquiryResultWithRssiView(hci_event)
        if inquiry_view is None:
            return False
        results = inquiry_view.GetInquiryResults()
        return any((address == result.address for result in results))

    @staticmethod
    def ExtendedInquiryResult(address):
        return lambda msg: NeighborMatchers._is_matching_extended_inquiry_result(msg.packet, address)

    @staticmethod
    def _is_matching_extended_inquiry_result(packet, address):
        hci_event = HciMatchers.ExtractEventWithCode(packet, EventCode.EXTENDED_INQUIRY_RESULT)
        if hci_event is None:
            return False
        extended_view = hci_packets.ExtendedInquiryResultView(hci_event)
        if extended_view is None:
            return False
        return address == extended_view.GetAddress()


class L2capMatchers(object):

    @staticmethod
    def ConnectionRequest(psm):
        return lambda packet: L2capMatchers._is_matching_connection_request(packet, psm)

    @staticmethod
    def ConnectionResponse(scid):
        return lambda packet: L2capMatchers._is_matching_connection_response(packet, scid)

    @staticmethod
    def ConfigurationResponse(result=ConfigurationResponseResult.SUCCESS):
        return lambda packet: L2capMatchers._is_matching_configuration_response(packet, result)

    @staticmethod
    def ConfigurationRequest(cid=None):
        return lambda packet: L2capMatchers._is_matching_configuration_request_with_cid(packet, cid)

    @staticmethod
    def ConfigurationRequestWithErtm():
        return lambda packet: L2capMatchers._is_matching_configuration_request_with_ertm(packet)

    @staticmethod
    def ConfigurationRequestView(dcid):
        return lambda request_view: request_view.GetDestinationCid() == dcid

    @staticmethod
    def DisconnectionRequest(scid, dcid):
        return lambda packet: L2capMatchers._is_matching_disconnection_request(packet, scid, dcid)

    @staticmethod
    def DisconnectionResponse(scid, dcid):
        return lambda packet: L2capMatchers._is_matching_disconnection_response(packet, scid, dcid)

    @staticmethod
    def EchoResponse():
        return lambda packet: L2capMatchers._is_control_frame_with_code(packet, CommandCode.ECHO_RESPONSE)

    @staticmethod
    def CommandReject():
        return lambda packet: L2capMatchers._is_control_frame_with_code(packet, CommandCode.COMMAND_REJECT)

    @staticmethod
    def LeCommandReject():
        return lambda packet: L2capMatchers._is_le_control_frame_with_code(packet, LeCommandCode.COMMAND_REJECT)

    @staticmethod
    def LeConnectionParameterUpdateRequest():
        return lambda packet: L2capMatchers._is_le_control_frame_with_code(
            packet, LeCommandCode.CONNECTION_PARAMETER_UPDATE_REQUEST)

    @staticmethod
    def LeConnectionParameterUpdateResponse(result=l2cap_packets.ConnectionParameterUpdateResponseResult.ACCEPTED):
        return lambda packet: L2capMatchers._is_matching_connection_parameter_update_response(packet, result)

    @staticmethod
    def CreditBasedConnectionRequest(psm):
        return lambda packet: L2capMatchers._is_matching_credit_based_connection_request(packet, psm)

    @staticmethod
    def CreditBasedConnectionResponse(result=LeCreditBasedConnectionResponseResult.SUCCESS):
        return lambda packet: L2capMatchers._is_matching_credit_based_connection_response(packet, result)

    @staticmethod
    def CreditBasedConnectionResponseUsedCid():
        return lambda packet: L2capMatchers._is_matching_credit_based_connection_response(
            packet, LeCreditBasedConnectionResponseResult.SOURCE_CID_ALREADY_ALLOCATED
        ) or L2capMatchers._is_le_control_frame_with_code(packet, LeCommandCode.COMMAND_REJECT)

    @staticmethod
    def LeDisconnectionRequest(scid, dcid):
        return lambda packet: L2capMatchers._is_matching_le_disconnection_request(packet, scid, dcid)

    @staticmethod
    def LeDisconnectionResponse(scid, dcid):
        return lambda packet: L2capMatchers._is_matching_le_disconnection_response(packet, scid, dcid)

    @staticmethod
    def LeFlowControlCredit(cid):
        return lambda packet: L2capMatchers._is_matching_le_flow_control_credit(packet, cid)

    @staticmethod
    def SFrame(req_seq=None, f=None, s=None, p=None):
        return lambda packet: L2capMatchers._is_matching_supervisory_frame(packet, req_seq, f, s, p)

    @staticmethod
    def IFrame(tx_seq=None, payload=None, f=None):
        return lambda packet: L2capMatchers._is_matching_information_frame(packet, tx_seq, payload, f, fcs=False)

    @staticmethod
    def IFrameWithFcs(tx_seq=None, payload=None, f=None):
        return lambda packet: L2capMatchers._is_matching_information_frame(packet, tx_seq, payload, f, fcs=True)

    @staticmethod
    def IFrameStart(tx_seq=None, payload=None, f=None):
        return lambda packet: L2capMatchers._is_matching_information_start_frame(packet, tx_seq, payload, f, fcs=False)

    @staticmethod
    def Data(payload):
        return lambda packet: packet.GetPayload().GetBytes() == payload

    @staticmethod
    def FirstLeIFrame(payload, sdu_size):
        return lambda packet: L2capMatchers._is_matching_first_le_i_frame(packet, payload, sdu_size)

    # this is a hack - should be removed
    @staticmethod
    def PartialData(payload):
        return lambda packet: payload in packet.GetPayload().GetBytes()

    # this is a hack - should be removed
    @staticmethod
    def PacketPayloadRawData(payload):
        return lambda packet: payload in packet.payload

    # this is a hack - should be removed
    @staticmethod
    def PacketPayloadWithMatchingPsm(psm):
        return lambda packet: None if psm != packet.psm else packet

    # this is a hack - should be removed
    @staticmethod
    def PacketPayloadWithMatchingCid(cid):
        return lambda packet: None if cid != packet.fixed_cid else packet

    @staticmethod
    def ExtractBasicFrame(scid):
        return lambda packet: L2capMatchers._basic_frame_for(packet, scid)

    @staticmethod
    def ExtractBasicFrameWithFcs(scid):
        return lambda packet: L2capMatchers._basic_frame_with_fcs_for(packet, scid)

    @staticmethod
    def InformationRequestWithType(info_type):
        return lambda packet: L2capMatchers._information_request_with_type(packet, info_type)

    @staticmethod
    def InformationResponseExtendedFeatures(supports_ertm=None,
                                            supports_streaming=None,
                                            supports_fcs=None,
                                            supports_fixed_channels=None):
        return lambda packet: L2capMatchers._is_matching_information_response_extended_features(
            packet, supports_ertm, supports_streaming, supports_fcs, supports_fixed_channels)

    @staticmethod
    def _basic_frame(packet):
        if packet is None:
            return None
        return l2cap_packets.BasicFrameView(bt_packets.PacketViewLittleEndian(list(packet.payload)))

    @staticmethod
    def _basic_frame_with_fcs(packet):
        if packet is None:
            return None
        return l2cap_packets.BasicFrameWithFcsView(bt_packets.PacketViewLittleEndian(list(packet.payload)))

    @staticmethod
    def _basic_frame_for(packet, scid):
        frame = L2capMatchers._basic_frame(packet)
        if frame.GetChannelId() != scid:
            return None
        return frame

    @staticmethod
    def _basic_frame_with_fcs_for(packet, scid):
        frame = L2capMatchers._basic_frame(packet)
        if frame.GetChannelId() != scid:
            return None
        frame = L2capMatchers._basic_frame_with_fcs(packet)
        if frame is None:
            return None
        return frame

    @staticmethod
    def _information_frame(packet):
        standard_frame = l2cap_packets.StandardFrameView(packet)
        if standard_frame.GetFrameType() != l2cap_packets.FrameType.I_FRAME:
            return None
        return l2cap_packets.EnhancedInformationFrameView(standard_frame)

    @staticmethod
    def _information_frame_with_fcs(packet):
        standard_frame = l2cap_packets.StandardFrameWithFcsView(packet)
        if standard_frame is None:
            return None
        if standard_frame.GetFrameType() != l2cap_packets.FrameType.I_FRAME:
            return None
        return l2cap_packets.EnhancedInformationFrameWithFcsView(standard_frame)

    @staticmethod
    def _information_start_frame(packet):
        start_frame = L2capMatchers._information_frame(packet)
        if start_frame is None:
            return None
        return l2cap_packets.EnhancedInformationStartFrameView(start_frame)

    @staticmethod
    def _information_start_frame_with_fcs(packet):
        start_frame = L2capMatchers._information_frame_with_fcs(packet)
        if start_frame is None:
            return None
        return l2cap_packets.EnhancedInformationStartFrameWithFcsView(start_frame)

    @staticmethod
    def _supervisory_frame(packet):
        standard_frame = l2cap_packets.StandardFrameView(packet)
        if standard_frame.GetFrameType() != l2cap_packets.FrameType.S_FRAME:
            return None
        return l2cap_packets.EnhancedSupervisoryFrameView(standard_frame)

    @staticmethod
    def _is_matching_information_frame(packet, tx_seq, payload, f, fcs=False):
        if fcs:
            frame = L2capMatchers._information_frame_with_fcs(packet)
        else:
            frame = L2capMatchers._information_frame(packet)
        if frame is None:
            return False
        if tx_seq is not None and frame.GetTxSeq() != tx_seq:
            return False
        if payload is not None and frame.GetPayload().GetBytes() != payload:
            return False
        if f is not None and frame.GetF() != f:
            return False
        return True

    @staticmethod
    def _is_matching_information_start_frame(packet, tx_seq, payload, f, fcs=False):
        if fcs:
            frame = L2capMatchers._information_start_frame_with_fcs(packet)
        else:
            frame = L2capMatchers._information_start_frame(packet)
        if frame is None:
            return False
        if tx_seq is not None and frame.GetTxSeq() != tx_seq:
            return False
        if payload is not None and frame.GetPayload().GetBytes() != payload:
            return False
        if f is not None and frame.GetF() != f:
            return False
        return True

    @staticmethod
    def _is_matching_supervisory_frame(packet, req_seq, f, s, p):
        frame = L2capMatchers._supervisory_frame(packet)
        if frame is None:
            return False
        if req_seq is not None and frame.GetReqSeq() != req_seq:
            return False
        if f is not None and frame.GetF() != f:
            return False
        if s is not None and frame.GetS() != s:
            return False
        if p is not None and frame.GetP() != p:
            return False
        return True

    @staticmethod
    def _is_matching_first_le_i_frame(packet, payload, sdu_size):
        first_le_i_frame = l2cap_packets.FirstLeInformationFrameView(packet)
        return first_le_i_frame.GetPayload().GetBytes() == payload and first_le_i_frame.GetL2capSduLength() == sdu_size

    @staticmethod
    def _control_frame(packet):
        if packet.GetChannelId() != 1:
            return None
        return l2cap_packets.ControlView(packet.GetPayload())

    @staticmethod
    def _le_control_frame(packet):
        if packet.GetChannelId() != 5:
            return None
        return l2cap_packets.LeControlView(packet.GetPayload())

    @staticmethod
    def control_frame_with_code(packet, code):
        frame = L2capMatchers._control_frame(packet)
        if frame is None or frame.GetCode() != code:
            return None
        return frame

    @staticmethod
    def le_control_frame_with_code(packet, code):
        frame = L2capMatchers._le_control_frame(packet)
        if frame is None or frame.GetCode() != code:
            return None
        return frame

    @staticmethod
    def _is_control_frame_with_code(packet, code):
        return L2capMatchers.control_frame_with_code(packet, code) is not None

    @staticmethod
    def _is_le_control_frame_with_code(packet, code):
        return L2capMatchers.le_control_frame_with_code(packet, code) is not None

    @staticmethod
    def _is_matching_connection_request(packet, psm):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.CONNECTION_REQUEST)
        if frame is None:
            return False
        request = l2cap_packets.ConnectionRequestView(frame)
        return request.GetPsm() == psm

    @staticmethod
    def _is_matching_connection_response(packet, scid):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.CONNECTION_RESPONSE)
        if frame is None:
            return False
        response = l2cap_packets.ConnectionResponseView(frame)
        return response.GetSourceCid() == scid and response.GetResult(
        ) == ConnectionResponseResult.SUCCESS and response.GetDestinationCid() != 0

    @staticmethod
    def _is_matching_configuration_request_with_cid(packet, cid=None):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.CONFIGURATION_REQUEST)
        if frame is None:
            return False
        request = l2cap_packets.ConfigurationRequestView(frame)
        dcid = request.GetDestinationCid()
        return cid is None or cid == dcid

    @staticmethod
    def _is_matching_configuration_request_with_ertm(packet):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.CONFIGURATION_REQUEST)
        if frame is None:
            return False
        request = l2cap_packets.ConfigurationRequestView(frame)
        config_bytes = request.GetBytes()
        # TODO(b/153189503): Use packet struct parser.
        return b"\x04\x09\x03" in config_bytes

    @staticmethod
    def _is_matching_configuration_response(packet, result=ConfigurationResponseResult.SUCCESS):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.CONFIGURATION_RESPONSE)
        if frame is None:
            return False
        response = l2cap_packets.ConfigurationResponseView(frame)
        return response.GetResult() == result

    @staticmethod
    def _is_matching_disconnection_request(packet, scid, dcid):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.DISCONNECTION_REQUEST)
        if frame is None:
            return False
        request = l2cap_packets.DisconnectionRequestView(frame)
        return request.GetSourceCid() == scid and request.GetDestinationCid() == dcid

    @staticmethod
    def _is_matching_disconnection_response(packet, scid, dcid):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.DISCONNECTION_RESPONSE)
        if frame is None:
            return False
        response = l2cap_packets.DisconnectionResponseView(frame)
        return response.GetSourceCid() == scid and response.GetDestinationCid() == dcid

    @staticmethod
    def _is_matching_le_disconnection_response(packet, scid, dcid):
        frame = L2capMatchers.le_control_frame_with_code(packet, LeCommandCode.DISCONNECTION_RESPONSE)
        if frame is None:
            return False
        response = l2cap_packets.LeDisconnectionResponseView(frame)
        return response.GetSourceCid() == scid and response.GetDestinationCid() == dcid

    @staticmethod
    def _is_matching_le_disconnection_request(packet, scid, dcid):
        frame = L2capMatchers.le_control_frame_with_code(packet, LeCommandCode.DISCONNECTION_REQUEST)
        if frame is None:
            return False
        request = l2cap_packets.LeDisconnectionRequestView(frame)
        return request.GetSourceCid() == scid and request.GetDestinationCid() == dcid

    @staticmethod
    def _is_matching_le_flow_control_credit(packet, cid):
        frame = L2capMatchers.le_control_frame_with_code(packet, LeCommandCode.LE_FLOW_CONTROL_CREDIT)
        if frame is None:
            return False
        request = l2cap_packets.LeFlowControlCreditView(frame)
        return request.GetCid() == cid

    @staticmethod
    def _information_request_with_type(packet, info_type):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.INFORMATION_REQUEST)
        if frame is None:
            return None
        request = l2cap_packets.InformationRequestView(frame)
        if request.GetInfoType() != info_type:
            return None
        return request

    @staticmethod
    def _information_response_with_type(packet, info_type):
        frame = L2capMatchers.control_frame_with_code(packet, CommandCode.INFORMATION_RESPONSE)
        if frame is None:
            return None
        response = l2cap_packets.InformationResponseView(frame)
        if response.GetInfoType() != info_type:
            return None
        return response

    @staticmethod
    def _is_matching_information_response_extended_features(packet, supports_ertm, supports_streaming, supports_fcs,
                                                            supports_fixed_channels):
        frame = L2capMatchers._information_response_with_type(packet,
                                                              InformationRequestInfoType.EXTENDED_FEATURES_SUPPORTED)
        if frame is None:
            return False
        features = l2cap_packets.InformationResponseExtendedFeaturesView(frame)
        if supports_ertm is not None and features.GetEnhancedRetransmissionMode() != supports_ertm:
            return False
        if supports_streaming is not None and features.GetStreamingMode != supports_streaming:
            return False
        if supports_fcs is not None and features.GetFcsOption() != supports_fcs:
            return False
        if supports_fixed_channels is not None and features.GetFixedChannels() != supports_fixed_channels:
            return False
        return True

    @staticmethod
    def _is_matching_connection_parameter_update_response(packet, result):
        frame = L2capMatchers.le_control_frame_with_code(packet, LeCommandCode.CONNECTION_PARAMETER_UPDATE_RESPONSE)
        if frame is None:
            return False
        return l2cap_packets.ConnectionParameterUpdateResponseView(frame).GetResult() == result

    @staticmethod
    def _is_matching_credit_based_connection_request(packet, psm):
        frame = L2capMatchers.le_control_frame_with_code(packet, LeCommandCode.LE_CREDIT_BASED_CONNECTION_REQUEST)
        if frame is None:
            return False
        request = l2cap_packets.LeCreditBasedConnectionRequestView(frame)
        return request.GetLePsm() == psm

    @staticmethod
    def _is_matching_credit_based_connection_response(packet, result):
        frame = L2capMatchers.le_control_frame_with_code(packet, LeCommandCode.LE_CREDIT_BASED_CONNECTION_RESPONSE)
        if frame is None:
            return False
        response = l2cap_packets.LeCreditBasedConnectionResponseView(frame)
        return response.GetResult() == result and (result != LeCreditBasedConnectionResponseResult.SUCCESS or
                                                   response.GetDestinationCid() != 0)

    @staticmethod
    def LinkSecurityInterfaceCallbackEvent(type):
        return lambda event: True if event.event_type == type else False


class SecurityMatchers(object):

    @staticmethod
    def UiMsg(type, address=None):
        return lambda event: True if event.message_type == type and (address == None or address == event.peer) else False

    @staticmethod
    def BondMsg(type, address=None, reason=None):
        return lambda event: True if event.message_type == type and (address == None or address == event.peer) and (reason == None or reason == event.reason) else False

    @staticmethod
    def HelperMsg(type, address=None):
        return lambda event: True if event.message_type == type and (address == None or address == event.peer) else False


class IsoMatchers(object):

    @staticmethod
    def Data(payload):
        return lambda packet: packet.payload == payload

    @staticmethod
    def PacketPayloadWithMatchingCisHandle(cis_handle):
        return lambda packet: None if cis_handle != packet.handle else packet