Python Tutorial: Broadcasting Battles with Hive On and Offchain

Let's explore another aspect of splinterlands in this tutorial series: the battle system. This time we're looking at how to submit a team with your hive account on and off chain.

For simplicities sake I'm just going to look at ranked battles. Tournaments are a whole different story. I won't be talking about team selection or making sure that a team is "valid" for the given ruleset. Just the mechanics needed to find a match and submit a team.

Finding a match

First we need to tell Splinterlands that we are looking for a match, to make things easier we're going to look at the on chain solution:

from beem import Hive


def broadcast_find_match(hive: Hive, user: str, match_type: str):
    request = {"match_type": match_type}

    trx: dict = hive.custom_json("sm_find_match", json_data=request,
                                 required_posting_auths=[user])
    return trx["trx_id"]

That part looks pretty straight forward. We create a json with the match type and broadcast a "sm_find_match" transaction and return the transaction id. Now, if you want to use it it looks like following:

user = "your_username"
hive = Hive(keys=["PRIVATE_POSTING_KEY"])
transaction_id = broadcast_find_match(hive, user, "Ranked")

Since this transaction is broadcast to the hive blockchain you need RC for this operation. If you have enough RC and execute this code splinterlands will immediately start looking for a match. Try it out if you want to, but expect to play immediately. Just reload the website a couple times.

Now we need to wait until we have a match. Since we don't have to website to tell us when a match starts we need to look that up ourselves:

There are two main ways. Either logging into the website via login request and checking the outstanding match. Or by calling the battle status endpoint with your transaction id. Since the latter doesn't require any signature whatsoever, we just use that one:

import requests

API2 = "https://api2.splinterlands.com"

def get_battle_status(battle_id: str):
    url: str = API2 + "/battle/status?id=" + battle_id
    return requests.get(url).json()

Done. Not much to say about that. A simple get request to /battle/status with the transaction id appended.

Now lets see whether we have an opponent:

import time

resp = get_battle_status(transaction_id)

while type(resp) == str or type(resp) == dict and not resp["opponent_player"]:
    print("Waiting for match")
    resp = get_battle_status(transaction_id)
    time.sleep(3)

Now we are calling the endpoint initially once. Afterwards we go into a while loop and check the battle status every 3 seconds until we have found an opponent. The check for string, is there to prevent error messages from breaking the loop. E.g. if the server hasn't found our transaction id yet.
Sometimes the loop doesn't even start if we find an opponent immediately.

After we've found an opponent you'll get the following response from the battle status endpoint:

{
   "id":"sl_4c95b6ea2787050fbecbb523f61220cf",
   "created_block_num":60782081,
   "expiration_block_num":60782141,
   "player":"aicu",
   "team_hash":"None",
   "match_type":"Ranked",
   "mana_cap":19,
   "opponent":"sl_019ff1649587177db68ba1c37984766f",
   "match_block_num":0,
   "status":1,
   "reveal_tx":"None",
   "reveal_block_id":"None",
   "team":"None",
   "summoner_level":"None",
   "ruleset":"Standard",
   "inactive":"",
   "opponent_player":"distortedtomish",
   "opponent_team_hash":"None",
   "submit_expiration_block_num":0,
   "settings":"{\"rating_level\":1}",
   "app":"None",
   "created_date":"2022-01-09T16:50:58.037Z",
   "expiration_date":"2022-01-09T16:53:58.037Z",
   "match_date":"2022-01-09T16:51:01.573Z",
   "submit_expiration_date":"2022-01-09T16:54:19.573Z",
   "recent_opponents":"[\"miudama\",\"eroserys\",\"wowlover3\",\"slimrubi\",\"recruit_776028\",\"asihcinta\",\"fever025\",\"lxfpx27060435\",\"chessbloke2\",\"jealousbianca\"]"
}

As you can see, we get all the information we need to start the team selection process. Ruleset (separated by |), mana_cap, inactive (splinters) etc.

How you do that is totally up to you :). Might expand on simple strategies in the future. I've had decent success in the past with Monte Carlo Tree Search. Statistical approaches work pretty good as well. Just try things out. Games with uncertainty (doge, miss etc) have always been challenging for any type of algorithm.

Submitting and revealing teams

Now we know our opponent and lets assume we've computed a team:

team = ["C7-440-HN96LKT08W", "C4-157-0W00ZMKJDS"]

This is just a dummy team from my account. The team is just a list of uids. For this to work you need to a) own the card b) have them not delegated to someone c) not listed on the marketplace.

Now let's prepare the team submit and the following reveal:

def broadcast_submit_team(hive: Hive, user: str, trx_id: str, team: List[str], secret: str):
    hash = generate_team_hash(team[0], team[1:], secret)

    request = {"trx_id": trx_id, "team_hash": hash}
    trx: dict = hive.custom_json("sm_submit_team", json_data=request,
                                 required_posting_auths=[user])
    trx_id = trx["trx_id"]

    return trx_id, hash

That's the simple on chain solution for broadcasting a team: notice that we are not submitting an actual team, but compute hash of our team. Let's look at the required function:

import hashlib

def generate_team_hash(summoner, monsters, secret):
    m = hashlib.md5()
    m.update((summoner + ',' + ','.join(monsters) + ',' + secret).encode("utf-8"))
    team_hash = m.hexdigest()
    return team_hash

The team hash is MD5 hash over the list of uids comma separated plus the secret also comma separated.

Now we just need our secret, you can pass anything in as the secret. But for securities sake we're going to use a "real" secret.

from secrets import choice

def generate_secret(length=10):
    return ''.join(choice(string.ascii_letters + string.digits) for i in range(length))

Now lets plug everything together:

secret = generate_secret()
trx_id, team_hash = broadcast_submit_team(hive, user, transaction_id,team,secret)

Now we have the team hash and the transaction id of the submit team. We don't need the trx_id of that transaction right now, but for debugging purposes it might help you.

Now we just need to reveal our team, for that we're going to use the following function:

def broadcast_reveal_team(hive: Hive, user: str, team: List[str], secret: str, trx_id, hash):
    request = {"trx_id": trx_id, "team_hash": hash, "summoner": team[0], "monsters": team[1:],
               "secret": secret}
    trx: dict = hive.custom_json("sm_team_reveal", json_data=request,
                                 required_posting_auths=[user])
    trx_id = trx["trx_id"]

    return trx_id

In use the function looks like that:

broadcast_reveal_team(hive, user, team, secret, transaction_id, team_hash)

Now we're done. We broadcasted a a complete match on the hive blockchain and just need to wait on the match ending. If you want to know the result:

def get_battle_result(battle_id: str):
    url: str = API2 + "/battle/result?id=" + battle_id
    return requests.get(url).json()

Call that function. Its going to tell you everything that happened in the match, including monster attacks and the winner etc pp. For that reason I'm not posting the json for that reason. Since it gets very convoluted.

Just wait until the value of "winner" is set and you'll know who won.

Broadcasting Transactions Off Chain

Now we know how to find and submit a splinterlands match using the hive blockchain.
But what if we don't have and Hive Power, and in turn only limited resource credits ?

For that reason, we can also make transaction off chain, by sending the transaction directly to the splinterlands server.

For that we must not broadcast the transaction to the blockchain. Only sign it. Let's write a helper functions and adapt the find match function:

def copy_hive(hive: Hive, user: str):
    private_key = hive.wallet.getPostingKeyForAccount(user)
    return Hive(nobroadcast=True, keys=[private_key])

This step is optional, but preserves your original hive instance. If you only want to do off chain transactions you can just set nobroadcast=True in your original hive instance and ignore this function.

Finding Matches Off Chain

The following code is easy to adjust if you aren't using the helper function above.

def broadcast_find_match(hive: Hive, user: str, match_type: str, on_chain: bool):
    hive = hive if on_chain else copy_hive(hive, user)

    trx_id = None
    trx: dict = hive.custom_json("sm_find_match", json_data={"match_type": match_type},
                                 required_posting_auths=[user])
    if not on_chain:
        trx = post_battle_transaction(trx)
        if trx["success"]:
            trx_id = trx["id"]
    else:
        trx_id = trx["trx_id"]
        
    return trx_id

Just by looking at this new find match function you should notice a couple differences. I introduced a boolean flag and used the copy function in the first line to adapt the hive instance for not broadcasting.

If you start looking below the custom json operation you notice that it returns now a dictionary. This dictionary is our hive transaction. Signed and not broadcasted to the hive blockchain. It essentially never happened.

Now let's make it happen: if we broadcast off chain we need to post it to the splinterlands server. And thats where "post_battle_transaction" comes into play:

import json

BASE_BATTLE = "https://battle.splinterlands.com"

def post_battle_transaction(trx: dict):
    url: str = BASE_BATTLE + "/battle/battle_tx"
    trx: str = json.dumps(trx)
    return requests.post(url, data={"signed_tx": trx}).json()

It's very concise but does a few very important things.

1. We're not using api2, but the battle api.
2. The transaction we created needs to be converted into a string.
3. We are sending it in using a post request.

Let's look at each point in detail. Point 1 is pretty much self explanatory.

1. Why convert it to a string and not pass it to the server as is ? Its because splinterland expects the mime type application/x-www-form-urlencoded. This is a standardised format where the request gets separated into key/value pairs in one long query string. It could look something like this:

a=12345&b=ABCD

In our case it's just one parameter and it's called "signed_tx".
This signed transaction requires our signed transaction as json string input. From what I understand the requests endpoint automatically creates a url encoded form if you pass a flat structure to it. (Correct me on that, might very likely be wrong).

Submitting and Revealing Transactions off chain

For submit and reveal we just need to do the same thing:

def broadcast_submit_team(hive: Hive, user: str, trx_id: str, team: List[str], secret: str, on_chain: bool):
    hive = hive if on_chain else copy_hive(hive, user)

    hash = generate_team_hash(team[0], team[1:], secret)

    request = {"trx_id": trx_id, "team_hash": hash}
    trx: dict = hive.custom_json("sm_submit_team", json_data=request,
                                 required_posting_auths=[user])
    trx_id = None
    if not on_chain:
        trx = post_battle_transaction(trx)
        if trx["success"]:
            trx_id = trx["id"]
    else:
        trx_id = trx["trx_id"]

    return trx_id, hash


def broadcast_reveal_team(hive: Hive, user: str, team: List[str], secret: str, trx_id, hash, on_chain: bool):
    hive = hive if on_chain else copy_hive(hive, user)

    request = {"trx_id": trx_id, "team_hash": hash, "summoner": team[0], "monsters": team[1:],
               "secret": secret}
    trx: dict = hive.custom_json("sm_team_reveal", json_data=request,
                                 required_posting_auths=[user])

    trx_id = None
    if not on_chain:
        trx = post_battle_transaction(trx)
        if trx["success"]:
            trx_id = trx["id"]
    else:
        trx_id = trx["trx_id"]

    return trx_id

Now we're done. Let's see how this code looks in action:

user = "username"
hive = Hive(keys=["PRIVATE_POSTING_KEY"])
transaction_id = broadcast_find_match(hive, user, "Ranked",False)

resp = get_battle_status(transaction_id)

while type(resp) == str or type(resp) == dict and not resp["opponent_player"]:
    print("Waiting for match")
    resp = get_battle_status(transaction_id)
    time.sleep(3)

team = ["C7-440-HN96LKT08W", "C4-157-0W00ZMKJDS"]

secret = generate_secret()
trx_id, team_hash = broadcast_submit_team(hive, user, transaction_id, team,False, secret)

broadcast_reveal_team(hive, user, team, secret,False, transaction_id, team_hash)

Looks pretty similar and works exactly the same.
If you take a closer look at the transaction_id you might notice that it is now prefixed with a "sl_" signifying that it wasn't broadcast to the hive blockchain.

And now we're done. You just learned how to find, submit and reveal transactions on and off chain.

Giveaway

This time I'm giving away just two Mother Khalas lvl 1 . Just comment below with your username. Drawing will be after seven days:

C4-156-25JKU14B28
C4-156-WPJ9GDJ3Z4

Complete Code

import hashlib
import json
import string
import time
from secrets import choice
from typing import List

import requests
from beem import Hive

API2 = "https://api2.splinterlands.com"
BASE_BATTLE = "https://battle.splinterlands.com"


def broadcast_find_match(hive: Hive, user: str, match_type: str, on_chain: bool):
    hive = hive if on_chain else copy_hive(hive, user)

    trx_id = None
    trx: dict = hive.custom_json("sm_find_match", json_data={"match_type": match_type},
                                 required_posting_auths=[user])
    if not on_chain:
        trx = post_battle_transaction(trx)
        if trx["success"]:
            trx_id = trx["id"]
    else:
        trx_id = trx["trx_id"]

    return trx_id


def broadcast_submit_team(hive: Hive, user: str, trx_id: str, team: List[str], secret: str,on_chain: bool):
    hive = hive if on_chain else copy_hive(hive, user)

    hash = generate_team_hash(team[0], team[1:], secret)

    request = {"trx_id": trx_id, "team_hash": hash}
    trx: dict = hive.custom_json("sm_submit_team", json_data=request,
                                 required_posting_auths=[user])
    trx_id = None
    if not on_chain:
        trx = post_battle_transaction(trx)
        if trx["success"]:
            trx_id = trx["id"]
    else:
        trx_id = trx["trx_id"]

    return trx_id, hash


def broadcast_reveal_team(hive: Hive, user: str, team: List[str], secret: str, trx_id, hash, on_chain: bool):
    hive = hive if on_chain else copy_hive(hive, user)

    request = {"trx_id": trx_id, "team_hash": hash, "summoner": team[0], "monsters": team[1:],
               "secret": secret}
    trx: dict = hive.custom_json("sm_team_reveal", json_data=request,
                                 required_posting_auths=[user])

    trx_id = None
    if not on_chain:
        trx = post_battle_transaction(trx)
        if trx["success"]:
            trx_id = trx["id"]
    else:
        trx_id = trx["trx_id"]

    return trx_id


def generate_secret(length=10):
    return ''.join(choice(string.ascii_letters + string.digits) for i in range(length))


def generate_team_hash(summoner, monsters, secret):
    m = hashlib.md5()
    m.update((summoner + ',' + ','.join(monsters) + ',' + secret).encode("utf-8"))
    team_hash = m.hexdigest()
    return team_hash


def get_battle_status(battle_id: str):
    url: str = API2 + "/battle/status?id=" + battle_id
    return requests.get(url).json()


def get_battle_result(battle_id: str):
    url: str = API2 + "/battle/result?id=" + battle_id
    return requests.get(url).json()


def copy_hive(hive: Hive, user: str):
    private_key = hive.wallet.getPostingKeyForAccount(user)
    return Hive(nobroadcast=True, keys=[private_key])


def post_battle_transaction(trx: dict):
    url: str = BASE_BATTLE + "/battle/battle_tx"
    trx: str = json.dumps(trx)
    return requests.post(url, data={"signed_tx": trx}).json()


user = "username"
hive = Hive(keys=["PRIVATE_POSTING_KEY"])
transaction_id = broadcast_find_match(hive, user, "Ranked", False)
print(transaction_id)
resp = get_battle_status(transaction_id)

while type(resp) == str or type(resp) == dict and not resp["opponent_player"]:
    print("Waiting for match")
    resp = get_battle_status(transaction_id)
    time.sleep(3)

team = ["Summoner UID", "MONSTER_1_UID","MONSTER_2_UID","ETC"]

secret = generate_secret()
trx_id, team_hash = broadcast_submit_team(hive, user, transaction_id, team, secret, False)

broadcast_reveal_team(hive, user, team, secret, transaction_id, team_hash,False)