包装io.BufferedIOBase，使其变为可搜索 [英] Wrap an io.BufferedIOBase such that it becomes seek-able

问题描述

我试图回答有关来自HTTP服务器的流音频的问题，然后使用 PyGame进行播放.我的代码大部分已经完成，但是遇到了一个错误，其中 PyGame音乐功能尝试在urllib.HTTPResponse对象上seek().

I was trying to craft a response to a question about streaming audio from a HTTP server, then play it with PyGame. I had the code mostly complete, but hit an error where the PyGame music functions tried to seek() on the urllib.HTTPResponse object.

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

According to the urlib docs, the urllib.HTTPResponse object (since v3.5) is an io.BufferedIOBase. I expected this would make the stream seek()able, however it does not.

是否有一种包装io.BufferedIOBase的方法，以使其足够聪明以缓冲足够的数据来处理查找操作?

Is there a way to wrap the io.BufferedIOBase such that it is smart enough to buffer enough data to handle the seek operation?

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 时，它将失败，并显示以下错误:

When this code runs, and Up is pushed, it fails with the error:

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流，以及对同一事物的各种其他重新实现.

I also tried re-opening the the io stream, and various other re-implementations of the same sort of thing.

推荐答案

如果可以使用requests模块(支持流式传输)代替urllib，那么可以使用包装器

If your fine with using the requests module (which supports streaming) instead of urllib, you could use a wrapper 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)

那我想你可以做这样的事情:

Then I guess you can do something like this:

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％有效，但请尝试一下.

I'm not sure this works 100%, but give it a try.

这篇关于包装io.BufferedIOBase，使其变为可搜索的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！