python-3.x - 包装io.BufferedIOBase，使其变为可搜索

Question

我试图回答有关从HTTP服务器流音频的问题，然后使用PyGame播放它。我的代码大部分已经完成，但是在urllib.HTTPResponse对象上的PyGame music functions试图seek()时遇到了一个错误。

根据urlib文档，urllib.HTTPResponse对象（自v3.5起）是io.BufferedIOBase。我希望这会使流seek（）成为可能，但事实并非如此。

有没有办法包装io.BufferedIOBase，使其足够聪明以缓冲足够的数据来处理搜索操作？

import pygame
import urllib.request
import io

# Window size
WINDOW_WIDTH  = 400
WINDOW_HEIGHT = 400
# background colour
SKY_BLUE      = (161, 255, 254)

### Begin the streaming of a file
### Return the urlib.HTTPResponse, a file-like-object
def openURL( url ):
    result = None

    try:
        http_response = urllib.request.urlopen( url )
        print( "streamHTTP() - Fetching URL [%s]" % ( http_response.geturl() ) )
        print( "streamHTTP() - Response Status [%d] / [%s]" % ( http_response.status, http_response.reason ) )
        result = http_response
    except:
        print( "streamHTTP() - Error Fetching URL [%s]" % ( url ) )

    return result


### MAIN
pygame.init()
window  = pygame.display.set_mode( ( WINDOW_WIDTH, WINDOW_HEIGHT ) )
pygame.display.set_caption("Music Streamer")


clock = pygame.time.Clock()
done = False
while not done:

    # Handle user-input
    for event in pygame.event.get():
        if ( event.type == pygame.QUIT ):
            done = True
    # Keys
    keys = pygame.key.get_pressed()
    if ( keys[pygame.K_UP] ):
        if ( pygame.mixer.music.get_busy() ):
            print("busy")
        else:
            print("play")
            remote_music = openURL( 'http://127.0.0.1/example.wav' )
            if ( remote_music != None and remote_music.status == 200 ):
                pygame.mixer.music.load( io.BufferedReader( remote_music ) )
                pygame.mixer.music.play()

    # Re-draw the screen
    window.fill( SKY_BLUE )

    # Update the window, but not more than 60fps
    pygame.display.flip()
    clock.tick_busy_loop( 60 )

pygame.quit()

当此代码运行并按下Up时，它将失败并显示以下错误：

streamHTTP() - Fetching URL [http://127.0.0.1/example.wav]
streamHTTP() - Response Status [200] / [OK]
io.UnsupportedOperation: seek
io.UnsupportedOperation: File or stream is not seekable.
io.UnsupportedOperation: seek
io.UnsupportedOperation: File or stream is not seekable.
Traceback (most recent call last):
  File "./sound_stream.py", line 57, in <module>
    pygame.mixer.music.load( io.BufferedReader( remote_music ) )
pygame.error: Unknown WAVE format

我还尝试重新打开io流，以及对同一事物的各种其他重新实现。

Answer 1

如果使用requests模块（支持流传输）而不是urllib很好，则可以使用包装器like this：

class ResponseStream(object):
    def __init__(self, request_iterator):
        self._bytes = BytesIO()
        self._iterator = request_iterator

    def _load_all(self):
        self._bytes.seek(0, SEEK_END)
        for chunk in self._iterator:
            self._bytes.write(chunk)

    def _load_until(self, goal_position):
        current_position = self._bytes.seek(0, SEEK_END)
        while current_position < goal_position:
            try:
                current_position = self._bytes.write(next(self._iterator))
            except StopIteration:
                break

    def tell(self):
        return self._bytes.tell()

    def read(self, size=None):
        left_off_at = self._bytes.tell()
        if size is None:
            self._load_all()
        else:
            goal_position = left_off_at + size
            self._load_until(goal_position)

        self._bytes.seek(left_off_at)
        return self._bytes.read(size)

    def seek(self, position, whence=SEEK_SET):
        if whence == SEEK_END:
            self._load_all()
        else:
            self._bytes.seek(position, whence)

然后，我想您可以执行以下操作：

WINDOW_WIDTH  = 400
WINDOW_HEIGHT = 400
SKY_BLUE      = (161, 255, 254)
URL           = 'http://localhost:8000/example.wav'

pygame.init()
window  = pygame.display.set_mode( ( WINDOW_WIDTH, WINDOW_HEIGHT ) )
pygame.display.set_caption("Music Streamer")
clock = pygame.time.Clock()
done = False
font = pygame.font.SysFont(None, 32)
state = 0

def play_music():
    response = requests.get(URL, stream=True)
    if (response.status_code == 200):
        stream = ResponseStream(response.iter_content(64))
        pygame.mixer.music.load(stream)
        pygame.mixer.music.play()
    else:
        state = 0

while not done:

    for event in pygame.event.get():
        if ( event.type == pygame.QUIT ):
            done = True

        if event.type == pygame.KEYDOWN and state == 0:
            Thread(target=play_music).start()
            state = 1

    window.fill( SKY_BLUE )
    window.blit(font.render(str(pygame.time.get_ticks()), True, (0,0,0)), (32, 32))
    pygame.display.flip()
    clock.tick_busy_loop( 60 )

pygame.quit()

使用Thread开始流式传输。

我不确定这是否100％有效，但请尝试一下。