#Python 3.6 No module named pip | Explore Tumblr posts and blogs

blubberquark · 6 years ago

Text

Pygame and AsyncIO Part V: Twitch Integration

In the last part of this tutorial sequence, we will implement twitch.tv integration in a pygame game. If you have previously thought that all this async stuff was a bit too elaborate for too little gain, then you might be pleasantly surpised by a real-world example of two-way online interaction. This is not exactly netcode for a fighting game, or a MMORPG, or a lock-step simulation you can use in an RTS game or a MOBA, but it is what asyncio is good for.

To run the examples, you need pygame, python 3.6, asyncio, and the irc module (https://pypi.org/project/irc/). Install it with python3 -m pip install irc

Some of the code in here will be twitch-specific, but twitch chat is based on good old IRC, and you can take out these bits and add chat interaction with any IRC network if you want to. Also, I won’t explain to you how to stream your gameplay to twitch, but I used OBS to test out my examples and they work.

Twitch Plays Wormy.py

This is based on https://inventwithpython.com/pygame/chapter6.html.

import pygame import asyncio import irc, irc.client_aio import sys if len(sys.argv)!=3: print("please get a twitch auth token from https://twitchapps.com/tmi/") sys.exit(1) username=sys.argv[1] token=sys.argv[2] irc_pw=token.strip() irc_user=username.strip() assert(irc_pw.startswith("oauth:"))

To start the game, you pass your twitch username to the game, like so python3 game.py myusername oauth:tokenstringfromtwitch You can get the IRC password from this oauth token generator here. If you are using another IRC network, you might want to change the code here already.

Let’s continue with the code. In the next part, we set up the IRC connecting and message handling

def parse_tags(event): """helper function to parse tags and twitch-specific user name field""" if hasattr(event,"tags_dict"): return event.tags_dict tags={} for tag in event.tags: tags[tag["key"]]=tag["value"] event.tags_dict=tags try: event.display_name=tags["display-name"] except: pass class AsyncIRC(object): """Wrapper for irc functionality, with chat command decorators""" def __init__(self, loop, prefix=">"): self.loop = loop or asyncio.get_event_loop() self.reactor=irc.client_aio.AioReactor(loop=self.loop) self.connection=self.reactor.server() self.commands=dict() self.prefix=prefix self.msgs=[] def dispatch_cmd(connection, event): """message handler for chat commands""" content=event.arguments[0] if content.startswith(self.prefix): sans_prefix=content[len(self.prefix):] try: cmd, *args = sans_prefix.split() except: cmd = sans_prefix args=[] cmd=cmd.lower() if cmd in self.commands: parse_tags(event) return self.commands[cmd](connection, event, *args) else: print('invalid command: ' + cmd) self.reactor.add_global_handler("pubmsg", dispatch_cmd, -1) def handle(self, event_type, *args, **kwargs): """decorator for callbacks to handle generic IRC events""" def decorator(fun): self.reactor.add_global_handler(event_type, fun, *args, **kwargs) return fun return decorator def broadcast(self, message): self.connection.privmsg(self.channel, message) def command(self, command): """decorator for callbacks to handle chat >commands""" def decorator(fun): self.commands[command]=fun return fun return decorator def connect(self, server, nickname, join_channel, password=None, port=6667): def on_connect(connection, event): connection.cap('REQ', ':twitch.tv/membership') connection.cap('REQ', ':twitch.tv/tags') connection.cap('REQ', ':twitch.tv/commands') connection.join(join_channel) if join_channel: self.channel=join_channel self.reactor.add_global_handler("welcome", on_connect) self.channel=join_channel coro=self.connection.connect(server=server, port=port, nickname=nickname, password=password) self.loop.run_until_complete(coro)

We create an IRC client now, and add callbacks for chat events:

loop = asyncio.get_event_loop() bot=AsyncIRC(loop) def run_once(loop): loop.call_soon(loop.stop) loop.run_forever() @bot.command("left") def command_left(connection, event): pygame.event.post( pygame.event.Event(pygame.KEYDOWN, key=pygame.K_LEFT, unicode=None, mod=None)) @bot.command("right") def command_right(connection, event): pygame.event.post( pygame.event.Event(pygame.KEYDOWN, key=pygame.K_RIGHT, unicode=None, mod=None)) @bot.command("up") def command_right(connection, event): pygame.event.post( pygame.event.Event(pygame.KEYDOWN, key=pygame.K_UP, unicode=None, mod=None)) @bot.command("down") def command_right(connection, event): pygame.event.post( pygame.event.Event(pygame.KEYDOWN, key=pygame.K_DOWN, unicode=None, mod=None)) @bot.command("newgame") def command_left(connection, event): pygame.event.post( pygame.event.Event(pygame.KEYUP, key=pygame.K_SPACE, unicode=None, mod=None)) @bot.handle("pubmsg") def on_pubmsg(connection, event): content=event.arguments[0] parse_tags(event) bot.msgs.append(f"{event.display_name}: {content}") print("#", bot.msgs)

The first five handlers translate chat commands into pygame events, while the last one takes every chat message and logs the logs the content into a list. This way we can display chat history in the game.

Now we can connect to the IRC like so:

bot.connect(server='irc.chat.twitch.tv', nickname=irc_user, password=irc_pw, join_channel="#"+irc_user)

and when we call run_once(loop), then the whole async IRC machinery will start looking for messages, handle them if there are any, and return.

Here comes the actual game part, based on wormy:

# Wormy (a Nibbles clone) # By Al Sweigart [email protected] # http://inventwithpython.com/pygame # Released under a "Simplified BSD" license # some IRC features added by Robert Pfeiffer import random, pygame, sys from pygame.locals import * FPS = 15 WINDOWWIDTH = 640 WINDOWHEIGHT = 480 CELLSIZE = 20 assert WINDOWWIDTH % CELLSIZE == 0, "Window width must be a multiple of cell size." assert WINDOWHEIGHT % CELLSIZE == 0, "Window height must be a multiple of cell size." CELLWIDTH = int(WINDOWWIDTH / CELLSIZE) CELLHEIGHT = int(WINDOWHEIGHT / CELLSIZE) # R G B WHITE = (255, 255, 255) BLACK = ( 0, 0, 0) RED = (255, 0, 0) GREEN = ( 0, 255, 0) DARKGREEN = ( 0, 155, 0) DARKGRAY = ( 40, 40, 40) BGCOLOR = BLACK UP = 'up' DOWN = 'down' LEFT = 'left' RIGHT = 'right' HEAD = 0 # syntactic sugar: index of the worm's head def main(): global FPSCLOCK, DISPLAYSURF, BASICFONT, CHATFONT pygame.init() FPSCLOCK = pygame.time.Clock() DISPLAYSURF = pygame.display.set_mode((WINDOWWIDTH, WINDOWHEIGHT)) BASICFONT = pygame.font.Font('freesansbold.ttf', 18) CHATFONT = pygame.font.Font('freesansbold.ttf', 8) pygame.display.set_caption('Wormy') showStartScreen() while True: runGame() showGameOverScreen()

Here, CHATFONT was added to the original wormy game to display chat in a different font.

In the game loop, we added chat messages when a new game is started, and we now call run_once(loop) before clock.tick(), like in the previous examples. To make the game playable with streaming lag (chat will usually be way ahead of the stream), the variable keydown_this_frame will ensure that the game state is only advanced when either a key has been pressed, or a keypress has been sent over a chat command.

def runGame(): bot.broadcast("new game") bot.broadcast(">UP >DOWN >LEFT >RIGHT to play") # Set a random start point. startx = random.randint(5, CELLWIDTH - 6) starty = random.randint(5, CELLHEIGHT - 6) wormCoords = [{'x': startx, 'y': starty}, {'x': startx - 1, 'y': starty}, {'x': startx - 2, 'y': starty}] direction = RIGHT # Start the apple in a random place. apple = getRandomLocation() while True: # main game loop keydown_this_frame=False for event in pygame.event.get(): # event handling loop if event.type == QUIT: terminate() elif event.type == KEYDOWN: keydown_this_frame=True if (event.key == K_LEFT or event.key == K_a) and direction != RIGHT: direction = LEFT elif (event.key == K_RIGHT or event.key == K_d) and direction != LEFT: direction = RIGHT elif (event.key == K_UP or event.key == K_w) and direction != DOWN: direction = UP elif (event.key == K_DOWN or event.key == K_s) and direction != UP: direction = DOWN elif event.key == K_ESCAPE: terminate() if keydown_this_frame: # BEGIN MOVEMENT CODE # check if the worm has hit itself or the edge if wormCoords[HEAD]['x'] == -1 or wormCoords[HEAD]['x'] == CELLWIDTH or wormCoords[HEAD]['y'] == -1 or wormCoords[HEAD]['y'] == CELLHEIGHT: return # game over for wormBody in wormCoords[1:]: if wormBody['x'] == wormCoords[HEAD]['x'] and wormBody['y'] == wormCoords[HEAD]['y']: return # game over # check if worm has eaten an apply if wormCoords[HEAD]['x'] == apple['x'] and wormCoords[HEAD]['y'] == apple['y']: # don't remove worm's tail segment apple = getRandomLocation() # set a new apple somewhere bot.broadcast("apple eaten") else: del wormCoords[-1] # remove worm's tail segment # move the worm by adding a segment in the direction it is moving if direction == UP: newHead = {'x': wormCoords[HEAD]['x'], 'y': wormCoords[HEAD]['y'] - 1} elif direction == DOWN: newHead = {'x': wormCoords[HEAD]['x'], 'y': wormCoords[HEAD]['y'] + 1} elif direction == LEFT: newHead = {'x': wormCoords[HEAD]['x'] - 1, 'y': wormCoords[HEAD]['y']} elif direction == RIGHT: newHead = {'x': wormCoords[HEAD]['x'] + 1, 'y': wormCoords[HEAD]['y']} wormCoords.insert(0, newHead) #END MOVEMENT CODE DISPLAYSURF.fill(BGCOLOR) drawGrid() drawWorm(wormCoords) drawApple(apple) drawScore(len(wormCoords) - 3) pygame.display.update() run_once(loop) FPSCLOCK.tick(FPS)

Logic to read IRC and run the event loop was added to the other game screens, and to the game termination function:

def terminate(): while len(asyncio.Task.all_tasks(loop)): run_once(loop) pygame.event.pump() loop.shutdown_asyncgens() loop.close() print("Thank you for playing!") pygame.quit() sys.exit() def getRandomLocation(): return {'x': random.randint(0, CELLWIDTH - 1), 'y': random.randint(0, CELLHEIGHT - 1)} def showGameOverScreen(): gameOverFont = pygame.font.Font('freesansbold.ttf', 150) gameSurf = gameOverFont.render('Game', True, WHITE) overSurf = gameOverFont.render('Over', True, WHITE) gameRect = gameSurf.get_rect() overRect = overSurf.get_rect() gameRect.midtop = (WINDOWWIDTH / 2, 10) overRect.midtop = (WINDOWWIDTH / 2, gameRect.height + 10 + 25) bot.broadcast("game over") DISPLAYSURF.blit(gameSurf, gameRect) DISPLAYSURF.blit(overSurf, overRect) drawPressKeyMsg() pygame.display.update() pygame.time.wait(500) checkForKeyPress() # clear out any key presses in the event queue while True: run_once(loop) if checkForKeyPress(): pygame.event.get() # clear event queue return

In drawScore, we also draw the last 3 lines of chat. in a small font. This way you know how big the lag in the stream is, and whose last chat command is currently shown on the screen.

def drawScore(score): scoreSurf = BASICFONT.render('Score: %s' % (score), True, WHITE) scoreRect = scoreSurf.get_rect() scoreRect.topleft = (WINDOWWIDTH - 120, 10) DISPLAYSURF.blit(scoreSurf, scoreRect) topleft=scoreRect.bottomleft for msg in bot.msgs[-3:]: IRCSurf = CHATFONT.render(msg, True, WHITE) IRCRect = IRCSurf.get_rect() IRCRect.topleft = topleft topleft= IRCRect.bottomleft DISPLAYSURF.blit(IRCSurf, IRCRect) def drawWorm(wormCoords): for coord in wormCoords: x = coord['x'] * CELLSIZE y = coord['y'] * CELLSIZE wormSegmentRect = pygame.Rect(x, y, CELLSIZE, CELLSIZE) pygame.draw.rect(DISPLAYSURF, DARKGREEN, wormSegmentRect) wormInnerSegmentRect = pygame.Rect(x + 4, y + 4, CELLSIZE - 8, CELLSIZE - 8) pygame.draw.rect(DISPLAYSURF, GREEN, wormInnerSegmentRect) def drawApple(coord): x = coord['x'] * CELLSIZE y = coord['y'] * CELLSIZE appleRect = pygame.Rect(x, y, CELLSIZE, CELLSIZE) pygame.draw.rect(DISPLAYSURF, RED, appleRect) def drawGrid(): for x in range(0, WINDOWWIDTH, CELLSIZE): # draw vertical lines pygame.draw.line(DISPLAYSURF, DARKGRAY, (x, 0), (x, WINDOWHEIGHT)) for y in range(0, WINDOWHEIGHT, CELLSIZE): # draw horizontal lines pygame.draw.line(DISPLAYSURF, DARKGRAY, (0, y), (WINDOWWIDTH, y)) if __name__ == '__main__': main()

This is the end of “twitch plays wormy”. You can copy all the segments into a python file together and run the game now. If you compare this code to the original wormy.py, you will see that it’s mostly the same, the changes are very limited - you can easily add the code to different games, as long as they aren’t quick-reaction-skill action games. If they are action games, you need to slow down the game to handle the lag. The code that reads chat only runs inside the event loop, so only when run_once(loop) is called, and interacts with the rest of the game by posting ordinary pygame events.

That’s one rather famous type of twitch integration out of the way. But another type is far more common:

Twitch Voting

Most of the time, chat doesn’t play the game - the streamer plays it, and chat comments on gameplay or talks to the streamer. Chat integration is not about sending button presses to the game, or about moment-to-moment gameplay, but about asking chatters to help make occasional, high-level decisions: Which hat should I wear? Should I go left or right? Which card should I add to this deck?

So here is a simple poll/voting pygame app. It’s not integrated into an existing game this time. Asking twitch chat to vote on decisions is something that needs to fit into the game UI and overall game design, and I did not want to retrofit a complex game for this tutorial.

We start with the same imports:

import pygame import asyncio import irc, irc.client_aio import sys import collections # we need collections this time too if len(sys.argv)!=3: print("please get a twitch auth token from https://twitchapps.com/tmi/") sys.exit(1) username=sys.argv[1] token=sys.argv[2] irc_pw=token.strip() irc_user=username.strip() assert(irc_pw.startswith("oauth:"))

This time, the AsyncIRC class is more complex. In addition to a handler for chat commands, there is also a handler for poll votes. In addition to a list of chat messages, there are also fields for counting votes, counting voters, voting options, and so on.

def parse_tags(event): if hasattr(event,"tags_dict"): return event.tags_dict tags={} for tag in event.tags: tags[tag["key"]]=tag["value"] event.tags_dict=tags try: event.display_name=tags["display-name"] except: pass class AsyncIRC(object): def __init__(self, loop, command_prefix=">", poll_prefix="#"): self.loop = loop or asyncio.get_event_loop() self.reactor=irc.client_aio.AioReactor(loop=self.loop) self.connection=self.reactor.server() self.commands=dict() self.command_prefix=command_prefix self.poll_prefix=poll_prefix self.msgs=[] self.has_voted=[] self.poll_votes=0 self.poll_count={} self.poll_options=[] self.poll_running=False def dispatch_cmd(connection, event): content=event.arguments[0] if content.startswith(self.command_prefix): sans_prefix=content[len(self.command_prefix):] try: cmd, *args = sans_prefix.split() except: cmd = sans_prefix args=[] cmd=cmd.lower() if cmd in self.commands: parse_tags(event) return self.commands[cmd](connection, event, *args) else: print('invalid command: ' + cmd) def count_poll(connection, event): content=event.arguments[0] if self.poll_running and content.startswith(self.poll_prefix): sans_prefix=content[len(self.poll_prefix):].strip().lower() if sans_prefix in self.poll_options: if not event.source in self.has_voted: self.poll_count[sans_prefix]+=1 self.poll_votes+=1 # self.has_voted.append(event.source) #commented out for testing #so you can vote twice self.reactor.add_global_handler("pubmsg", dispatch_cmd, -1) self.reactor.add_global_handler("pubmsg", count_poll, -1) def start_poll(self, title, options): self.poll_votes=0 self.has_voted=[] self.poll_count=collections.defaultdict(int) self.poll_options=[str(option).strip().lower() for option in options] self.poll_running=True self.poll_title=title message=f"Poll: {title} - " message+=", ".join([self.poll_prefix+option for option in options]) self.connection.privmsg(self.channel, message) self.connection.privmsg(self.channel, "voting starts NOW") def stop_poll(self): self.poll_running=False best_option=None for option in self.poll_options: if self.poll_count[option] > self.poll_count[best_option]: best_option=option nvotes=self.poll_count[best_option] percent=int(100*nvotes/self.poll_votes) message=f"poll {self.poll_title} ended - {best_option} won with {nvotes} votes ({percent}%)" self.connection.privmsg(self.channel, message) def handle(self, event_type, *args, **kwargs): def decorator(fun): self.reactor.add_global_handler(event_type, fun, *args, **kwargs) return fun return decorator def broadcast(self, message): self.connection.privmsg(self.channel, message) def command(self, command): def decorator(fun): self.commands[command]=fun return fun return decorator def connect(self, server, nickname, join_channel, password=None, port=6667): def on_connect(connection, event): connection.cap('REQ', ':twitch.tv/membership') connection.cap('REQ', ':twitch.tv/tags') connection.cap('REQ', ':twitch.tv/commands') connection.join(join_channel) if join_channel: self.channel=join_channel self.reactor.add_global_handler("welcome", on_connect) self.channel=join_channel coro=self.connection.connect(server=server, port=port, nickname=nickname, password=password) self.loop.run_until_complete(coro) loop = asyncio.get_event_loop() bot=AsyncIRC(loop)

Again, we log all public messages.

@bot.handle("pubmsg") def on_pubmsg(connection, event): content=event.arguments[0] parse_tags(event) bot.msgs.append(f"{event.display_name}: {content}") #print("#", bot.msgs) bot.connect(server='irc.chat.twitch.tv', nickname=irc_user, password=irc_pw, join_channel="#"+irc_user) def run_once(loop): loop.call_soon(loop.stop) loop.run_forever()

The actual UI for this twitch polling app is rather short. When the user presses Return, the poll starts, if it isn’t already running. The duration of the poll is set by the variable poll_ticks in the game loop, not in the event loop. While the poll hasn’t been stopped, it keeps on counting on the event loop.

Since the vote counting happens only inside run_once(loop), there are no race conditions between polling and game loop, and we don’t need to worry about race conditions for votes that arrive just as the poll ends, or votes that arrive while the UI is being drawn.

poll_ticks=-1 my_options=["A", "B", "C"] pygame.init() screen_size=640,480 screen=pygame.display.set_mode(screen_size) clock=pygame.time.Clock() running=True flying_frames=0 best=0 color=(50,50,50) font=pygame.font.SysFont("Helvetica Neue,Helvetica,Ubuntu Sans,Bitstream Vera Sans,DejaVu Sans,Latin Modern Sans,Liberation Sans,Nimbus Sans L,Noto Sans,Calibri,Futura,Beteckna,Arial", 16) while running: clock.tick(30) events=pygame.event.get() for e in events: if e.type==pygame.QUIT: running=False if e.type==pygame.KEYDOWN and e.key==pygame.K_RETURN: if poll_ticks<=0: bot.start_poll("Test Vote", my_options) poll_ticks=30*45 #FPS*45 seconds screen.fill((255,255,255)) if poll_ticks>=0: if poll_ticks>0: poll_seconds=poll_ticks//30 time_text=font.render(f"{poll_seconds}s remaining", True, (0,0,0)) else: time_text=font.render("results:", True, (0,0,0)) screen.blit(time_text,(10,30)) for i, option in enumerate(bot.poll_options): nvotes=bot.poll_count[option] if bot.poll_votes>0: percent=int(100*nvotes/bot.poll_votes) else: percent=0 option_text=font.render( f"{bot.poll_prefix}{option}: {nvotes} ({percent}%)", True, (0,0,0)) y_pos=100+i*30 screen.blit(option_text,(10,y_pos)) if bot.poll_votes < 20: bar_length=nvotes*20 else: bar_length=percent*5 pygame.draw.rect(screen, color, pygame.Rect(100, y_pos, bar_length,25)) if poll_ticks>0: poll_ticks-=1 if poll_ticks==0: bot.stop_poll() fps=clock.get_fps() pygame.display.update() run_once(loop) while len(asyncio.Task.all_tasks(loop)): run_once(loop) loop.shutdown_asyncgens() loop.close() print("Thank you for playing!")

Now you can add voting to your own pygame games. Maybe you shouldn’t call the options “1”, “2”, and “3”, but give them descriptive names depending on the current situation.

#gamedev #python #modern PyGame in 2018 #asyncio

1 note · View note