Blog

OTPSpectate on Twitch.tv

2017-08-19

The twitch community has become a pillar of gaming culture. One of its most popular contributions was an automated stream called Twitch Plays Pokémon, in which a playthrough of Pokémon Red was controlled by twitch chat via a python script. Other notable automated streams include Salty Bet, where viewers can place virtual bets on the winner of fighting games. League of Legends has its own SaltyTeemo stream, where viewers can place bets on the winner of some of the lowest ranked matchmade games.

I thought it might be an interesting project to create a similar stream which spectated highly ranked League players who were considered one-trick ponies. In other words, these are highly skilled League of Legends players that climbed the ranked ladder by playing (mostly) a single champion in their games.

Using my spare time over a few weeks I built a twitch.tv stream automator using python 3, broadcasting with OBS Studio. The project was recently approved by Riot Games for a production API key. It currently broadcasts on the twitch channel OTPSpectate on an old laptop running Windows 7 (the PC I had played League on for the past 4 years).

This was a fun project to work on and I think it’s been working fairly well so far. In this post I’ll describe the core components of the stream with a few code snippets.

• Riot API
• Twitch Chat
• OBS Studio
• League of Legends Client
• Stream execution

Riot API

Using the requests library, I wrote a simple wrapper around the Riot API for a few endpoints that I would need to run the stream. The wrapper also employs a rate limiter class, which uses a deque object (double-ended queue) to track API call timestamps, and self-imposes wait times based on the rate limits allotted to the API key.

Here are the endpoints needed for the stream.

API_URL = 'https://{region}.api.riotgames.com'

endpoints = {
    'challenger': API_URL + '/lol/league/v3/challengerleagues/by-queue/{queue}',
    'master': API_URL + '/lol/league/v3/masterleagues/by-queue/{queue}',
    'spectator': API_URL + '/lol/spectator/v3/active-games/by-summoner/{summoner_id}',
    'summoner_name': API_URL + '/lol/summoner/v3/summoners/by-name/{summoner_name}',
    'summoner': API_URL + '/lol/summoner/v3/summoners/{summoner_id}',
    'versions': API_URL + '/lol/static-data/v3/versions',
    'champions': API_URL + '/lol/static-data/v3/champions',
    'matchlist': API_URL + '/lol/match/v3/matchlists/by-account/{account_id}',
    'league_position': API_URL + '/lol/league/v3/positions/by-summoner/{summoner_id}',
}

The challenger and master endpoints are for retrieving the list of top players. The summoner, summoner_name, matchlist, and league_position endpoints are for looking up individual information on a player. The champions endpoint returns a map of champion id to champion name. The versions endpoint is for tracking client version in the case of an upgrade. Finally, the spectator endpoint is for retrieving live game information for each player.

Once per day, the code will retrieve the list of challenger and master tier players from every region. For each player it will then retrieve their matchlist of ranked games from the current season. If the player has played more than 50% of those games on a single champion, they are considered a one-trick pony (OTP) and are added to the pool of OTP players for the stream.

When the program first launches, the code hits the spectator endpoint for each OTP player to find games in progress. The active games are written to a file that is referenced by the streaming software which presents the information to the players via stream. Players can then vote on which game they want to see next, and the spectator client launches the winner. This process repeats once the game is finished.

Twitch Chat

The Twitch chat service is essentially an IRC server. I used python’s built-in socket and ssl packages to write an IRC class which handles the lower level functionality with chat, namely things like connecting and messaging I/O. The twitch servers also PING the connection every few minutes, so there is also a pong() method which PONGs back the same message to maintain the connection.

On top of the IRC class is the OTPBot class which handles some of the higher level chat functionality, like managing the game selection polls between games, logging results, and handling custom chat commands (on my TODO list). For example, the method which handles chat voting between games:

import re

from otp.twitch.irc import IRC


class OTPBot(IRC):
    """The OTP Twitch chat bot."""

    # ... __init__ and other methods

    def _poll_response(self, user, message):
        """Handle the response if a poll is active."""
        if user in self.poll_voters:
            return
        if message.startswith('!vote '):
            remessage = re.match(r'^!vote ([0-9]+)$', message)
            try:
                vote = int(remessage.group(1))
            except:
                return
            print(vote)
            if vote in self.poll_votes:
                self.poll_votes[vote] += 1
                self.poll_voters.append(user)

OBS Studio

The OTP project contains an apps module for interacting with other applications.

Within the apps module is the OBS class, which handles broadcast controls for the stream by interacting with the OBS Studio application. Aside from initially launching the app and activating the stream, the class handles the changing of scenes between spectating to polling. This is done by assigning hotkeys to specific actions in OBS Studio, and then using the pywinauto package to send keystrokes to the application once it’s been launched. The package makes it very easy to interact with application windows. Here is a snippet of the change_to_scene method from the OBS class.

from pywinauto.keyboard import SendKeys
from pywinauto.findwindows import WindowNotFoundError
import pywinauto as windows

from otp.config import obs_config, obs_hotkeys


class OBS(object):
    """The OBS class.

    A class for interfacing with the OBS Studio application. Launches OBS
    and operates the application.
    """

    # ... __init__ and other methods

    def change_to_scene(self, scene_name):
        """Change the current scene to a new one."""
        self.app.top_window().set_focus()
        SendKeys(obs_hotkeys['scene_' + scene_name])

Here is a sample of the between games scene.

League of Legends Client

The other class of the apps module is LeagueOfLegends, which handles the spectator client. After assigning hotkeys in the client for toggling time controls and scoreboard, we can send keystrokes to the client using pywinauto to get a proper spectator view on the stream. By default the scoreboard is off and the time controls are on in spectator mode; both need to be changed using keystrokes.

To actually launch the spectator client I use the subprocess python built-in module, using the command line configuration outlined in the developer guide. Here is the spectator launch function, which constructs the launch command from configuration settings and then launches the client.

from otp.config import spectator_config

# ...

def spectate_game(region, encryption_key, match_id):
    """Launch the spectator client to spectate the match."""
    # Get latest spectator client version (highest version number)
    spectator_version_dir = sorted(
        [d for d in os.listdir(spectator_config['path'])
         if os.path.isdir(os.path.join(spectator_config['path'], d))])[-1]

    # Goto client dir
    launch_directory = spectator_config['path'] + '\\' + \
        spectator_version_dir + '\\deploy\\'
    cwd = os.getcwd()
    os.chdir(launch_directory)

    # Launch the spectator client
    spectate_match = spectator_config['spectate'].format(
        domain=spectator_grid[region]['domain'],
        port=spectator_grid[region]['port'],
        encryption_key=encryption_key,
        match_id=match_id,
        platform_id=spectator_grid[region]['platform']
    )
    launch_cli = [
        spectator_config['client'],
        spectator_config['maestro1'],
        spectator_config['maestro2'],
        spectator_config['third'],
        spectate_match
    ]
    subprocess.Popen(launch_cli)

    # Go back to original dir
    os.chdir(cwd)

Stream execution

The core loop of the stream is handled by a script called stream.py. Here is a rough outline of its tasks.

1. Launch OBS
2. Search for active games
3. Poll chat
4. Launch the spectator client
5. Kill the client after game completes
6. GOTO 2

If no games are found, the script stops the stream, waits a while and then searches again. If only one game is found it skips the poll.

Throughout this loop the chat bot is keeping up with chat commands at regular intervals, and PONGing back the server as required.

Since the Riot API rate limits are region-specific, I used the built-in threading package to use multi-threading to perform the active game search. Each region gets its own thread, so that any waiting initiated by the rate limiter or by the API doesn’t block API calls from other regions. This also ensures that the search is completed within the 3 minute spectator delay between the end of a game and end of a spectated game. Here is the code portion that executes the game search:

import json
from threading import Thread

from otp.riot.api import Riot


# ...

# Globally track the games via threading
active_games = []


class RegionThread(Thread):
    """Region specific thread to find active games."""

    def __init__(self, region, otps):
        """Instantiate the thread."""
        Thread.__init__(self)
        self.riot = Riot(riot_config['api_key'])
        self.region = region
        self.otps = otps
        print(region, len(otps))

    def run(self):
        """Find the active games."""
        for otp in self.otps:
            game = self.riot.get_player_spectator(
                otp['summoner_id'], self.region)
            # Ensure active game is ranked
            # and player is playing their otp
            # and game is less than 20 minutes old.
            # If satisfied, get their current ranked info for the overlay
            if (game is not None
                    and game['queue'] == 420
                    and otp['champion_id'] == game['champion_id']
                    and time.time() - game['match_start'] / 1000 < 20 * 60):
                league_info = self.riot.get_player_league_position(
                    otp['summoner_id'], self.region)
                game['tier'] = league_info['tier']
                game['rank'] = league_info['rank']
                game['points'] = league_info['points']
                active_games.append(game)
                print(game)


def search_games():
    """Find active games."""
    global active_games
    active_games = []

    otps = json.load(open('otp/data/otps.json', 'r'))

    threads = []
    for region in otps:
        threads.append(RegionThread(region, otps[region]))
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

# ...

In between operations the script writes to a few files that OBS reads to display overlay information on the stream. There are also checks for refreshing the OTP pool every 24 hours, and for checking the client version to automatically perform an upgrade. For these operations the script will turn off the stream as they take more than a few minutes.

The main loop is also wrapped in try-except logic, so that if anything breaks it stops the stream and sends me a notification.

Potential features

Some features I would like to add.

• Live vote counts on the poll screen.
• Twitch links to OTPs that stream their own gameplay.
• Camera locking the correct champion. (Can’t do this with the API since players are re-arranged according to position)

Check out the stream on twitch! twitch.tv/otpspectate

_OTPSpectate isn’t endorsed by Riot Games and doesn’t reflect the views or opinions of Riot Games or anyone officially involved in producing or managing League of Legends. League of Legends and Riot Games are trademarks or registered trademarks of Riot Games, Inc. League of Legends © Riot Games, Inc._

Further reading

OTPSpectate

• twitch.tv/otpspectate

Riot Games

• Developer Console
• API Documentation
• Spectator Grid

Twitch.tv

• Twitch chat and IRC
• SaltyTeemo
• SaltyBet
• TwitchPlaysPokémon

Python

• requests (Riot API I/O)
• socket (Twitch chat I/O)
• ssl (Twitch chat I/O)
• pywinauto (League of Legends and OBS Studio controls)
• subprocess (Launching the spectator)
• deque objects (API rate limit tracking)

OBS Studio

• OBS

Back to the posts.