Max/DragonMinded_bemaniutils
1
0
Fork 0
mirror of https://github.com/DragonMinded/bemaniutils.git synced 2026-10-12 17:28:13 +03:00
Code Issues Packages Projects Releases Wiki Activity
Files
cff4da2dfa4d359cafdbbbdfab2e990c2a3a3fab
DragonMinded_bemaniutils/bemani/client/popn/fantasia.py
T

477 lines
18 KiB
Python
Raw Blame History

import random
import time
from typing import Optional, Dict, List, Tuple, Any
from bemani.client.base import BaseClient
from bemani.protocol import Node
class PopnMusicFantasiaClient(BaseClient):
NAME = "ＴＥＳＴ"
def verify_game_active(self) -> None:
call = self.call_node()
# Construct node
game = Node.void("game")
call.add_child(game)
game.set_attribute("method", "active")
# Add what Pop'n 20 would add after full unlock
game.set_attribute("method", "active")
game.add_child(Node.s8("event", 0))
game.add_child(Node.s8("card_use", 0))
game.add_child(Node.string("name", ""))
game.add_child(Node.string("location_id", "JP-1"))
game.add_child(Node.string("shop_name_facility", "."))
game.add_child(Node.s8("pref", 50))
game.add_child(Node.string("shop_addr", "127.0.0.1 10000"))
game.add_child(Node.string("shop_name", ""))
game.add_child(Node.string("eacoin_price", "200,260,200,200"))
game.add_child(Node.s8("eacoin_available", 1))
game.add_child(Node.s8("dipswitch", 0))
game.add_child(Node.u8("max_stage", 1))
game.add_child(Node.u8("difficult", 4))
game.add_child(Node.s8("free_play", 1))
game.add_child(Node.s8("event_mode", 0))
game.add_child(Node.s8("popn_card", 0))
game.add_child(Node.s16("close_time", -1))
game.add_child(Node.s32("game_phase", 2))
game.add_child(Node.s32("net_phase", 0))
game.add_child(Node.s32("event_phase", 0))
game.add_child(Node.s32("card_phase", 3))
game.add_child(Node.s32("ir_phase", 0))
game.add_child(Node.u8("coin_to_credit", 1))
game.add_child(Node.u8("credit_to_start", 2))
game.add_child(Node.s32("sound_attract", 100))
game.add_child(Node.s8("bookkeeping", 0))
game.add_child(Node.s8("set_clock", 0))
game.add_child(Node.s32("local_clock_sec", 80011))
game.add_child(Node.s32("revision_sec", 0))
game.add_child(Node.string("crash_log", ""))
# Swap with server
resp = self.exchange("pnm20/game", call)
# Verify that response is correct
self.assert_path(resp, "response/game/@status")
def verify_game_get(self) -> None:
call = self.call_node()
# Construct node
game = Node.void("game")
call.add_child(game)
game.set_attribute("location_id", "JP-1")
game.set_attribute("method", "get")
game.add_child(Node.s8("card_enable", 0))
game.add_child(Node.s8("card_soldout", 1))
# Swap with server
resp = self.exchange("pnm20/game", call)
# Verify that response is correct
self.assert_path(resp, "response/game")
for name in [
"game_phase",
"ir_phase",
"event_phase",
"netvs_phase",
"illust_phase",
"psp_phase",
"other_phase",
"jubeat_phase",
"public_phase",
"kac_phase",
"local_matching_enable",
"n_matching_sec",
"l_matching_sec",
"is_check_cpu",
"week_no",
]:
node = resp.child("game").child(name)
if node is None:
raise Exception(f"Missing node '{name}' in response!")
if node.data_type != "s32":
raise Exception(f"Node '{name}' has wrong data type!")
sel_ranking = resp.child("game").child("sel_ranking")
up_ranking = resp.child("game").child("up_ranking")
ng_illust = resp.child("game").child("ng_illust")
for name, node, dtype, length in [
("sel_ranking", sel_ranking, "s16", 10),
("up_ranking", up_ranking, "s16", 10),
("ng_illust", ng_illust, "s32", 64),
]:
if node is None:
raise Exception(f"Missing node '{name}' in response!")
if node.data_type != dtype:
raise Exception(f"Node '{name}' has wrong data type!")
if not node.is_array:
raise Exception(f"Node '{name}' is not array!")
if len(node.value) != length:
raise Exception(f"Node '{name}' is wrong array length!")
def verify_playerdata_get(
self, ref_id: str, msg_type: str
) -> Optional[Dict[str, Any]]:
call = self.call_node()
# Construct node
playerdata = Node.void("playerdata")
call.add_child(playerdata)
playerdata.set_attribute("method", "get")
if msg_type == "new":
playerdata.set_attribute(
"model", self.config["old_profile_model"].split(":")[0]
)
playerdata.add_child(Node.string("ref_id", ref_id))
playerdata.add_child(Node.string("shop_name", ""))
playerdata.add_child(Node.s8("pref", 50))
playerdata.add_child(Node.s32("navigate", 1))
# Swap with server
resp = self.exchange("pnm20/playerdata", call)
if msg_type == "new":
# Verify that response is correct
self.assert_path(resp, "response/playerdata/@status")
status = int(resp.child("playerdata").attribute("status"))
if status != 109:
raise Exception(
f"Reference ID '{ref_id}' returned invalid status '{status}'"
)
# No score data
return None
elif msg_type == "query":
# Verify that the response is correct
self.assert_path(resp, "response/playerdata/base/g_pm_id")
self.assert_path(resp, "response/playerdata/base/name")
self.assert_path(resp, "response/playerdata/base/my_best")
self.assert_path(resp, "response/playerdata/base/latest_music")
self.assert_path(resp, "response/playerdata/base/clear_medal")
self.assert_path(resp, "response/playerdata/base/clear_medal_sub")
self.assert_path(resp, "response/playerdata/player_card")
self.assert_path(resp, "response/playerdata/player_card_ex")
self.assert_path(resp, "response/playerdata/hiscore")
self.assert_path(resp, "response/playerdata/netvs")
self.assert_path(resp, "response/playerdata/sp_data")
self.assert_path(resp, "response/playerdata/reflec_data")
self.assert_path(resp, "response/playerdata/navigate")
name = resp.child("playerdata").child("base").child("name").value
if name != self.NAME:
raise Exception(f"Invalid name '{name}' returned for Ref ID '{ref_id}'")
# Extract and return score data
self.assert_path(resp, "response/playerdata/base/clear_medal")
def transform_medals(medal: int) -> Tuple[int, int, int, int]:
return (
(medal >> 0) & 0xF,
(medal >> 4) & 0xF,
(medal >> 8) & 0xF,
(medal >> 12) & 0xF,
)
medals = [
transform_medals(medal)
for medal in resp.child("playerdata")
.child("base")
.child("clear_medal")
.value
]
hiscore = resp.child("playerdata").child("hiscore").value
hiscores = []
for i in range(0, len(hiscore) * 8, 17):
byte_offset = int(i / 8)
bit_offset = int(i % 8)
value = hiscore[byte_offset]
value = value + (hiscore[byte_offset + 1] << 8)
value = value + (hiscore[byte_offset + 2] << 16)
value = value >> bit_offset
hiscores.append(value & 0x1FFFF)
scores = [
(hiscores[x], hiscores[x + 1], hiscores[x + 2], hiscores[x + 3])
for x in range(0, len(hiscores), 4)
]
return {"medals": medals, "scores": scores}
else:
raise Exception(f"Unrecognized message type '{msg_type}'")
def verify_playerdata_set(self, ref_id: str, scores: List[Dict[str, Any]]) -> None:
call = self.call_node()
# Construct node
playerdata = Node.void("playerdata")
call.add_child(playerdata)
playerdata.set_attribute("method", "set")
playerdata.set_attribute("ref_id", ref_id)
playerdata.set_attribute("shop_name", "")
# Add required children
playerdata.add_child(Node.s16("chara", 1543))
# Add requested scores
for score in scores:
stage = Node.void("stage")
playerdata.add_child(stage)
stage.add_child(Node.s16("no", score["id"]))
stage.add_child(
Node.u8(
"sheet",
{
0: 2,
1: 0,
2: 1,
3: 3,
}[score["chart"]],
)
)
stage.add_child(
Node.u16("n_data", (score["medal"] << (4 * score["chart"])))
)
stage.add_child(Node.u8("e_data", 0))
stage.add_child(Node.s32("score", score["score"]))
stage.add_child(Node.u8("ojama_1", 0))
stage.add_child(Node.u8("ojama_2", 0))
# Swap with server
resp = self.exchange("pnm20/playerdata", call)
# Verify nodes that cause crashes if they don't exist
self.assert_path(resp, "response/playerdata/name")
name = resp.child("playerdata").child("name").value
if name != self.NAME:
raise Exception(f"Invalid name '{name}' returned for Ref ID '{ref_id}'")
def verify_playerdata_new(self, ref_id: str) -> None:
call = self.call_node()
# Construct node
playerdata = Node.void("playerdata")
call.add_child(playerdata)
playerdata.set_attribute("method", "new")
playerdata.add_child(Node.string("ref_id", ref_id))
playerdata.add_child(Node.string("name", self.NAME))
playerdata.add_child(Node.string("shop_name", ""))
playerdata.add_child(Node.s8("pref", 51))
# Swap with server
resp = self.exchange("pnm20/playerdata", call)
# Verify nodes that cause crashes if they don't exist
self.assert_path(resp, "response/playerdata/base/g_pm_id")
self.assert_path(resp, "response/playerdata/base/name")
self.assert_path(resp, "response/playerdata/base/my_best")
self.assert_path(resp, "response/playerdata/base/latest_music")
self.assert_path(resp, "response/playerdata/base/clear_medal")
self.assert_path(resp, "response/playerdata/base/clear_medal_sub")
self.assert_path(resp, "response/playerdata/player_card")
self.assert_path(resp, "response/playerdata/player_card_ex")
self.assert_path(resp, "response/playerdata/hiscore")
self.assert_path(resp, "response/playerdata/netvs")
self.assert_path(resp, "response/playerdata/sp_data")
self.assert_path(resp, "response/playerdata/reflec_data")
self.assert_path(resp, "response/playerdata/navigate")
def verify(self, cardid: Optional[str]) -> None:
# Verify boot sequence is okay
self.verify_services_get(
expected_services=[
"pcbtracker",
"pcbevent",
"local",
"message",
"facility",
"cardmng",
"package",
"posevent",
"pkglist",
"dlstatus",
"eacoin",
"lobby",
"ntp",
"keepalive",
]
)
paseli_enabled = self.verify_pcbtracker_alive()
self.verify_message_get()
self.verify_package_list()
self.verify_facility_get()
self.verify_pcbevent_put()
self.verify_game_active()
self.verify_game_get()
# Verify card registration and profile lookup
if cardid is not None:
card = cardid
else:
card = self.random_card()
print(f"Generated random card ID {card} for use.")
if cardid is None:
self.verify_cardmng_inquire(
card, msg_type="unregistered", paseli_enabled=paseli_enabled
)
ref_id = self.verify_cardmng_getrefid(card)
if len(ref_id) != 16:
raise Exception(
f"Invalid refid '{ref_id}' returned when registering card"
)
if ref_id != self.verify_cardmng_inquire(
card, msg_type="new", paseli_enabled=paseli_enabled
):
raise Exception(f"Invalid refid '{ref_id}' returned when querying card")
self.verify_playerdata_get(ref_id, msg_type="new")
self.verify_playerdata_new(ref_id)
else:
print("Skipping new card checks for existing card")
ref_id = self.verify_cardmng_inquire(
card, msg_type="query", paseli_enabled=paseli_enabled
)
# Verify pin handling and return card handling
self.verify_cardmng_authpass(ref_id, correct=True)
self.verify_cardmng_authpass(ref_id, correct=False)
if ref_id != self.verify_cardmng_inquire(
card, msg_type="query", paseli_enabled=paseli_enabled
):
raise Exception(f"Invalid refid '{ref_id}' returned when querying card")
if cardid is None:
# Verify score handling
scores = self.verify_playerdata_get(ref_id, msg_type="query")
if scores is None:
raise Exception("Expected to get scores back, didn't get anything!")
for medal in scores["medals"]:
for i in range(4):
if medal[i] != 0:
raise Exception("Got nonzero medals count on a new card!")
for score in scores["scores"]:
for i in range(4):
if score[i] != 0:
raise Exception("Got nonzero scores count on a new card!")
for phase in [1, 2]:
if phase == 1:
dummyscores = [
# An okay score on a chart
{
"id": 987,
"chart": 2,
"medal": 5,
"score": 76543,
},
# A good score on an easier chart of the same song
{
"id": 987,
"chart": 0,
"medal": 6,
"score": 99999,
},
# A bad score on a hard chart
{
"id": 741,
"chart": 3,
"medal": 2,
"score": 45000,
},
# A terrible score on an easy chart
{
"id": 742,
"chart": 1,
"medal": 2,
"score": 1,
},
]
# Random score to add in
songid = random.randint(907, 950)
chartid = random.randint(0, 3)
score = random.randint(0, 100000)
medal = random.choice([1, 2, 3, 5, 6, 7, 9, 10, 11, 15])
dummyscores.append(
{
"id": songid,
"chart": chartid,
"medal": medal,
"score": score,
}
)
if phase == 2:
dummyscores = [
# A better score on the same chart
{
"id": 987,
"chart": 2,
"medal": 5,
"score": 98765,
},
# A worse score on another same chart
{
"id": 987,
"chart": 0,
"medal": 3,
"score": 12345,
"expected_score": 99999,
"expected_medal": 6,
},
]
self.verify_playerdata_set(ref_id, dummyscores)
scores = self.verify_playerdata_get(ref_id, msg_type="query")
for score in dummyscores:
newscore = scores["scores"][score["id"]][score["chart"]]
newmedal = scores["medals"][score["id"]][score["chart"]]
if "expected_score" in score:
expected_score = score["expected_score"]
else:
expected_score = score["score"]
if "expected_medal" in score:
expected_medal = score["expected_medal"]
else:
expected_medal = score["medal"]
if newscore != expected_score:
raise Exception(
f'Expected a score of \'{expected_score}\' for song \'{score["id"]}\' chart \'{score["chart"]}\' but got score \'{newscore}\''
)
if newmedal != expected_medal:
raise Exception(
f'Expected a medal of \'{expected_medal}\' for song \'{score["id"]}\' chart \'{score["chart"]}\' but got medal \'{newmedal}\''
)
# Sleep so we don't end up putting in score history on the same second
time.sleep(1)
else:
print("Skipping score checks for existing card")
# Verify paseli handling
if paseli_enabled:
print("PASELI enabled for this PCBID, executing PASELI checks")
else:
print("PASELI disabled for this PCBID, skipping PASELI checks")
return
sessid, balance = self.verify_eacoin_checkin(card)
if balance == 0:
print("Skipping PASELI consume check because card has 0 balance")
else:
self.verify_eacoin_consume(sessid, balance, random.randint(0, balance))
self.verify_eacoin_checkout(sessid)
Reference in New Issue View Git Blame Copy Permalink