(request: WSGIRequest)
return HttpResponse("Connected. Set output device to activate.")
Connect to a given bluetooth device.
Connect to a given bluetooth device.
Connect to a given bluetooth device .
def connect_bluetooth(request: WSGIRequest) -> HttpResponse:
    """Connect to a given bluetooth device."""
    address = request.POST.get("address")
    if address is None or address == "":
        return HttpResponseBadRequest("No device selected")
    bluetoothctl = _start_bluetoothctl()
    if not bluetoothctl:
        return HttpResponseBadRequest("Stop scanning before connecting")

    assert bluetoothctl.stdin
    error = ""

    # A Function that acts as a timeout for unexpected errors (or timeouts)
    def _timeout() -> None:
        nonlocal error
        time.sleep(20)
        error = "Timed out"
        if bluetoothctl is not None:
            _stop_bluetoothctl(bluetoothctl)

    Thread(target=_timeout).start()

    # Sometimes, pairing hangs forever. Since connecting alone is enough, skip pairing.
    # bluetoothctl.stdin.write(b"pair " + address.encode() + b"\n")
    # bluetoothctl.stdin.flush()
    # while True:
    #     line = _get_bluetoothctl_line(bluetoothctl)
    #     if not line:
    #         break
    #     if re.match(".*Device " + address + " not available", line):
    #         error = "Device unavailable"
    #         break
    #     if re.match(".*Failed to pair: org.bluez.Error.AlreadyExists", line):
    #         break
    #     if re.match(".*Pairing successful", line):
    #         break

    # if error:
    #     _stop_bluetoothctl()
    #     return HttpResponseBadRequest(error)

    bluetoothctl.stdin.write(b"connect " + address.encode() + b"\n")
    bluetoothctl.stdin.flush()
    while True:
        line = _get_bluetoothctl_line(bluetoothctl)
        if not line:
            break
        if re.match(".*Device " + address + " not available", line):
            error = "Device unavailable"
            break
        if re.match(".*Failed to connect: org.bluez.Error.Failed", line):
            error = "Connect Failed"
            break
        if re.match(".*Failed to connect: org.bluez.Error.InProgress", line):
            error = "Connect in progress"
            break
        if re.match(".*Connection successful", line):
            break
    # trust the device to automatically reconnect when it is available again
    bluetoothctl.stdin.write(b"trust " + address.encode() + b"\n")
    bluetoothctl.stdin.flush()

    _stop_bluetoothctl(bluetoothctl)
    if error:
        return HttpResponseBadRequest(error)

    return HttpResponse("Connected. Set output device to activate.")
def connect_bluetooth ( request : WSGIRequest ) -> HttpResponse : address = request . POST . get ( "address" ) if address is None or address == "" : return HttpResponseBadRequest ( "No device selected" ) bluetoothctl = _start_bluetoothctl ( ) if not bluetoothctl : return HttpResponseBadRequest ( "Stop scanning before connecting" ) assert bluetoothctl . stdin error = "" # A Function that acts as a timeout for unexpected errors (or timeouts) def _timeout ( ) -> None : nonlocal error time . sleep ( 20 ) error = "Timed out" if bluetoothctl is not None : _stop_bluetoothctl ( bluetoothctl ) Thread ( target = _timeout ) . start ( ) # Sometimes, pairing hangs forever. Since connecting alone is enough, skip pairing. # bluetoothctl.stdin.write(b"pair " + address.encode() + b"\n") # bluetoothctl.stdin.flush() # while True: #     line = _get_bluetoothctl_line(bluetoothctl) #     if not line: #         break #     if re.match(".*Device " + address + " not available", line): #         error = "Device unavailable" #         break #     if re.match(".*Failed to pair: org.bluez.Error.AlreadyExists", line): #         break #     if re.match(".*Pairing successful", line): #         break # if error: #     _stop_bluetoothctl() #     return HttpResponseBadRequest(error) bluetoothctl . stdin . write ( b"connect " + address . encode ( ) + b"\n" ) bluetoothctl . stdin . flush ( ) while True : line = _get_bluetoothctl_line ( bluetoothctl ) if not line : break if re . match ( ".*Device " + address + " not available" , line ) : error = "Device unavailable" break if re . match ( ".*Failed to connect: org.bluez.Error.Failed" , line ) : error = "Connect Failed" break if re . match ( ".*Failed to connect: org.bluez.Error.InProgress" , line ) : error = "Connect in progress" break if re . match ( ".*Connection successful" , line ) : break # trust the device to automatically reconnect when it is available again bluetoothctl . stdin . write ( b"trust " + address . encode ( ) + b"\n" ) bluetoothctl . stdin . flush ( ) _stop_bluetoothctl ( bluetoothctl ) if error : return HttpResponseBadRequest ( error ) return HttpResponse ( "Connected. Set output device to activate." )