Merge branch 'trunk' into supernova2

# Conflicts:
#	bemani/data/mysql/music.py
This commit is contained in:
Darren Thompson 2025-06-27 13:44:43 -04:00
commit 48361b2da7
52 changed files with 2948 additions and 277 deletions

View File

@ -280,7 +280,7 @@ This should be given the same config file as "api", "frontend" and "services".
Development version of an eAmusement protocol server using flask and the protocol
libraries also used in "bemanishark" and "trafficgen". Currently it lets most modern
BEMANI games boot and supports full scores, profile and events for Beatmania IIDX 20-26,
Pop'n Music 19-26, Jubeat Saucer, Saucer Fulfill, Prop, Qubell, Clan and Festo, Sound
Pop'n Music 19-27, Jubeat Saucer, Saucer Fulfill, Prop, Qubell, Clan and Festo, Sound
Voltex 1, 2, 3 Season 1/2 and 4, Dance Dance Revolution X2, X3, 2013, 2014 and Ace,
MÚSECA 1, MÚSECA 1+1/2, MÚSECA Plus, Reflec Beat, Limelight, Colette, groovin'!! Upper,
Volzza 1 and Volzza 2, Metal Gear Arcade, and finally The\*BishiBashi. Note that it also
@ -306,7 +306,7 @@ which has the paths set up for correct imports.
A convenience utility for helping reverse-engineer structures out of game DLLs/EXEs.
You can give this a physical DLL offset or a virtual memory address for the start and
end of the data as well as a python struct format (documentation at
https://docs.python.org/3.6/library/struct.html) and this will print the decoded
https://docs.python.org/3.8/library/struct.html) and this will print the decoded
data to the screen one entry per line. It includes several enhancements for decoding
pointers to sub-structures and pointers to C strings. Note that much like "psmap", this
has the ability to print out structures that are dynamically constructed at runtime by
@ -328,7 +328,7 @@ this will run through and attempt to verify simple operation of that service. No
guarantees are made on the accuracy of the emulation though I've strived to be
correct. In some cases, I will verify the response, and in other cases I will
simply verify that certain things exist so as not to crash a real client. This
currently generates traffic emulating Beatmania IIDX 20-26, Pop'n Music 19-26, Jubeat
currently generates traffic emulating Beatmania IIDX 20-26, Pop'n Music 19-27, Jubeat
Saucer, Fulfill, Prop, Qubell, Clan and Festo, Sound Voltex 1, 2, 3 Season 1/2 and 4,
Dance Dance Revolution X2, X3, 2013, 2014 and Ace, The\*BishiBashi, MÚSECA 1 and MÚSECA
1+1/2, Reflec Beat, Reflec Beat Limelight, Reflec Beat Colette, groovin'!! Upper,
@ -379,10 +379,10 @@ you aren't introducing any type errors into the codebase.
## Dependency Setup
The code contained here assumes Python 3.6 as the base although it should work with
The code contained here assumes Python 3.8 as the base although it should work with
any newer version of python as well. If you don't have or don't want to install Python
3.6 as your system python, it is recommended to use virtualenv to create a virtual
environment. The rest of the installation will assume you have Python 3.6 working
3.8 as your system python, it is recommended to use virtualenv to create a virtual
environment. The rest of the installation will assume you have Python 3.8 working
properly (and are in an activated virtual environment if this is the route you've
chosen to go). If you have a newer version of python available this code should be
compatible with that as well. This code is designed to run on Linux. However, it has
@ -485,7 +485,7 @@ for how exactly to do that.
### Pop'n Music
For Pop'n Music, get the game DLL from the version of the game you want to import and
run a command like so. This network supports versions 19-26 so you will want to run this
run a command like so. This network supports versions 19-27 so you will want to run this
command once for every version, giving the correct DLL file. Note that there are several
versions of each game floating around and the "read" script attempts to support as many
as it can but you might encounter a version of the game which hasn't been mapped yet.
@ -845,7 +845,7 @@ free certificate that you can manage easily. There are other ways to run this so
and provide SSL credentials but I have no experience with or advice on them.
The easiest way to get up and running is to install MySQL 5.7, nginx and uWSGI along
with Python 3.6 or higher. Create a directory where the services will live and place
with Python 3.8 or higher. Create a directory where the services will live and place
a virtualenv inside it (outside the scope of this document). Then, the wsgi files found
in `bemani/wsgi/` can be placed in the directory, uWSGI pointed at them and nginx set up.
The setup for the top-level package will include all of the frontend templates, so you

View File

@ -46,7 +46,7 @@ def before_request() -> None:
authtype = None
authtoken = None
if authtype.lower() == "token":
if authtype is not None and authtoken is not None and authtype.lower() == "token":
g.authorized = g.data.local.api.validate_client(authtoken)
@ -270,6 +270,7 @@ def lookup(protoversion: str, requestgame: str, requestversion: str) -> Dict[str
"24": VersionConstants.POPN_MUSIC_USANEKO,
"25": VersionConstants.POPN_MUSIC_PEACE,
"26": VersionConstants.POPN_MUSIC_KAIMEI_RIDDLES,
"27": VersionConstants.POPN_MUSIC_UNILAB,
},
GameConstants.REFLEC_BEAT: {
"1": VersionConstants.REFLEC_BEAT,

View File

@ -3,11 +3,12 @@ import binascii
import base64
try:
# Python <= 3.9
from collections import Iterable
except ImportError:
# Python > 3.9
from collections.abc import Iterable
except ImportError:
# Python <= 3.9
from collections import Iterable # type: ignore
from typing import Any, Dict, List, Sequence, Union
from bemani.backend.bishi.base import BishiBashiBase

View File

@ -167,7 +167,7 @@ class JubeatBase(CoreHandler, CardManagerHandler, PASELIHandler, Base):
return None
cache_key = f"get_scores_by_extid-{extid}"
score: Optional[List[Score]]
scores: Optional[List[Score]]
if partition == 1:
# We fetch all scores on the first partition and then divy up
@ -176,9 +176,13 @@ class JubeatBase(CoreHandler, CardManagerHandler, PASELIHandler, Base):
scores = self.data.remote.music.get_scores(self.game, self.music_version, userid)
else:
# We will want to fetch the remaining scores that were in our
# cache.
scores = self.cache.get(cache_key) # type: ignore
# cache. If the cache is empty, due to some error, or because
# we cached nothing below, then we will end up returning an
# empty list. This shouldn't happen, but guard against crashing
# if this returns None anyway.
scores = self.cache.get(cache_key) or []
rest: List[Score]
if len(scores) < 50:
# We simply return the whole amount for this, and cache nothing.
rest = []
@ -360,9 +364,9 @@ class JubeatBase(CoreHandler, CardManagerHandler, PASELIHandler, Base):
normalindex = 2
premiumindex = 1
if normalemblems:
normalindex = random.sample(normalemblems, 1)[0]
normalindex = random.sample(list(normalemblems), 1)[0]
if premiumemblems:
premiumindex = random.sample(premiumemblems, 1)[0]
premiumindex = random.sample(list(premiumemblems), 1)[0]
return normalindex, premiumindex

View File

@ -95,10 +95,12 @@ class JubeatClan(
# range, but it will be a different ID depending on the prefecture set in settings. This means its not safe to send
# these song IDs, so we explicitly exclude them.
start_time, end_time = data.local.network.get_schedule_duration("daily")
all_songs = set(
song.id
for song in data.local.music.get_all_songs(cls.game, cls.version)
if song.id not in cls.FIVE_PLAYS_UNLOCK_EVENT_SONG_IDS
all_songs = list(
set(
song.id
for song in data.local.music.get_all_songs(cls.game, cls.version)
if song.id not in cls.FIVE_PLAYS_UNLOCK_EVENT_SONG_IDS
)
)
if len(all_songs) >= 2:
daily_songs = random.sample(all_songs, 2)

View File

@ -151,7 +151,7 @@ class JubeatFesto(
if data.local.network.should_schedule(cls.game, cls.version, "fc_challenge", "daily"):
# Generate a new list of two FC challenge songs.
start_time, end_time = data.local.network.get_schedule_duration("daily")
all_songs = set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version))
all_songs = list(set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version)))
if len(all_songs) >= 2:
daily_songs = random.sample(all_songs, 2)
data.local.game.put_time_sensitive_settings(
@ -530,10 +530,14 @@ class JubeatFesto(
],
[
(
80000034
if dataver < 2020062900
else (
30000108 if dataver < 2020091300 else (40000107 if dataver < 2021020100 else 30000004)
(
80000034
if dataver < 2020062900
else (
30000108
if dataver < 2020091300
else (40000107 if dataver < 2021020100 else 30000004)
)
),
0,
),

View File

@ -335,7 +335,7 @@ class JubeatProp(
if data.local.network.should_schedule(cls.game, cls.version, "league_course", "weekly"):
# Generate a new league course list, save it to the DB.
start_time, end_time = data.local.network.get_schedule_duration("weekly")
all_songs = set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version))
all_songs = list(set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version)))
if len(all_songs) >= 3:
league_songs = random.sample(all_songs, 3)
data.local.game.put_time_sensitive_settings(
@ -382,7 +382,7 @@ class JubeatProp(
if data.local.network.should_schedule(cls.game, cls.version, "fc_challenge", "daily"):
# Generate a new list of two FC challenge songs.
start_time, end_time = data.local.network.get_schedule_duration("daily")
all_songs = set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version))
all_songs = list(set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version)))
if len(all_songs) >= 2:
daily_songs = random.sample(all_songs, 2)
data.local.game.put_time_sensitive_settings(

View File

@ -64,7 +64,7 @@ class JubeatQubell(
if data.local.network.should_schedule(cls.game, cls.version, "fc_challenge", "daily"):
# Generate a new list of two FC challenge songs.
start_time, end_time = data.local.network.get_schedule_duration("daily")
all_songs = set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version))
all_songs = list(set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version)))
if len(all_songs) >= 2:
daily_songs = random.sample(all_songs, 2)
data.local.game.put_time_sensitive_settings(

View File

@ -42,7 +42,7 @@ class JubeatSaucer(
if data.local.network.should_schedule(cls.game, cls.version, "fc_challenge", "daily"):
# Generate a new list of two FC challenge songs.
start_time, end_time = data.local.network.get_schedule_duration("daily")
all_songs = set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version))
all_songs = list(set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version)))
if all_songs:
today_song = random.sample(all_songs, 1)[0]
data.local.game.put_time_sensitive_settings(

View File

@ -54,7 +54,7 @@ class JubeatSaucerFulfill(
if data.local.network.should_schedule(cls.game, cls.version, "fc_challenge", "daily"):
# Generate a new list of two FC challenge songs.
start_time, end_time = data.local.network.get_schedule_duration("daily")
all_songs = set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version))
all_songs = list(set(song.id for song in data.local.music.get_all_songs(cls.game, cls.version)))
if len(all_songs) >= 2:
daily_songs = random.sample(all_songs, 2)
data.local.game.put_time_sensitive_settings(

View File

@ -29,6 +29,7 @@ from bemani.backend.popn.eclale import PopnMusicEclale
from bemani.backend.popn.usaneko import PopnMusicUsaNeko
from bemani.backend.popn.peace import PopnMusicPeace
from bemani.backend.popn.kaimei import PopnMusicKaimei
from bemani.backend.popn.unilab import PopnMusicUnilab
from bemani.common import Model, VersionConstants
from bemani.data import Config, Data
@ -61,6 +62,7 @@ class PopnMusicFactory(Factory):
PopnMusicUsaNeko,
PopnMusicPeace,
PopnMusicKaimei,
PopnMusicUnilab,
]
@classmethod
@ -87,8 +89,10 @@ class PopnMusicFactory(Factory):
return VersionConstants.POPN_MUSIC_USANEKO
if date >= 2018101700 and date < 2021042600:
return VersionConstants.POPN_MUSIC_PEACE
if date >= 2021042600:
if date >= 2021042600 and date < 2022091300:
return VersionConstants.POPN_MUSIC_KAIMEI_RIDDLES
if date >= 2022091300:
return VersionConstants.POPN_MUSIC_UNILAB
return None
if model.gamecode == "G15":
@ -131,6 +135,8 @@ class PopnMusicFactory(Factory):
return PopnMusicUsaNeko(data, config, model)
if parentversion == VersionConstants.POPN_MUSIC_KAIMEI_RIDDLES:
return PopnMusicPeace(data, config, model)
if parentversion == VersionConstants.POPN_MUSIC_UNILAB:
return PopnMusicKaimei(data, config, model)
# Unknown older version
return None
@ -148,6 +154,8 @@ class PopnMusicFactory(Factory):
return PopnMusicPeace(data, config, model)
if version == VersionConstants.POPN_MUSIC_KAIMEI_RIDDLES:
return PopnMusicKaimei(data, config, model)
if version == VersionConstants.POPN_MUSIC_UNILAB:
return PopnMusicUnilab(data, config, model)
# Unknown game version
return None

View File

@ -47,7 +47,7 @@ class PopnMusicKaimei(PopnMusicModernBase):
"category": "game_config",
"setting": "music_phase",
"values": {
# The value goes to 30 now, but it starts where usaneko left off at 23
# The value goes to 30 now, but it starts where peace left off at 23
# Unlocks a total of 10 songs
23: "No music unlocks",
24: "Phase 1",
@ -149,7 +149,7 @@ class PopnMusicKaimei(PopnMusicModernBase):
"category": "game_config",
"setting": "peace_soundtrack",
"values": {
0: "Not stated",
0: "Not started",
1: "Active",
2: "Ended",
},
@ -160,7 +160,7 @@ class PopnMusicKaimei(PopnMusicModernBase):
"category": "game_config",
"setting": "tanteisha_joshu",
"values": {
0: "Not stated",
0: "Not started",
1: "Active",
2: "Ended",
},

View File

@ -0,0 +1,715 @@
# vim: set fileencoding=utf-8
import math
import random
from typing import Any, Dict, List, Tuple
from bemani.backend.popn.base import PopnMusicBase
from bemani.backend.popn.common import PopnMusicModernBase
from bemani.backend.popn.kaimei import PopnMusicKaimei
from bemani.common import VersionConstants
from bemani.common.validateddict import Profile
from bemani.data.types import UserID
from bemani.protocol.node import Node
class PopnMusicUnilab(PopnMusicModernBase):
name: str = "Pop'n Music Unilab"
version: int = VersionConstants.POPN_MUSIC_UNILAB
# Biggest ID in the music DB
GAME_MAX_MUSIC_ID: int = 2188
# Biggest deco part ID in the game
GAME_MAX_DECO_ID: int = 81
def previous_version(self) -> PopnMusicBase:
return PopnMusicKaimei(self.data, self.config, self.model)
@classmethod
def get_settings(cls) -> Dict[str, Any]:
"""
Return all of our front-end modifiably settings.
"""
return {
"ints": [
{
"name": "Music Open Phase",
"tip": "Default music phase for all players.",
"category": "game_config",
"setting": "music_phase",
"values": {
# The value goes to 30 now, but it starts where usaneko left off at 23
# Unlocks a total of 10 songs
0: "No music unlocks",
1: "Phase 1",
2: "Phase 2",
3: "Phase 3",
4: "Phase 4",
5: "Phase 5",
6: "Phase MAX",
},
},
{
# Shutchou! pop'n quest Lively II event
"name": "Shutchou! pop'n quest Lively II phase",
"tip": "Shutchou! pop'n quest Lively II phase for all players.",
"category": "game_config",
"setting": "popn_quest_lively_2",
"values": {
0: "Not started",
1: "fes 1",
2: "fes 2",
3: "fes FINAL",
4: "fes EXTRA",
5: "fes THE END",
6: "Ended",
},
},
{
"name": "Narunaru♪ UniLab jikkenshitsu! event Phase",
"tip": "Narunaru♪ UniLab jikkenshitsu! event Phase for all players.",
"category": "game_config",
"setting": "narunaru_phase",
"values": {
0: "Disabled",
1: "ラブケミ / 悪夢♡ショコラティエ",
2: "001 -どうしんのかいろ-",
3: "MA・TSU・RI / MOVE! (We Keep It Movin')",
4: "斑咲花 / ユメブキ",
5: "ホムンクルスレシピ",
6: "脳ミソ de 向上",
7: "Awakening Wings",
8: "カタルシスの月 (UPPER) / ちくわパフェだよ☆CKP (UPPER) / ホーンテッド★メイドランチ (UPPER)",
9: "HAGURUMA / ノープラン・デイズ / Sweet Illusion",
10: "左脳スパーク (UPPER)",
11: "東京メモリー",
12: "にゃんのパレードマーチ♪",
13: "明滅の果てに",
14: "Shout It Out",
15: "グランデーロの守り",
16: "恋するMonstro",
17: "Versa (UPPER)",
18: "Xジェネの逆襲",
19: "Engraved on my heart ft. 小林マナ",
20: "fallen leaves -IIDX edition-",
21: "Τέλος",
22: "Candy Crime Toe Shoes",
23: "High Speed Junkie!",
24: "Pure Rude",
25: "地方創生☆チクワクティクス (UPPER) / 乙女繚乱 舞い咲き誇れ (UPPER)",
26: "pastel@sweets labo(*'v'*) / 恋はどうモロ◎波動OK☆方程式 (UPPER) / Mecha Kawa Breaker!!",
27: "あまるがむ",
28: "勇猛無比",
29: "Unknown Region",
30: "unisonote",
31: "灰の羽搏",
32: "情熱タンデムRUNAWAY",
33: "Satan",
34: "粋 -IKI-",
35: "Treasure Hoard (UPPER)",
36: "SOLID STATE SQUAD -RISEN RELIC REMIX-",
37: "夏色のセーブデータ",
38: "革命パッショネイト (UPPER) / めうめうぺったんたん!! (UPPER)",
39: "Gabbalungang",
40: "Caldwell 99",
41: "葬送のエウロパ / ただ、それだけの理由で",
42: "ISERBROOK",
43: "Amulet of Enbarr",
44: "Sword of Vengeance",
45: "Caldwell 99",
46: "満漢全席火花ノ舞",
47: "mathematical good-bye → Hexer → F/S",
48: "Ended",
},
},
{
# Kakusei no Elem event Phase
"name": "Kakusei no Elem event Phase",
"tip": "Kakusei no Elem event Phase for all players.",
"category": "game_config",
"setting": "kakusei_phase",
"values": {
0: "Disabled",
1: "Tan♪Tan♪Tan♪",
2: "Keep the Faith",
3: "Lovin' You",
4: "Redemption Tears",
5: "Dancin' in シャングリラ",
6: "ココロコースター",
7: "ma plume / ma plume (UPPER)",
8: "いばら姫",
9: "螺旋",
10: "めうめうぺったんたん!! (ZAQUVA Remix) / ちくわパフェだよ☆CKP (Yvya Remix)",
11: "狼弦暴威",
12: "The Escape",
13: "謎情の雫 ft. Kanae Asaba",
14: "黒猫と珈琲",
15: "Head Scratcher",
16: "ドーナツホール (UPPER) / マトリョシカ (UPPER)",
17: "遊戯大熊猫",
18: "Stylus",
19: "Crazy Shuffle",
20: "speedstar[02]",
21: "少年A",
22: "what I wish",
23: "TAKE YOU AWAY",
24: "Dragon Blade -The Arrange-",
25: "Pump up dA CORE",
26: "TURBO BOOSTER",
27: "夜虹",
28: "天泣 ",
29: "オッタマゲッター",
30: "luck (UPPER) / 脳漿炸裂ガール (UPPER)",
31: "TYPHØN",
32: "REFLEXED MANIPULATION",
33: "オーバー ",
34: "Knockin' On Red Button",
35: "The Metalist",
36: "イマココ!この瞬間 ",
37: "チョコレートスマイル (UPPER)",
38: "キリステゴメン (UPPER)",
39: "Liar×Girl / Hades Doll",
40: "Jazz is Rad / アモ",
41: "encounter / 不可説不可説転",
42: "弾幕信仰 / 閉塞的フレーション / 残像ニ繋ガレタ追憶HIDEAWAY",
43: "ROBOROS OVERDIVE / Megalara Garuda",
44: "Megalara Garuda (UPPER)",
},
},
{
# Awakening Boost
"name": "Super Unilab BOOST!",
"tip": "Super Unilab BOOST! for all players.",
"category": "game_config",
"setting": "super_unilab_boost",
"values": {
0: "Disabled",
1: "Active",
2: "Ended",
},
},
{
# CanCan's Super Awakening Boost
"name": "CanCan's Super Awakening Boost",
"tip": "CanCan's Super Awakening Boost for all players.",
"category": "game_config",
"setting": "cancan_boost",
"values": {
0: "Disabled",
1: "Active",
2: "Ended",
},
},
{
# KAC 2023
"name": "KAC Lab Phase",
"tip": "KAC Lab for all players",
"category": "game_config",
"setting": "kac_2023",
"values": {
0: "Not Started",
1: "Caldwell 99 (KAC Woman/Free Set A)",
3: "Hexer / mathematical good-bye (KAC Woman/Free Set B)",
4: "Ended",
},
},
# We don't currently support lobbies or anything, so this is commented out until
# somebody gets around to implementing it.
# {
# # Net Taisen and local mode
# "name": "Net Taisen / Local Mode",
# "tip": "Enable Net Taisen and Local Mode",
# "category": "game_config",
# "setting": "enable_net_taisen_local_mode",
# "values": {
# 0: "Disabled",
# 1: "Net Taisen",
# 2: "Net Taisen / Local Mode",
# },
# },
],
"bools": [
{
"name": "Force Song Unlock",
"tip": "Force unlock all songs.",
"category": "game_config",
"setting": "force_unlock_songs",
},
{
"name": "Force Deco Unlock",
"tip": "Force unlock all Deco parts.",
"category": "game_config",
"setting": "force_unlock_deco",
},
{
"name": "Unlock KAC Qualifier (パーフェクトイーター)",
"tip": "Force unlock Perfect Eater for all players.",
"category": "game_config",
"setting": "force_unlock_perfect_eater",
},
{
# Overly complicated event where you'd play songs from other games to unlock them in other games.
# Unlocks the following songs after one play when set:
# 2045 - 鴉
# 2046 - 蒼氷のフラグメント
# 2047 - Indigo Nocturne
# 2048 - 輪廻の鴉
# 2049 - VOLAQUAS
"name": "Unlock いちかのごちゃまぜMix UP Songs",
"tip": "Force unlock Ichika no Gochamaze Mix UP! songs for all players.",
"category": "game_config",
"setting": "force_unlock_ichika",
},
],
}
def get_common_config(self) -> Tuple[Dict[int, int], bool]:
game_config = self.get_game_config()
music_phase = game_config.get_int("music_phase")
narunaru_phase = game_config.get_int("narunaru_phase")
enable_net_taisen = False # game_config.get_bool('enable_net_taisen')
super_unilab_boost = game_config.get_int("super_unilab_boost")
cancan_boost = game_config.get_int("cancan_boost")
kakusei_phase = game_config.get_int("kakusei_phase")
popn_quest_lively_2 = game_config.get_int("popn_quest_lively_2")
kac_2023 = game_config.get_int("kac_2023")
# Enable event and mark complete
if game_config.get_bool("force_unlock_deco"):
kakusei_phase = 1
# Event phases
return (
{
# Default song phase availability (0-6)
# 1 - 2071 - Hopes and Dreams/夢と希望
# 2072 - MEGALOVANIA
# 2073 - Battle Against a True Hero/本物のヒーローとの戦い
# 2 - 2146 - ポラリスノウタ
# 3 - 2149 - 第ゼロ感
# 4 - 2150 - 強風オールバック
# 2151 - 恋愛パクチー
# 5 - 2172 - レイドバックジャーニー
# 6 - 2188 - Super Heroine
0: music_phase,
# Shutchou! pop'n quest Lively II (0-6)
# When active, the following songs are available for unlock
# 1 - 1989 - Ketter
# 1990 - Petit Queen
# 1991 - 波と凪の挟間で
# 2 - 1984 - コルドバの女
# 1985 - say...but in vain
# 1992 - Northern Cross
# 3 - 1982 - Surf on the Light
# 1983 - バッドエンド・シンドローム
# 1988 - Danza Pantera
# 4 - 1986 - virkatoの主題によるperson09風超絶技巧変奏曲
# 1987 - 水晶塔のオルカ
# 1993 - Un Happy Heart
# 5 - 2017 - virkatoの主題によるperson09風超絶技巧変奏曲 upper
# 6 - Event Ended
1: popn_quest_lively_2,
# KAC 2023 (0-4) - Please see the site below for what songs are in set A and set B
# https://bemaniwiki.com/?%B8%F8%BC%B0%C2%E7%B2%F1/KONAMI+Arcade+Championship%282023%29/%CD%BD%C1%AA%A5%E9%A5%A6%A5%F3%A5%C9#popn
# 0 - Disabled
# 1 - Caldwell 99 (KAC Woman/Free Set A)
# 2 - Disabled
# 3 - Hexer / mathematical good-bye (KAC Woman/Free Set B)
# 4 - Disabled
2: kac_2023,
# Enable Net Taisen, including win/loss display on song select (0-2)
# 0 - Disable
# 1 - Net taisen
# 2 - Net taisen + Local mode
3: 1 if enable_net_taisen else 0,
# Unknown event (0-7)
4: 7,
# Narunaru♪ UniLab jikkenshitsu! (0-48)
# 6500 clear points are needed unless otherwise specified
# 1 - 2040 - ラブケミ - 1000 points
# 2043 - 悪夢♡ショコラティエ
# 2 - 2044 - 001 -どうしんのかいろ-
# 3 - 2050 - MA・TSU・RI
# 2051 - MOVE! (We Keep It Movin')
# 4 - 2052 - 斑咲花
# 2053 - ユメブキ
# 5 - 2054 - ホムンクルスレシピ
# 6 - 2055 - 脳ミソ de 向上
# 7 - 2059 - Awakening Wings
# 8 - 2062 - カタルシスの月 upper - 5000 points
# 2061 - ホーンテッド★メイドランチ upper - 5000 points
# 2060 - ちくわパフェだよ☆CKP upper - 5000 points
# 9 - 2074 - HAGURUMA - 5000 points
# 2075 - ノープラン・デイズ - 5000 points
# 502 - Sweet Illusion [ex] - 5000 points
# 10 - 2076 - 左脳スパーク upper - 5000 points
# 11 - 2077 - 東京メモリー - 5000 points
# 12 - 2078 - にゃんのパレードマーチ♪ - 5000 points
# 13 - 2079 - 明滅の果てに - 5000 points
# 14 - 2080 - Shout It Out
# 15 - 2081 - グランデーロの守り
# 16 - 2082 - 恋するMonstro
# 17 - 2083 - Versa upper
# 18 - 2084 - Xジェネの逆襲
# 19 - 2085 - Engraved on my heart ft. 小林マナ
# 20 - 2086 - fallen leaves -IIDX edition-
# 21 - 2087 - Τέλος
# 22 - 2088 - Candy Crime Toe Shoes
# 23 - 2089 - High Speed Junkie!
# 24 - 2090 - Pure Rude
# 25 - 2092 - 地方創生☆チクワクティクス upper - 5000 points
# 2091 - 乙女繚乱 舞い咲き誇れ upper - 5000 points
# 26 - 2093 - pastel@sweets labo(*'v'*)
# 2095 - 恋はどうモロ◎波動OK☆方程式 upper - 5000 points
# 2094 - Mecha Kawa Breaker!!
# 27 - 2096 - あまるがむ
# 28 - 2097 - 勇猛無比
# 29 - 2098 - Unknown Region
# 30 - 2110 - unisonote
# 31 - 2107 - 灰の羽搏
# 32 - 2113 - 情熱タンデムRUNAWAY
# 33 - 2108 - Satan
# 34 - 2111 - 粋 -IKI-
# 35 - 2112 - Treasure Hoard upper
# 36 - 2109 - SOLID STATE SQUAD -RISEN RELIC REMIX-
# 37 - 2114 - 夏色のセーブデータ
# 38 - 2117 - めうめうぺったんたん!! upper - 5000 points
# 2116 - 革命パッショネイト upper - 5000 points
# 39 - 2118 - Gabbalungang
# 40 - 2120 - Caldwell 99 - 13000 points
# 41 - 2065 - 葬送のエウロパ
# 2064 - ただ、それだけの理由で
# 42 - 2121 - ISERBROOK
# 43 - 2122 - Amulet of Enbarr
# 44 - 2123 - Sword of Vengeance
# 45 - 2120 - Caldwell 99 (set B) - 13000 points
# 46 - 2124 - 満漢全席火花ノ舞
# 47 - 2126 - mathematical good-bye - 13000 points
# 2125 - Hexer - 13000 points
# 2127 - F/S - 14000
# 48 - Ended
5: narunaru_phase,
# Super Unilab BOOST! (0-2)
# Boost should be 120, 150, or 200, bemaniwiki has the explanation and it's based on the unlocks left to do
6: super_unilab_boost,
# Unknown event (0-6)
7: 6,
# Unknown event (0-2)
8: 2,
# Kakusei no Elem - Awakening Elem (0-44)
# Songs are unlocked as a percentage of 280 points unless otherwise specified
# 0 - Disabled
# 1 - 2128 - Tan♪Tan♪Tan♪
# 2 - 2129 - Keep the Faith
# 3 - 2130 - Lovin' You
# 4 - 2131 - Redemption Tears
# 5 - 2132 - Dancin' in シャングリラ
# 6 - 2133 - ココロコースター
# 7 - 2136 - ma plume
# 2137 - ma plume upper
# 8 - 2135 - いばら姫
# 9 - 2134 - 螺旋
# 10 - 2147 - めうめうぺったんたん!! (ZAQUVA Remix)
# 2148 - ちくわパフェだよ☆CKP (Yvya Remix)
# 11 - 2152 - 狼弦暴威
# - Player must complete the first 12 before this will show up in the event
# 12 - 2067 - The Escape
# 13 - 2153 - 謎情の雫 ft. Kanae Asaba
# 14 - 2154 - 黒猫と珈琲
# 15 - 2155 - Head Scratcher
# 16 - 2156 - ドーナツホール upper - 230 points
# 2157 - マトリョシカ upper - 230 points
# 17 - 2063 - 遊戯大熊猫
# 18 - 2138 - Stylus
# 19 - 2158 - Crazy Shuffle
# 20 - 2159 - speedstar[02]
# 21 - 2160 - 少年A
# 22 - 2161 - what I wish
# 23 - 2068 - TAKE YOU AWAY
# 24 - 2070 - Dragon Blade -The Arrange-
# 25 - 2162 - Pump up dA CORE
# 26 - 2175 - TURBO BOOSTER
# 27 - 2176 - 夜虹
# 28 - 2066 - 天泣
# 29 - 2173 - オッタマゲッター
# 30 - 2177 - luck upper - 230 points
# 2178 - 脳漿炸裂ガール upper - 230 points
# 31 - 2179 - TYPHØN
# 32 - 2069 - REFLEXES MANIPULATION
# 33 - 2174 - オーバー
# 34 - 2115 - Knockin' On Red Button
# 35 - 2180 - The Metalist
# 36 - 2181 - イマココ!この瞬間
# 37 - 2182 - チョコレートスマイル upper - 230 points
# 38 - 2183 - キリステゴメン upper - 230 points
# 39 - 2099 - Liar×Girl
# 2101 - Hades Doll
# 40 - 2102 - Jazz is Rad
# 2104 - アモ
# 41 - 2100 - encounter
# 2103 - 不可説不可説転
# 42 - 2185 - 閉塞的フレーション
# 2186 - 残像ニ繋ガレタ追憶HIDEAWAY
# 2187 - 弾幕信仰
# 43 - 2105 - UROBOROS OVERDIVE
# 2106 - Megalara Garuda
# 44 - 2184 - Megalara Garuda upper
9: kakusei_phase,
# Enable Awakening Elem (0-1)
10: 1 if (kakusei_phase > 0) else 0,
# CanCan's Super Awakening Boost (0-2)
11: cancan_boost,
# Unknown event (0-2)
12: 2,
# Unknown event (0-2)
13: 2,
},
False,
)
def format_profile(self, userid: UserID, profile: Profile) -> Node:
root = super().format_profile(userid, profile)
account = root.child("account")
account.add_child(Node.s16("sp_riddles_id", profile.get_int("sp_riddles_id")))
# options
option = root.child("option")
option_dict = profile.get_dict("option")
option.add_child(Node.bool("lift", option_dict.get_bool("lift")))
option.add_child(Node.s16("lift_rate", option_dict.get_int("lift_rate")))
# Kaimei riddles events
event2021 = Node.void("event2021")
root.add_child(event2021)
event2021.add_child(Node.u32("point", profile.get_int("point")))
event2021.add_child(Node.u8("step", profile.get_int("step")))
event2021.add_child(Node.u32_array("quest_point", profile.get_int_array("quest_point", 8, [0] * 8)))
event2021.add_child(Node.u8("step_nos", profile.get_int("step_nos")))
event2021.add_child(Node.u32_array("quest_point_nos", profile.get_int_array("quest_point_nos", 13, [0] * 13)))
riddles_data = Node.void("riddles_data")
root.add_child(riddles_data)
# Generate Short Riddles for MN tanteisha
randomRiddles: List[int] = []
for _ in range(3):
riddle = 0
while True:
riddle = math.floor(random.randrange(1, 21, 1))
try:
randomRiddles.index(riddle)
except ValueError:
break
randomRiddles.append(riddle)
sh_riddles = Node.void("sh_riddles")
riddles_data.add_child(sh_riddles)
sh_riddles.add_child(Node.u32("sh_riddles_id", riddle))
# Set up kaimei riddles achievements
achievements = self.data.local.user.get_achievements(self.game, self.version, userid)
for achievement in achievements:
if achievement.type == "riddle":
kaimei_gauge = achievement.data.get_int("kaimei_gauge")
is_cleared = achievement.data.get_bool("is_cleared")
riddles_cleared = achievement.data.get_bool("riddles_cleared")
select_count = achievement.data.get_int("select_count")
other_count = achievement.data.get_int("other_count")
sp_riddles = Node.void("sp_riddles")
riddles_data.add_child(sp_riddles)
sp_riddles.add_child(Node.u16("kaimei_gauge", kaimei_gauge))
sp_riddles.add_child(Node.bool("is_cleared", is_cleared))
sp_riddles.add_child(Node.bool("riddles_cleared", riddles_cleared))
sp_riddles.add_child(Node.u8("select_count", select_count))
sp_riddles.add_child(Node.u32("other_count", other_count))
# Narunaru♪ UniLab jikkenshitsu! event
event_p27 = Node.void("event_p27")
root.add_child(event_p27)
event_p27.add_child(Node.s16("team_id", profile.get_int("team_id")))
event_p27.add_child(Node.bool("first_play", profile.get_bool("first_play", True)))
event_p27.add_child(Node.s16("select_battery_id", profile.get_int("select_battery_id", 1)))
event_p27.add_child(Node.bool("elem_first_play", profile.get_bool("elem_first_play", True)))
event_p27.add_child(Node.bool("today_first_play", profile.get_bool("today_first_play", True)))
# Set up Narunaru♪ UniLab jikkenshitsu! achievements
for achievement in achievements:
if achievement.type == "lab":
team_id = achievement.data.get_int("team_id")
ex_no = achievement.data.get_int("ex_no")
point = achievement.data.get_int("point")
is_cleared = achievement.data.get_bool("is_cleared")
team = Node.void("team")
event_p27.add_child(team)
team.add_child(Node.s16("team_id", team_id))
team.add_child(Node.s16("ex_no", ex_no))
team.add_child(Node.u32("point", point))
team.add_child(Node.bool("is_cleared", is_cleared))
# Set up Kakusei no Elem achievements
game_config = self.get_game_config()
if game_config.get_bool("force_unlock_deco"):
battery = Node.void("battery")
event_p27.add_child(battery)
battery.add_child(Node.s16("battery_id", 1))
battery.add_child(Node.u32("energy", 300))
battery.add_child(Node.bool("is_cleared", True))
else:
for achievement in achievements:
if achievement.type == "battery":
battery_id = achievement.data.get_int("battery_id")
energy = achievement.data.get_int("energy")
is_cleared = achievement.data.get_bool("is_cleared")
battery = Node.void("battery")
event_p27.add_child(battery)
battery.add_child(Node.s16("battery_id", battery_id))
battery.add_child(Node.u32("energy", energy))
battery.add_child(Node.bool("is_cleared", is_cleared))
return root
def unformat_profile(self, userid: UserID, request: Node, oldprofile: Profile) -> Profile:
newprofile = super().unformat_profile(userid, request, oldprofile)
game_config = self.get_game_config()
account = request.child("account")
if account is not None:
newprofile.replace_int("card_again_count", account.child_value("card_again_count"))
newprofile.replace_int("sp_riddles_id", account.child_value("sp_riddles_id"))
option_dict = newprofile.get_dict("option")
option = request.child("option")
if option is not None:
option_dict.replace_bool("lift", option.child_value("lift"))
option_dict.replace_int("lift_rate", option.child_value("lift_rate"))
newprofile.replace_dict("option", option_dict)
# Kaimei riddles events
event2021 = request.child("event2021")
if event2021 is not None:
newprofile.replace_int("point", event2021.child_value("point"))
newprofile.replace_int("step", event2021.child_value("step"))
newprofile.replace_int_array("quest_point", 8, event2021.child_value("quest_point"))
newprofile.replace_int("step_nos", event2021.child_value("step_nos"))
newprofile.replace_int_array("quest_point_nos", 13, event2021.child_value("quest_point_nos"))
# Extract kaimei riddles achievements
for node in request.children:
if node.name == "riddles_data":
riddle_id = 0
playedRiddle = request.child("account").child_value("sp_riddles_id")
for riddle in node.children:
kaimei_gauge = riddle.child_value("kaimei_gauge")
is_cleared = riddle.child_value("is_cleared")
riddles_cleared = riddle.child_value("riddles_cleared")
select_count = riddle.child_value("select_count")
other_count = riddle.child_value("other_count")
if riddles_cleared or select_count >= 3:
select_count = 3
elif playedRiddle == riddle_id:
select_count += 1
self.data.local.user.put_achievement(
self.game,
self.version,
userid,
riddle_id,
"riddle",
{
"kaimei_gauge": kaimei_gauge,
"is_cleared": is_cleared,
"riddles_cleared": riddles_cleared,
"select_count": select_count,
"other_count": other_count,
},
)
riddle_id += 1
# Unilab event
event_p27 = request.child("event_p27")
if event_p27 is not None:
newprofile.replace_int("team_id", event_p27.child_value("team_id"))
newprofile.replace_bool("first_play", False)
newprofile.replace_bool("select_battery_id", event_p27.child_value("select_battery_id"))
newprofile.replace_bool("elem_first_play", False)
newprofile.replace_bool("today_first_play", False)
# Extract Narunaru♪ UniLab jikkenshitsu! achievements
lab_data = event_p27.child("team")
if lab_data is not None:
team_id = lab_data.child_value("team_id")
ex_no = lab_data.child_value("ex_no")
point = lab_data.child_value("point")
is_cleared = lab_data.child_value("is_cleared")
self.data.local.user.put_achievement(
self.game,
self.version,
userid,
ex_no,
"lab",
{
"team_id": team_id,
"ex_no": ex_no,
"point": point,
"is_cleared": is_cleared,
},
)
# Extract Kakusei no Elem achievements
battery_data = event_p27.child("battery")
if battery_data is not None:
battery_id = battery_data.child_value("battery_id")
energy = battery_data.child_value("energy")
is_cleared = battery_data.child_value("is_cleared")
if not game_config.get_bool("force_unlock_deco"):
self.data.local.user.put_achievement(
self.game,
self.version,
userid,
battery_id,
"battery",
{
"battery_id": battery_id,
"energy": energy,
"is_cleared": is_cleared,
},
)
# Unlock 2119 - Perfect Eater, KAC Qualifier song after one play. Opens KAC Lab.
if game_config.get_bool("force_unlock_perfect_eater"):
self.data.local.user.put_achievement(
self.game,
self.version,
userid,
2119,
"item_0",
{
"param": 0,
"is_new": False,
"get_time": 0,
},
)
# Unlock Ichika no gochamaze Mix UP! songs after one play.
if game_config.get_bool("force_unlock_ichika"):
for songid in range(2045, 2050): # song IDs 2045 to 2049
self.data.local.user.put_achievement(
self.game,
self.version,
userid,
songid,
"item_0",
{
"param": 0,
"is_new": False,
"get_time": 0,
},
)
return newprofile

View File

@ -429,6 +429,7 @@ class ReflecBeatGroovin(ReflecBeatBase):
self.version,
songid=songid,
)
profiles.update({u: p for u, p in self.get_any_profiles([s[0] for s in allscores if s[0] not in profiles])})
for ng in [
self.CHART_TYPE_BASIC,

View File

@ -213,6 +213,7 @@ class ReflecBeatVolzzaBase(ReflecBeatBase):
self.version,
songid=songid,
)
profiles.update({u: p for u, p in self.get_any_profiles([s[0] for s in allscores if s[0] not in profiles])})
for ng in [
self.CHART_TYPE_BASIC,

View File

@ -6,6 +6,7 @@ from bemani.client.popn.eclale import PopnMusicEclaleClient
from bemani.client.popn.usaneko import PopnMusicUsaNekoClient
from bemani.client.popn.peace import PopnMusicPeaceClient
from bemani.client.popn.kaimei import PopnMusicKaimeiClient
from bemani.client.popn.unilab import PopnMusicUnilabClient
__all__ = [
@ -17,4 +18,5 @@ __all__ = [
"PopnMusicUsaNekoClient",
"PopnMusicPeaceClient",
"PopnMusicKaimeiClient",
"PopnMusicUnilabClient",
]

View File

@ -0,0 +1,683 @@
import random
import time
from typing import Any, Dict, Optional
from bemani.client.base import BaseClient
from bemani.protocol import Node
class PopnMusicUnilabClient(BaseClient):
NAME = ""
def verify_pcb24_boot(self, loc: str) -> None:
call = self.call_node()
# Construct node
pcb24 = Node.void("pcb24")
call.add_child(pcb24)
pcb24.set_attribute("method", "boot")
pcb24.add_child(Node.string("loc_id", loc))
pcb24.add_child(Node.u8("loc_type", 0))
pcb24.add_child(Node.string("loc_name", ""))
pcb24.add_child(Node.string("country", "US"))
pcb24.add_child(Node.string("region", "."))
pcb24.add_child(Node.s16("pref", 51))
pcb24.add_child(Node.string("customer", ""))
pcb24.add_child(Node.string("company", ""))
pcb24.add_child(Node.ipv4("gip", "127.0.0.1"))
pcb24.add_child(Node.u16("gp", 10011))
pcb24.add_child(Node.string("rom_number", "M39-JB-G01"))
pcb24.add_child(Node.u64("c_drive", 10028228608))
pcb24.add_child(Node.u64("d_drive", 47945170944))
pcb24.add_child(Node.u64("e_drive", 10394677248))
pcb24.add_child(Node.string("etc", ""))
# Swap with server
resp = self.exchange("", call)
# Verify that response is correct
self.assert_path(resp, "response/pcb24/@status")
def __verify_common(self, root: str, resp: Node) -> None:
self.assert_path(resp, f"response/{root}/phase/event_id")
self.assert_path(resp, f"response/{root}/phase/phase")
# Area stuff is not needed unless enabling events.
# self.assert_path(resp, f"response/{root}/area/area_id")
# self.assert_path(resp, f"response/{root}/area/end_date")
# self.assert_path(resp, f"response/{root}/area/medal_id")
# self.assert_path(resp, f"response/{root}/area/is_limit")
self.assert_path(resp, f"response/{root}/choco/choco_id")
self.assert_path(resp, f"response/{root}/choco/param")
self.assert_path(resp, f"response/{root}/goods/item_id")
self.assert_path(resp, f"response/{root}/goods/item_type")
self.assert_path(resp, f"response/{root}/goods/price")
self.assert_path(resp, f"response/{root}/goods/goods_type")
def verify_info24_common(self, loc: str) -> None:
call = self.call_node()
# Construct node
info24 = Node.void("info24")
call.add_child(info24)
info24.set_attribute("loc_id", loc)
info24.set_attribute("method", "common")
# Swap with server
resp = self.exchange("", call)
# Verify that response is correct
self.__verify_common("info24", resp)
def verify_lobby24_getlist(self, loc: str) -> None:
call = self.call_node()
# Construct node
lobby24 = Node.void("lobby24")
call.add_child(lobby24)
lobby24.set_attribute("method", "getList")
lobby24.add_child(Node.string("location_id", loc))
lobby24.add_child(Node.u8("net_version", 63))
# Swap with server
resp = self.exchange("", call)
# Verify that response is correct
self.assert_path(resp, "response/lobby24/@status")
def __verify_profile(self, resp: Node) -> None:
self.assert_path(resp, "response/player24/account/name")
self.assert_path(resp, "response/player24/account/g_pm_id")
self.assert_path(resp, "response/player24/account/tutorial")
self.assert_path(resp, "response/player24/account/area_id")
self.assert_path(resp, "response/player24/account/use_navi")
self.assert_path(resp, "response/player24/account/read_news")
self.assert_path(resp, "response/player24/account/nice")
self.assert_path(resp, "response/player24/account/favorite_chara")
self.assert_path(resp, "response/player24/account/special_area")
self.assert_path(resp, "response/player24/account/chocolate_charalist")
self.assert_path(resp, "response/player24/account/chocolate_sp_chara")
self.assert_path(resp, "response/player24/account/chocolate_pass_cnt")
self.assert_path(resp, "response/player24/account/chocolate_hon_cnt")
self.assert_path(resp, "response/player24/account/teacher_setting")
self.assert_path(resp, "response/player24/account/welcom_pack")
self.assert_path(resp, "response/player24/account/ranking_node")
self.assert_path(resp, "response/player24/account/chara_ranking_kind_id")
self.assert_path(resp, "response/player24/account/navi_evolution_flg")
self.assert_path(resp, "response/player24/account/ranking_news_last_no")
self.assert_path(resp, "response/player24/account/power_point")
self.assert_path(resp, "response/player24/account/player_point")
self.assert_path(resp, "response/player24/account/power_point_list")
self.assert_path(resp, "response/player24/account/staff")
self.assert_path(resp, "response/player24/account/item_type")
self.assert_path(resp, "response/player24/account/item_id")
self.assert_path(resp, "response/player24/account/is_conv")
self.assert_path(resp, "response/player24/account/license_data")
self.assert_path(resp, "response/player24/account/my_best")
self.assert_path(resp, "response/player24/account/latest_music")
self.assert_path(resp, "response/player24/account/total_play_cnt")
self.assert_path(resp, "response/player24/account/today_play_cnt")
self.assert_path(resp, "response/player24/account/consecutive_days")
self.assert_path(resp, "response/player24/account/total_days")
self.assert_path(resp, "response/player24/account/interval_day")
self.assert_path(resp, "response/player24/account/active_fr_num")
self.assert_path(resp, "response/player24/eaappli/relation")
self.assert_path(resp, "response/player24/info/ep")
self.assert_path(resp, "response/player24/config")
self.assert_path(resp, "response/player24/option")
self.assert_path(resp, "response/player24/custom_cate")
self.assert_path(resp, "response/player24/navi_data")
self.assert_path(resp, "response/player24/mission/mission_id")
self.assert_path(resp, "response/player24/mission/gauge_point")
self.assert_path(resp, "response/player24/mission/mission_comp")
self.assert_path(resp, "response/player24/netvs")
self.assert_path(resp, "response/player24/customize")
self.assert_path(resp, "response/player24/stamp/stamp_id")
self.assert_path(resp, "response/player24/stamp/cnt")
def verify_player24_read(self, ref_id: str, msg_type: str) -> Dict[str, Dict[int, Dict[str, int]]]:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("method", "read")
player24.add_child(Node.string("ref_id", ref_id))
player24.add_child(Node.s8("pref", 51))
# Swap with server
resp = self.exchange("", call)
if msg_type == "new":
# Verify that response is correct
self.assert_path(resp, "response/player24/result")
status = resp.child_value("player24/result")
if status != 2:
raise Exception(f"Reference ID '{ref_id}' returned invalid status '{status}'")
return {
"items": {},
"characters": {},
"points": {},
}
elif msg_type == "query":
# Verify that the response is correct
self.__verify_profile(resp)
self.assert_path(resp, "response/player24/result")
status = resp.child_value("player24/result")
if status != 0:
raise Exception(f"Reference ID '{ref_id}' returned invalid status '{status}'")
name = resp.child_value("player24/account/name")
if name != self.NAME:
raise Exception(f"Invalid name '{name}' returned for Ref ID '{ref_id}'")
# Medals and items
items: Dict[int, Dict[str, int]] = {}
charas: Dict[int, Dict[str, int]] = {}
courses: Dict[int, Dict[str, int]] = {}
for obj in resp.child("player24").children:
if obj.name == "item":
items[obj.child_value("id")] = {
"type": obj.child_value("type"),
"param": obj.child_value("param"),
}
elif obj.name == "chara_param":
charas[obj.child_value("chara_id")] = {
"friendship": obj.child_value("friendship"),
}
elif obj.name == "course_data":
courses[obj.child_value("course_id")] = {
"clear_type": obj.child_value("clear_type"),
"clear_rank": obj.child_value("clear_rank"),
"total_score": obj.child_value("total_score"),
"count": obj.child_value("update_count"),
"sheet_num": obj.child_value("sheet_num"),
}
return {
"items": items,
"characters": charas,
"courses": courses,
"points": {0: {"points": resp.child_value("player24/account/player_point")}},
}
else:
raise Exception(f"Unrecognized message type '{msg_type}'")
def verify_player24_read_score(self, ref_id: str) -> Dict[str, Dict[int, Dict[int, int]]]:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("method", "read_score")
player24.add_child(Node.string("ref_id", ref_id))
player24.add_child(Node.s8("pref", 51))
# Swap with server
resp = self.exchange("", call)
# Verify defaults
self.assert_path(resp, "response/player24/@status")
# Grab scores
scores: Dict[int, Dict[int, int]] = {}
medals: Dict[int, Dict[int, int]] = {}
ranks: Dict[int, Dict[int, int]] = {}
for child in resp.child("player24").children:
if child.name != "music":
continue
musicid = child.child_value("music_num")
chart = child.child_value("sheet_num")
score = child.child_value("score")
medal = child.child_value("clear_type")
rank = child.child_value("clear_rank")
if musicid not in scores:
scores[musicid] = {}
if musicid not in medals:
medals[musicid] = {}
if musicid not in ranks:
ranks[musicid] = {}
scores[musicid][chart] = score
medals[musicid][chart] = medal
ranks[musicid][chart] = rank
return {
"scores": scores,
"medals": medals,
"ranks": ranks,
}
def verify_player24_start(self, ref_id: str, loc: str) -> None:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("loc_id", loc)
player24.set_attribute("ref_id", ref_id)
player24.set_attribute("method", "start")
player24.set_attribute("start_type", "0")
pcb_card = Node.void("pcb_card")
player24.add_child(pcb_card)
pcb_card.add_child(Node.s8("card_enable", 1))
pcb_card.add_child(Node.s8("card_soldout", 0))
# Swap with server
resp = self.exchange("", call)
# Verify that response is correct
self.__verify_common("player24", resp)
def verify_player24_update_ranking(self, ref_id: str, loc: str) -> None:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("method", "update_ranking")
player24.add_child(Node.s16("pref", 51))
player24.add_child(Node.string("location_id", loc))
player24.add_child(Node.string("ref_id", ref_id))
player24.add_child(Node.string("name", self.NAME))
player24.add_child(Node.s16("chara_num", 1))
player24.add_child(Node.s16("course_id", 12345))
player24.add_child(Node.s32("total_score", 86000))
player24.add_child(Node.s16("music_num", 1375))
player24.add_child(Node.u8("sheet_num", 2))
player24.add_child(Node.u8("clear_type", 7))
player24.add_child(Node.u8("clear_rank", 5))
# Swap with server
resp = self.exchange("", call)
# Verify that response is correct
self.assert_path(resp, "response/player24/all_ranking/name")
self.assert_path(resp, "response/player24/all_ranking/chara_num")
self.assert_path(resp, "response/player24/all_ranking/total_score")
self.assert_path(resp, "response/player24/all_ranking/clear_type")
self.assert_path(resp, "response/player24/all_ranking/clear_rank")
self.assert_path(resp, "response/player24/all_ranking/player_count")
self.assert_path(resp, "response/player24/all_ranking/player_rank")
def verify_player24_logout(self, ref_id: str) -> None:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("ref_id", ref_id)
player24.set_attribute("method", "logout")
# Swap with server
resp = self.exchange("", call)
# Verify that response is correct
self.assert_path(resp, "response/player24/@status")
def verify_player24_write(
self,
ref_id: str,
item: Optional[Dict[str, int]] = None,
character: Optional[Dict[str, int]] = None,
) -> None:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("method", "write")
player24.add_child(Node.string("ref_id", ref_id))
# Add required children
config = Node.void("config")
player24.add_child(config)
config.add_child(Node.s16("chara", 1543))
if item is not None:
itemnode = Node.void("item")
player24.add_child(itemnode)
itemnode.add_child(Node.u8("type", item["type"]))
itemnode.add_child(Node.u16("id", item["id"]))
itemnode.add_child(Node.u16("param", item["param"]))
itemnode.add_child(Node.bool("is_new", False))
itemnode.add_child(Node.u64("get_time", 0))
if character is not None:
chara_param = Node.void("chara_param")
player24.add_child(chara_param)
chara_param.add_child(Node.u16("chara_id", character["id"]))
chara_param.add_child(Node.u16("friendship", character["friendship"]))
# Swap with server
resp = self.exchange("", call)
self.assert_path(resp, "response/player24/@status")
def verify_player24_buy(self, ref_id: str, item: Dict[str, int]) -> None:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("method", "buy")
player24.add_child(Node.s32("play_id", 0))
player24.add_child(Node.string("ref_id", ref_id))
player24.add_child(Node.u16("id", item["id"]))
player24.add_child(Node.u8("type", item["type"]))
player24.add_child(Node.u16("param", item["param"]))
player24.add_child(Node.s32("lumina", item["points"]))
player24.add_child(Node.u16("price", item["price"]))
# Swap with server
resp = self.exchange("", call)
self.assert_path(resp, "response/player24/@status")
def verify_player24_write_music(self, ref_id: str, score: Dict[str, Any]) -> None:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("method", "write_music")
player24.add_child(Node.string("ref_id", ref_id))
player24.add_child(Node.string("data_id", ref_id))
player24.add_child(Node.string("name", self.NAME))
player24.add_child(Node.u8("stage", 0))
player24.add_child(Node.s16("music_num", score["id"]))
player24.add_child(Node.u8("sheet_num", score["chart"]))
player24.add_child(Node.u8("clear_type", score["medal"]))
player24.add_child(Node.s32("score", score["score"]))
player24.add_child(Node.s16("combo", 0))
player24.add_child(Node.s16("cool", 0))
player24.add_child(Node.s16("great", 0))
player24.add_child(Node.s16("good", 0))
player24.add_child(Node.s16("bad", 0))
# Swap with server
resp = self.exchange("", call)
self.assert_path(resp, "response/player24/@status")
def verify_player24_new(self, ref_id: str) -> None:
call = self.call_node()
# Construct node
player24 = Node.void("player24")
call.add_child(player24)
player24.set_attribute("method", "new")
player24.add_child(Node.string("ref_id", ref_id))
player24.add_child(Node.string("name", self.NAME))
player24.add_child(Node.s8("pref", 51))
# Swap with server
resp = self.exchange("", call)
# Verify nodes
self.__verify_profile(resp)
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()
location = self.verify_facility_get()
self.verify_pcbevent_put()
self.verify_pcb24_boot(location)
self.verify_info24_common(location)
self.verify_lobby24_getlist(location)
# 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_player24_read(ref_id, msg_type="new")
self.verify_player24_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")
# Verify proper handling of basic stuff
self.verify_player24_read(ref_id, msg_type="query")
self.verify_player24_start(ref_id, location)
self.verify_player24_write(ref_id)
self.verify_player24_logout(ref_id)
if cardid is None:
# Verify unlocks/story mode work
unlocks = self.verify_player24_read(ref_id, msg_type="query")
for item in unlocks["items"]:
if item in [1592, 1608]:
# Song unlocks after one play
continue
raise Exception("Got nonzero items count on a new card!")
for _ in unlocks["characters"]:
raise Exception("Got nonzero characters count on a new card!")
for _ in unlocks["courses"]:
raise Exception("Got nonzero course count on a new card!")
if unlocks["points"][0]["points"] != 300:
raise Exception("Got wrong default value for points on a new card!")
self.verify_player24_write(ref_id, item={"id": 4, "type": 2, "param": 69})
unlocks = self.verify_player24_read(ref_id, msg_type="query")
if 4 not in unlocks["items"]:
raise Exception("Expecting to see item ID 4 in items!")
if unlocks["items"][4]["type"] != 2:
raise Exception("Expecting to see item ID 4 to have type 2 in items!")
if unlocks["items"][4]["param"] != 69:
raise Exception("Expecting to see item ID 4 to have param 69 in items!")
self.verify_player24_write(ref_id, character={"id": 5, "friendship": 420})
unlocks = self.verify_player24_read(ref_id, msg_type="query")
if 5 not in unlocks["characters"]:
raise Exception("Expecting to see chara ID 5 in characters!")
if unlocks["characters"][5]["friendship"] != 420:
raise Exception("Expecting to see chara ID 5 to have type 2 in characters!")
# Verify purchases work
self.verify_player24_buy(
ref_id,
item={"id": 6, "type": 3, "param": 8, "points": 400, "price": 250},
)
unlocks = self.verify_player24_read(ref_id, msg_type="query")
if 6 not in unlocks["items"]:
raise Exception("Expecting to see item ID 6 in items!")
if unlocks["items"][6]["type"] != 3:
raise Exception("Expecting to see item ID 6 to have type 3 in items!")
if unlocks["items"][6]["param"] != 8:
raise Exception("Expecting to see item ID 6 to have param 8 in items!")
if unlocks["points"][0]["points"] != 150:
raise Exception(f'Got wrong value for points {unlocks["points"][0]["points"]} after purchase!')
# Verify course handling
self.verify_player24_update_ranking(ref_id, location)
unlocks = self.verify_player24_read(ref_id, msg_type="query")
if 12345 not in unlocks["courses"]:
raise Exception("Expecting to see course ID 12345 in courses!")
if unlocks["courses"][12345]["clear_type"] != 7:
raise Exception("Expecting to see item ID 12345 to have clear_type 7 in courses!")
if unlocks["courses"][12345]["clear_rank"] != 5:
raise Exception("Expecting to see item ID 12345 to have clear_rank 5 in courses!")
if unlocks["courses"][12345]["total_score"] != 86000:
raise Exception("Expecting to see item ID 12345 to have total_score 86000 in courses!")
if unlocks["courses"][12345]["count"] != 1:
raise Exception("Expecting to see item ID 12345 to have count 1 in courses!")
if unlocks["courses"][12345]["sheet_num"] != 2:
raise Exception("Expecting to see item ID 12345 to have sheet_num 2 in courses!")
# Verify score handling
scores = self.verify_player24_read_score(ref_id)
for _ in scores["medals"]:
raise Exception("Got nonzero medals count on a new card!")
for _ in scores["scores"]:
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(920, 950)
chartid = random.randint(0, 3)
score = random.randint(0, 100000)
medal = random.randint(1, 11)
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": 6,
"score": 98765,
},
# A worse score on another same chart
{
"id": 987,
"chart": 0,
"medal": 3,
"score": 12345,
"expected_score": 99999,
"expected_medal": 6,
},
]
for dummyscore in dummyscores:
self.verify_player24_write_music(ref_id, dummyscore)
scores = self.verify_player24_read_score(ref_id)
for expected in dummyscores:
newscore = scores["scores"][expected["id"]][expected["chart"]]
newmedal = scores["medals"][expected["id"]][expected["chart"]]
newrank = scores["ranks"][expected["id"]][expected["chart"]]
if "expected_score" in expected:
expected_score = expected["expected_score"]
else:
expected_score = expected["score"]
if "expected_medal" in expected:
expected_medal = expected["expected_medal"]
else:
expected_medal = expected["medal"]
if newscore < 50000:
expected_rank = 1
elif newscore < 62000:
expected_rank = 2
elif newscore < 72000:
expected_rank = 3
elif newscore < 82000:
expected_rank = 4
elif newscore < 90000:
expected_rank = 5
elif newscore < 95000:
expected_rank = 6
elif newscore < 98000:
expected_rank = 7
else:
expected_rank = 8
if newscore != expected_score:
raise Exception(
f'Expected a score of \'{expected_score}\' for song \'{expected["id"]}\' chart \'{expected["chart"]}\' but got score \'{newscore}\''
)
if newmedal != expected_medal:
raise Exception(
f'Expected a medal of \'{expected_medal}\' for song \'{expected["id"]}\' chart \'{expected["chart"]}\' but got medal \'{newmedal}\''
)
if newrank != expected_rank:
raise Exception(
f'Expected a rank of \'{expected_rank}\' for song \'{expected["id"]}\' chart \'{expected["chart"]}\' but got rank \'{newrank}\''
)
# 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)

View File

@ -21,6 +21,7 @@ class CardCipher:
INTERNAL_CIPHER = DES3.new(DES_KEY, DES3.MODE_ECB)
VALID_CHARS: Final[str] = "0123456789ABCDEFGHJKLMNPRSTUWXYZ"
REVERSE_CHARS: Final[Dict[str, int]] = {char: off for off, char in enumerate("0123456789ABCDEFGHJKLMNPRSTUWXYZ")}
CONV_CHARS: Final[Dict[str, str]] = {
"I": "1",
"O": "0",
@ -28,9 +29,9 @@ class CardCipher:
@staticmethod
def __type_from_cardid(cardid: str) -> int:
if cardid[:2].upper() == "E0":
if cardid[:4].upper() == "E004":
return 1
if cardid[:2].upper() == "01":
if cardid[:1] == "0":
return 2
raise CardCipherException("Unrecognized card type")
@ -123,10 +124,7 @@ class CardCipher:
groups = [0] * 16
for i in range(0, 16):
for j in range(0, 32):
if cardid[i] == CardCipher.VALID_CHARS[j]:
groups[i] = j
break
groups[i] = CardCipher.REVERSE_CHARS[cardid[i]]
# Verify scheme and checksum
if groups[14] != 1 and groups[14] != 2:

View File

@ -125,6 +125,7 @@ class VersionConstants:
POPN_MUSIC_USANEKO: Final[int] = 24
POPN_MUSIC_PEACE: Final[int] = 25
POPN_MUSIC_KAIMEI_RIDDLES: Final[int] = 26
POPN_MUSIC_UNILAB: Final[int] = 27
REFLEC_BEAT: Final[int] = 1
REFLEC_BEAT_LIMELIGHT: Final[int] = 2

View File

@ -85,6 +85,10 @@ class PEFile:
def __init__(self, data: bytes) -> None:
self.data = data
self.__pe = pefile.PE(data=data, fast_load=True)
# Mapping of ad-hoc virtual addresses, which get added to during runtime. For the purpose
# of our emulation, we just tack values to the end of the physical binary and add an ad-hoc
# mapping. The mapping is indexed by virtual address and points to a physical binary offset.
self.__adhoc_mapping: Dict[int, int] = {}
def virtual_to_physical(self, offset: int) -> int:
@ -95,9 +99,8 @@ class PEFile:
if offset >= start and offset < end:
return (offset - start) + section.PointerToRawData
for virtual, physical in self.__adhoc_mapping.items():
if offset == virtual:
return physical
if offset in self.__adhoc_mapping:
return self.__adhoc_mapping[offset]
raise InvalidVirtualOffsetException(f"Couldn't find physical offset for virtual offset 0x{offset:08x}")
@ -360,7 +363,7 @@ class PEFile:
vprint(f"pop {dest}")
size = get_size(src)
size = get_size(dest)
if size is None:
raise Exception(f"Could not determine size of {mnemonic} operation!")
result = fetch(registers, memory, size, "[rsp]" if self.is_64bit() else "[esp]")

View File

@ -54,15 +54,20 @@ class APIClient:
def __repr__(self) -> str:
# Specifically defined so that two different instances of the same API client
# cache under the same key, as we want to share results from a given server
# to all local requests.
return (
# to all local requests. We also have to be sensitive to any control character
# limitations for memcached.
repr_val = (
"APIClient("
+ f"base_uri={self.base_uri!r}, "
+ f"token={self.token!r}, "
+ f"allow_stats={self.allow_stats!r}, "
+ f"base_uri={self.base_uri!r},"
+ f"token={self.token!r},"
+ f"allow_stats={self.allow_stats!r},"
+ f"allow_scores={self.allow_scores!r}"
+ ")"
)
repr_val = repr_val.replace(" ", "_")
repr_val = repr_val.replace("\r", "_")
repr_val = repr_val.replace("\n", "_")
return repr_val
def _content_type_valid(self, content_type: str) -> bool:
if ";" in content_type:
@ -195,6 +200,7 @@ class APIClient:
VersionConstants.POPN_MUSIC_USANEKO: "24",
VersionConstants.POPN_MUSIC_PEACE: "25",
VersionConstants.POPN_MUSIC_KAIMEI_RIDDLES: "26",
VersionConstants.POPN_MUSIC_UNILAB: "27",
},
GameConstants.REFLEC_BEAT: {
VersionConstants.REFLEC_BEAT: "1",

View File

@ -335,12 +335,12 @@ class GlobalMusicData(BaseGlobalData):
{
"rank": self.__max(oldscore.data["rank"], newscore.data["rank"]),
"halo": self.__max(oldscore.data["halo"], newscore.data["halo"]),
"ghost": oldscore.data.get("ghost")
if oldscore.points > newscore.points
else newscore.data.get("ghost"),
"trace": oldscore.data.get("trace")
if oldscore.points > newscore.points
else newscore.data.get("trace"),
"ghost": (
oldscore.data.get("ghost") if oldscore.points > newscore.points else newscore.data.get("ghost")
),
"trace": (
oldscore.data.get("trace") if oldscore.points > newscore.points else newscore.data.get("trace")
),
"combo": self.__max(oldscore.data["combo"], newscore.data["combo"]),
},
)
@ -360,19 +360,23 @@ class GlobalMusicData(BaseGlobalData):
oldscore.plays + newscore.plays,
{
"clear_status": self.__max(oldscore.data["clear_status"], newscore.data["clear_status"]),
"ghost": oldscore.data.get("ghost")
if oldscore.points > newscore.points
else newscore.data.get("ghost"),
"ghost": (
oldscore.data.get("ghost") if oldscore.points > newscore.points else newscore.data.get("ghost")
),
"miss_count": self.__min(
oldscore.data.get_int("miss_count", -1),
newscore.data.get_int("miss_count", -1),
),
"pgreats": oldscore.data.get_int("pgreats", -1)
if oldscore.points > newscore.points
else newscore.data.get_int("pgreats", -1),
"greats": oldscore.data.get_int("greats", -1)
if oldscore.points > newscore.points
else newscore.data.get_int("greats", -1),
"pgreats": (
oldscore.data.get_int("pgreats", -1)
if oldscore.points > newscore.points
else newscore.data.get_int("pgreats", -1)
),
"greats": (
oldscore.data.get_int("greats", -1)
if oldscore.points > newscore.points
else newscore.data.get_int("greats", -1)
),
},
)
@ -392,9 +396,9 @@ class GlobalMusicData(BaseGlobalData):
oldscore.location, # Always propagate location from local setup if possible
oldscore.plays + newscore.plays,
{
"ghost": oldscore.data.get("ghost")
if oldscore.points > newscore.points
else newscore.data.get("ghost"),
"ghost": (
oldscore.data.get("ghost") if oldscore.points > newscore.points else newscore.data.get("ghost")
),
"combo": self.__max(oldscore.data["combo"], newscore.data["combo"]),
"medal": self.__max(oldscore.data["medal"], newscore.data["medal"]),
# Conditionally include this if we have any info for it.

View File

@ -1,6 +1,6 @@
import copy
import os
from sqlalchemy.engine import Engine # type: ignore
from sqlalchemy.engine import Engine
from typing import Any, Dict, Optional, Set
from bemani.common import GameConstants, RegionConstants

View File

@ -2,13 +2,13 @@ import os
import alembic.config
from alembic.migration import MigrationContext
from alembic.autogenerate import compare_metadata # type: ignore
from sqlalchemy import create_engine # type: ignore
from sqlalchemy.orm import scoped_session # type: ignore
from alembic.autogenerate import compare_metadata
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.engine import Engine # type: ignore
from sqlalchemy.sql import text # type: ignore
from sqlalchemy.exc import ProgrammingError # type: ignore
from sqlalchemy.engine import Engine
from sqlalchemy.sql import text
from sqlalchemy.exc import ProgrammingError
from bemani.data.api.user import GlobalUserData
from bemani.data.api.game import GlobalGameData
@ -97,7 +97,6 @@ class Data:
session_factory = sessionmaker(
bind=config.database.engine,
autoflush=True,
autocommit=True,
)
self.__config = config
self.__session = scoped_session(session_factory)
@ -136,7 +135,7 @@ class Data:
# See if the DB was already created
try:
cursor = self.__session.execute(text("SELECT COUNT(version_num) AS count FROM alembic_version"))
return cursor.fetchone()["count"] == 1
return cursor.mappings().fetchone()["count"] == 1
except ProgrammingError:
return False
@ -153,7 +152,7 @@ class Data:
]
alembicArgs.extend(args)
os.chdir(base_dir)
alembic.config.main(argv=alembicArgs) # type: ignore
alembic.config.main(argv=alembicArgs)
def create(self) -> None:
"""

View File

@ -1,6 +1,6 @@
import uuid
from sqlalchemy import Table, Column # type: ignore
from sqlalchemy.types import String, Integer # type: ignore
from sqlalchemy import Table, Column
from sqlalchemy.types import String, Integer
from typing import Any, Dict, List, Optional
from bemani.common import Time
@ -55,7 +55,7 @@ class APIData(APIProviderInterface, BaseData):
result["name"],
result["token"],
)
for result in cursor
for result in cursor.mappings()
]
def validate_client(self, token: str) -> bool:
@ -70,7 +70,7 @@ class APIData(APIProviderInterface, BaseData):
"""
sql = "SELECT count(*) AS count FROM client WHERE token = :token"
cursor = self.execute(sql, {"token": token})
return cursor.fetchone()["count"] == 1
return cursor.mappings().fetchone()["count"] == 1 # type: ignore
def create_client(self, name: str) -> int:
"""
@ -109,7 +109,7 @@ class APIData(APIProviderInterface, BaseData):
# Couldn't find an entry with this ID
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return Client(
clientid,
result["timestamp"],
@ -159,7 +159,7 @@ class APIData(APIProviderInterface, BaseData):
sql = "SELECT id, timestamp, uri, token, config FROM server ORDER BY timestamp ASC"
cursor = self.execute(sql)
return [format_result(result) for result in cursor]
return [format_result(result) for result in cursor.mappings()]
def create_server(self, uri: str, token: str) -> int:
"""
@ -199,7 +199,7 @@ class APIData(APIProviderInterface, BaseData):
# Couldn't find an entry with this ID
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
allow_stats = (result["config"] & 0x1) == 0
allow_scores = (result["config"] & 0x2) == 0
return Server(

View File

@ -6,11 +6,11 @@ from typing_extensions import Final
from bemani.common import Time
from bemani.data.config import Config
from sqlalchemy.engine.base import Connection # type: ignore
from sqlalchemy.engine import CursorResult # type: ignore
from sqlalchemy.sql import text # type: ignore
from sqlalchemy.types import String, Integer # type: ignore
from sqlalchemy import Table, Column, MetaData # type: ignore
from sqlalchemy.engine import CursorResult
from sqlalchemy.orm import scoped_session
from sqlalchemy.sql import text
from sqlalchemy.types import String, Integer
from sqlalchemy import Table, Column, MetaData
metadata = MetaData()
@ -40,7 +40,7 @@ class _BytesEncoder(json.JSONEncoder):
class BaseData:
SESSION_LENGTH: Final[int] = 32
def __init__(self, config: Config, conn: Connection) -> None:
def __init__(self, config: Config, conn: scoped_session) -> None:
"""
Initialize any DB singleton.
@ -82,10 +82,12 @@ class BaseData:
includes = all(s in lowered for s in write_statement_group)
if includes and not safe_write_operation:
raise Exception("Read-only mode is active!")
return self.__conn.execute(
result = self.__conn.execute(
text(sql),
params if params is not None else {},
)
self.__conn.commit()
return result
def serialize(self, data: Dict[str, Any]) -> str:
"""
@ -141,7 +143,7 @@ class BaseData:
# Couldn't find a user with this session
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return result["id"]
def _create_session(self, opid: int, optype: str, expiration: int = (30 * 86400)) -> str:

View File

@ -1,6 +1,6 @@
from sqlalchemy import Table, Column, UniqueConstraint # type: ignore
from sqlalchemy.types import String, Integer, JSON # type: ignore
from sqlalchemy.dialects.mysql import BIGINT as BigInteger # type: ignore
from sqlalchemy import Table, Column, UniqueConstraint
from sqlalchemy.types import String, Integer, JSON
from sqlalchemy.dialects.mysql import BIGINT as BigInteger
from typing import Any, Dict, List, Optional
from bemani.common import GameConstants, ValidatedDict, Time
@ -96,7 +96,7 @@ class GameData(BaseData):
# Settings doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return ValidatedDict(self.deserialize(result["data"]))
def put_settings(self, game: GameConstants, userid: UserID, settings: Dict[str, Any]) -> None:
@ -158,7 +158,7 @@ class GameData(BaseData):
# score doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return ValidatedDict(self.deserialize(result["data"]))
def get_achievements(self, game: GameConstants, userid: UserID) -> List[Achievement]:
@ -182,7 +182,7 @@ class GameData(BaseData):
None,
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]
def put_achievement(
@ -251,7 +251,7 @@ class GameData(BaseData):
# setting doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
retval = ValidatedDict(self.deserialize(result["data"]))
retval["start_time"] = result["start_time"]
retval["end_time"] = result["end_time"]
@ -288,7 +288,7 @@ class GameData(BaseData):
"end_time": result["end_time"],
}
)
for result in cursor
for result in cursor.mappings()
]
def put_time_sensitive_settings(
@ -336,7 +336,7 @@ class GameData(BaseData):
"end_time": end_time,
},
)
for result in cursor:
for result in cursor.mappings():
if result["start_time"] == start_time and result["end_time"] == end_time:
# This is just this event being updated, that's fine.
continue
@ -387,7 +387,7 @@ class GameData(BaseData):
# entry doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return ValidatedDict(self.deserialize(result["data"]))
def get_items(self, game: GameConstants, version: int) -> List[Item]:
@ -410,5 +410,5 @@ class GameData(BaseData):
result["id"],
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]

View File

@ -1,8 +1,8 @@
import copy
from sqlalchemy import Table, Column, UniqueConstraint # type: ignore
from sqlalchemy.types import String, Integer, JSON # type: ignore
from sqlalchemy.dialects.mysql import BIGINT as BigInteger # type: ignore
from sqlalchemy import Table, Column, UniqueConstraint
from sqlalchemy.types import String, Integer, JSON
from sqlalchemy.dialects.mysql import BIGINT as BigInteger
from typing import Optional, Dict, List, Tuple, Any
from bemani.common import GameConstants, ValidatedDict, Time
@ -82,7 +82,7 @@ class LobbyData(BaseData):
# Settings doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
data = ValidatedDict(self.deserialize(result["data"]))
data["id"] = result["id"]
data["time"] = result["time"]
@ -119,7 +119,7 @@ class LobbyData(BaseData):
data["time"] = result["time"]
return data
return [(UserID(result["userid"]), format_result(result)) for result in cursor]
return [(UserID(result["userid"]), format_result(result)) for result in cursor.mappings()]
def put_play_session_info(self, game: GameConstants, version: int, userid: UserID, data: Dict[str, Any]) -> None:
"""
@ -214,7 +214,7 @@ class LobbyData(BaseData):
# Settings doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
data = ValidatedDict(self.deserialize(result["data"]))
data["id"] = result["id"]
data["time"] = result["time"]
@ -252,7 +252,7 @@ class LobbyData(BaseData):
data["time"] = result["time"]
return data
return [(UserID(result["userid"]), format_result(result)) for result in cursor]
return [(UserID(result["userid"]), format_result(result)) for result in cursor.mappings()]
def put_lobby(self, game: GameConstants, version: int, userid: UserID, data: Dict[str, Any]) -> None:
"""

View File

@ -1,6 +1,6 @@
from sqlalchemy import Table, Column, UniqueConstraint # type: ignore
from sqlalchemy.types import String, Integer, JSON # type: ignore
from sqlalchemy.dialects.mysql import BIGINT as BigInteger # type: ignore
from sqlalchemy import Table, Column, UniqueConstraint
from sqlalchemy.types import String, Integer, JSON
from sqlalchemy.dialects.mysql import BIGINT as BigInteger
from typing import Optional, Dict, List, Tuple, Any
from typing_extensions import Final
@ -102,7 +102,7 @@ class MachineData(BaseData):
# Machine doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return result["pcbid"]
def from_machine_id(self, machine_id: int) -> Optional[str]:
@ -122,7 +122,7 @@ class MachineData(BaseData):
# Machine doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return result["pcbid"]
def from_userid(self, userid: UserID) -> List[ArcadeID]:
@ -137,7 +137,7 @@ class MachineData(BaseData):
"""
sql = "SELECT arcadeid FROM arcade_owner WHERE userid = :userid"
cursor = self.execute(sql, {"userid": userid})
return [ArcadeID(result["arcadeid"]) for result in cursor]
return [ArcadeID(result["arcadeid"]) for result in cursor.mappings()]
def from_session(self, session: str) -> Optional[ArcadeID]:
"""
@ -173,7 +173,7 @@ class MachineData(BaseData):
# Machine doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return Machine(
result["id"],
pcbid,
@ -212,7 +212,7 @@ class MachineData(BaseData):
result["version"],
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]
def put_machine(self, machine: Machine) -> None:
@ -279,7 +279,7 @@ class MachineData(BaseData):
port = None
else:
# Grab highest port
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
port = result["port"]
if port is not None:
port = port + 1
@ -383,7 +383,7 @@ class MachineData(BaseData):
# Arcade doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
sql = "SELECT userid FROM arcade_owner WHERE arcadeid = :id"
cursor = self.execute(sql, {"id": arcadeid})
@ -396,7 +396,7 @@ class MachineData(BaseData):
result["pref"],
result["area"] or None,
self.deserialize(result["data"]),
[owner["userid"] for owner in cursor],
[owner["userid"] for owner in cursor.mappings()],
)
def put_arcade(self, arcade: Arcade) -> None:
@ -464,7 +464,7 @@ class MachineData(BaseData):
sql = "SELECT userid, arcadeid FROM arcade_owner"
cursor = self.execute(sql)
arcade_to_owners: Dict[int, List[UserID]] = {}
for row in cursor:
for row in cursor.mappings():
arcade = row["arcadeid"]
owner = UserID(row["userid"])
if arcade not in arcade_to_owners:
@ -484,7 +484,7 @@ class MachineData(BaseData):
self.deserialize(result["data"]),
arcade_to_owners.get(result["id"], []),
)
for result in cursor
for result in cursor.mappings()
]
def get_settings(
@ -512,7 +512,7 @@ class MachineData(BaseData):
# Settings doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return ValidatedDict(self.deserialize(result["data"]))
def put_settings(
@ -566,7 +566,7 @@ class MachineData(BaseData):
UserID(entry["userid"]),
entry["balance"],
)
for entry in cursor
for entry in cursor.mappings()
]
def create_session(self, arcadeid: ArcadeID, expiration: int = (30 * 86400)) -> str:

View File

@ -1,7 +1,7 @@
from sqlalchemy import Table, Column, UniqueConstraint # type: ignore
from sqlalchemy.exc import IntegrityError # type: ignore
from sqlalchemy.types import String, Integer, JSON # type: ignore
from sqlalchemy.dialects.mysql import BIGINT as BigInteger # type: ignore
from sqlalchemy import Table, Column, UniqueConstraint
from sqlalchemy.exc import IntegrityError
from sqlalchemy.types import String, Integer, JSON
from sqlalchemy.dialects.mysql import BIGINT as BigInteger
from typing import Optional, Dict, List, Tuple, Any
from bemani.common import GameConstants, Time, VersionConstants
@ -145,7 +145,7 @@ class MusicData(BaseData):
# music doesn't exist
raise Exception(f"Song {songid} chart {songchart} doesn't exist for game {game} version {version}")
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return result["id"]
def put_score(
@ -334,7 +334,7 @@ class MusicData(BaseData):
# score doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return Score(
result["scorekey"],
result["songid"],
@ -394,7 +394,7 @@ class MusicData(BaseData):
# score doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return (
UserID(result["userid"]),
Score(
@ -477,7 +477,7 @@ class MusicData(BaseData):
result["plays"],
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]
def get_most_played(self, game: GameConstants, version: int, userid: UserID, count: int) -> List[Tuple[int, int]]:
@ -510,7 +510,7 @@ class MusicData(BaseData):
{"userid": userid, "game": game.value, "version": version, "count": count},
)
return [(result["songid"], result["plays"]) for result in cursor]
return [(result["songid"], result["plays"]) for result in cursor.mappings()]
def get_last_played(self, game: GameConstants, version: int, userid: UserID, count: int) -> List[Tuple[int, int]]:
"""
@ -542,7 +542,7 @@ class MusicData(BaseData):
{"userid": userid, "game": game.value, "version": version, "count": count},
)
return [(result["songid"], result["timestamp"]) for result in cursor]
return [(result["songid"], result["timestamp"]) for result in cursor.mappings()]
def get_hit_chart(
self,
@ -589,7 +589,7 @@ class MusicData(BaseData):
},
)
return [(result["songid"], result["plays"]) for result in cursor]
return [(result["songid"], result["plays"]) for result in cursor.mappings()]
def get_song(
self,
@ -635,7 +635,7 @@ class MusicData(BaseData):
if cursor.rowcount != 1:
# music doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return Song(
game,
version,
@ -685,7 +685,7 @@ class MusicData(BaseData):
result["genre"],
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]
def get_all_scores(
@ -790,7 +790,7 @@ class MusicData(BaseData):
self.deserialize(result["data"]),
),
)
for result in cursor
for result in cursor.mappings()
]
def get_all_records(
@ -901,7 +901,7 @@ class MusicData(BaseData):
self.deserialize(result["data"]),
),
)
for result in cursor
for result in cursor.mappings()
]
def get_attempt_by_key(self, game: GameConstants, version: int, key: int) -> Optional[Tuple[UserID, Attempt]]:
@ -946,7 +946,7 @@ class MusicData(BaseData):
# score doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return (
UserID(result["userid"]),
Attempt(
@ -1057,5 +1057,5 @@ class MusicData(BaseData):
self.deserialize(result["data"]),
),
)
for result in cursor
for result in cursor.mappings()
]

View File

@ -1,6 +1,6 @@
from sqlalchemy import Table, Column, UniqueConstraint # type: ignore
from sqlalchemy.types import String, Integer, Text, JSON # type: ignore
from sqlalchemy.dialects.mysql import BIGINT as BigInteger # type: ignore
from sqlalchemy import Table, Column, UniqueConstraint
from sqlalchemy.types import String, Integer, Text, JSON
from sqlalchemy.dialects.mysql import BIGINT as BigInteger
from typing import Optional, Dict, List, Tuple, Any
from bemani.common import GameConstants, Time
@ -73,7 +73,7 @@ class NetworkData(BaseData):
result["title"],
result["body"],
)
for result in cursor
for result in cursor.mappings()
]
def create_news(self, title: str, body: str) -> int:
@ -107,7 +107,7 @@ class NetworkData(BaseData):
# Couldn't find an entry with this ID
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return News(
newsid,
result["timestamp"],
@ -181,7 +181,7 @@ class NetworkData(BaseData):
# No scheduled work was registered, so time to get going!
return True
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
if schedule == "daily":
# Just look at the day and year, make sure it matches
@ -317,7 +317,7 @@ class NetworkData(BaseData):
result["type"],
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]
def delete_events(self, oldest_event_ts: int) -> None:

View File

@ -1,8 +1,8 @@
import random
from sqlalchemy import Table, Column, UniqueConstraint # type: ignore
from sqlalchemy.types import String, Integer, JSON # type: ignore
from sqlalchemy.dialects.mysql import BIGINT as BigInteger # type: ignore
from sqlalchemy.exc import IntegrityError # type: ignore
from sqlalchemy import Table, Column, UniqueConstraint
from sqlalchemy.types import String, Integer, JSON
from sqlalchemy.dialects.mysql import BIGINT as BigInteger
from sqlalchemy.exc import IntegrityError
from typing import Optional, Dict, List, Tuple, Any
from typing_extensions import Final
from passlib.hash import pbkdf2_sha512 # type: ignore
@ -194,7 +194,7 @@ class UserData(BaseData):
# Couldn't find a user with this card
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return UserID(result["userid"])
def from_username(self, username: str) -> Optional[UserID]:
@ -213,7 +213,7 @@ class UserData(BaseData):
# Couldn't find this username
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return UserID(result["id"])
def from_refid(self, game: GameConstants, version: int, refid: str) -> Optional[UserID]:
@ -238,7 +238,7 @@ class UserData(BaseData):
# Couldn't find a user with this refid
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return UserID(result["userid"])
def from_extid(self, game: GameConstants, version: int, extid: int) -> Optional[UserID]:
@ -263,7 +263,7 @@ class UserData(BaseData):
# Couldn't find a user with this refid
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return UserID(result["userid"])
def from_session(self, session: str) -> Optional[UserID]:
@ -297,7 +297,7 @@ class UserData(BaseData):
# User doesn't exist, but we have a reference?
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return User(userid, result["username"], result["email"], result["admin"] == 1)
def get_all_users(self) -> List[User]:
@ -316,7 +316,7 @@ class UserData(BaseData):
result["email"],
result["admin"] == 1,
)
for result in cursor
for result in cursor.mappings()
]
def get_all_usernames(self) -> List[str]:
@ -331,7 +331,7 @@ class UserData(BaseData):
"""
sql = "SELECT username FROM user WHERE username is not null"
cursor = self.execute(sql)
return [res["username"] for res in cursor]
return [res["username"] for res in cursor.mappings()]
def get_all_cards(self) -> List[Tuple[str, UserID]]:
"""
@ -342,7 +342,7 @@ class UserData(BaseData):
"""
sql = "SELECT id, userid FROM card"
cursor = self.execute(sql)
return [(str(res["id"]).upper(), UserID(res["userid"])) for res in cursor]
return [(str(res["id"]).upper(), UserID(res["userid"])) for res in cursor.mappings()]
def get_cards(self, userid: UserID) -> List[str]:
"""
@ -356,7 +356,7 @@ class UserData(BaseData):
"""
sql = "SELECT id FROM card WHERE userid = :userid"
cursor = self.execute(sql, {"userid": userid})
return [str(res["id"]).upper() for res in cursor]
return [str(res["id"]).upper() for res in cursor.mappings()]
def add_card(self, userid: UserID, cardid: str) -> None:
"""
@ -437,7 +437,7 @@ class UserData(BaseData):
# User doesn't exist, but we have a reference?
return False
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return pin == result["pin"]
def update_pin(self, userid: UserID, pin: str) -> None:
@ -468,7 +468,7 @@ class UserData(BaseData):
# User doesn't exist, but we have a reference?
return False
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
passhash = result["password"]
try:
@ -517,7 +517,7 @@ class UserData(BaseData):
# Profile doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return Profile(
game,
version,
@ -577,7 +577,7 @@ class UserData(BaseData):
cursor = self.execute(sql, {"game": game.value, "userids": userids})
profilever: Dict[UserID, int] = {}
for result in cursor:
for result in cursor.mappings():
tuid = UserID(result["userid"])
tver = result["version"]
@ -626,7 +626,7 @@ class UserData(BaseData):
vals["game"] = game.value
cursor = self.execute(sql, vals)
return [(GameConstants(result["game"]), result["version"]) for result in cursor]
return [(GameConstants(result["game"]), result["version"]) for result in cursor.mappings()]
def get_all_profiles(self, game: GameConstants, version: int) -> List[Tuple[UserID, Profile]]:
"""
@ -662,7 +662,7 @@ class UserData(BaseData):
self.deserialize(result["data"]),
),
)
for result in cursor
for result in cursor.mappings()
]
def get_all_players(self, game: GameConstants, version: int) -> List[UserID]:
@ -682,7 +682,7 @@ class UserData(BaseData):
"""
cursor = self.execute(sql, {"game": game.value, "version": version})
return [UserID(result["userid"]) for result in cursor]
return [UserID(result["userid"]) for result in cursor.mappings()]
def get_all_achievements(
self,
@ -732,7 +732,7 @@ class UserData(BaseData):
self.deserialize(result["data"]),
),
)
for result in cursor
for result in cursor.mappings()
]
def put_profile(self, game: GameConstants, version: int, userid: UserID, profile: Profile) -> None:
@ -808,7 +808,7 @@ class UserData(BaseData):
# score doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return ValidatedDict(self.deserialize(result["data"]))
def get_achievements(self, game: GameConstants, version: int, userid: UserID) -> List[Achievement]:
@ -834,7 +834,7 @@ class UserData(BaseData):
None,
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]
def put_achievement(
@ -945,7 +945,7 @@ class UserData(BaseData):
result["timestamp"],
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]
def put_time_based_achievement(
@ -1023,7 +1023,7 @@ class UserData(BaseData):
self.deserialize(result["data"]),
),
)
for result in cursor
for result in cursor.mappings()
]
def get_link(
@ -1074,7 +1074,7 @@ class UserData(BaseData):
# score doesn't exist
return None
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return ValidatedDict(self.deserialize(result["data"]))
def get_links(self, game: GameConstants, version: int, userid: UserID) -> List[Link]:
@ -1103,7 +1103,7 @@ class UserData(BaseData):
UserID(result["other_userid"]),
self.deserialize(result["data"]),
)
for result in cursor
for result in cursor.mappings()
]
def put_link(
@ -1196,7 +1196,7 @@ class UserData(BaseData):
sql = "SELECT balance FROM balance WHERE userid = :userid AND arcadeid = :arcadeid"
cursor = self.execute(sql, {"userid": userid, "arcadeid": arcadeid})
if cursor.rowcount == 1:
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return result["balance"]
else:
return 0
@ -1242,7 +1242,7 @@ class UserData(BaseData):
sql = "SELECT refid FROM refid WHERE userid = :userid AND game = :game AND version = :version"
cursor = self.execute(sql, {"userid": userid, "game": game.value, "version": version})
if cursor.rowcount == 1:
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return result["refid"]
else:
return self.create_refid(game, version, userid)
@ -1265,7 +1265,7 @@ class UserData(BaseData):
sql = "SELECT extid FROM extid WHERE userid = :userid AND game = :game"
cursor = self.execute(sql, {"userid": userid, "game": game.value})
if cursor.rowcount == 1:
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return result["extid"]
else:
return None
@ -1371,7 +1371,7 @@ class UserData(BaseData):
sql = "SELECT refid FROM refid WHERE userid = :userid AND game = :game AND version = :version"
cursor = self.execute(sql, {"userid": userid, "game": game.value, "version": version})
if cursor.rowcount == 1:
result = cursor.fetchone()
result = cursor.mappings().fetchone() # type: ignore
return result["refid"]
# Shouldn't be possible, but here we are
raise AccountCreationException("Failed to recover lost race refid!")

View File

@ -1371,8 +1371,7 @@ class TXP2File(TrackedCoverage, VerboseOutput):
raise Exception("Cannot update texture with different size!")
# Now, get the raw image data, and let the TDXT container refresh the raw.
img = img.convert("RGBA")
texture.img = img
texture.img = img.convert("RGBA")
return
else:

View File

@ -377,12 +377,14 @@ class AP2PlaceObjectTag(Tag):
"blend": self.blend,
"update": self.update,
"transform": self.transform.as_dict(*args, **kwargs) if self.transform is not None else None,
"rotation_origin": self.rotation_origin.as_dict(*args, **kwargs)
if self.rotation_origin is not None
else None,
"projection": "none"
if self.projection == self.PROJECTION_NONE
else ("affine" if self.projection == self.PROJECTION_AFFINE else "perspective"),
"rotation_origin": (
self.rotation_origin.as_dict(*args, **kwargs) if self.rotation_origin is not None else None
),
"projection": (
"none"
if self.projection == self.PROJECTION_NONE
else ("affine" if self.projection == self.PROJECTION_AFFINE else "perspective")
),
"mult_color": self.mult_color.as_dict(*args, **kwargs) if self.mult_color is not None else None,
"add_color": self.add_color.as_dict(*args, **kwargs) if self.add_color is not None else None,
"hsl_shift": self.hsl_shift.as_dict(*args, **kwargs) if self.hsl_shift else None,

View File

@ -72,7 +72,7 @@ class TrackedCoverage:
def print_coverage(self, req_start: Optional[int] = None, req_end: Optional[int] = None) -> None:
for start, offset in self.get_uncovered_chunks(req_start, req_end):
print(
f"Uncovered: {hex(start)} - {hex(offset)} ({offset-start} bytes)",
f"Uncovered: {hex(start)} - {hex(offset)} ({offset - start} bytes)",
file=sys.stderr,
)

View File

@ -175,15 +175,17 @@ def viewevents() -> Response:
"refresh": url_for("admin_pages.listevents", since=-1),
"backfill": url_for("admin_pages.backfillevents", until=-1),
"viewuser": url_for("admin_pages.viewuser", userid=-1),
"jubeatsong": url_for("jubeat_pages.viewtopscores", musicid=-1)
if GameConstants.JUBEAT in g.config.support
else None,
"iidxsong": url_for("iidx_pages.viewtopscores", musicid=-1)
if GameConstants.IIDX in g.config.support
else None,
"pnmsong": url_for("popn_pages.viewtopscores", musicid=-1)
if GameConstants.POPN_MUSIC in g.config.support
else None,
"jubeatsong": (
url_for("jubeat_pages.viewtopscores", musicid=-1) if GameConstants.JUBEAT in g.config.support else None
),
"iidxsong": (
url_for("iidx_pages.viewtopscores", musicid=-1) if GameConstants.IIDX in g.config.support else None
),
"pnmsong": (
url_for("popn_pages.viewtopscores", musicid=-1)
if GameConstants.POPN_MUSIC in g.config.support
else None
),
},
)

View File

@ -10,7 +10,6 @@ from bemani.data import Data, Config, Score, Attempt, Link, Song, UserID, Remote
class FrontendBase(ABC):
"""
All subclasses should override this attribute with the string
the game series uses in the DB.

View File

@ -20,9 +20,9 @@ class EAmuseProtocol:
A wrapper object that encapsulates encoding/decoding the E-Amusement protocol by Konami.
"""
SHARED_SECRET: Final[
bytes
] = b"\x69\xD7\x46\x27\xD9\x85\xEE\x21\x87\x16\x15\x70\xD0\x8D\x93\xB1\x24\x55\x03\x5B\x6D\xF0\xD8\x20\x5D\xF5"
SHARED_SECRET: Final[bytes] = (
b"\x69\xD7\x46\x27\xD9\x85\xEE\x21\x87\x16\x15\x70\xD0\x8D\x93\xB1\x24\x55\x03\x5B\x6D\xF0\xD8\x20\x5D\xF5"
)
XML: Final[int] = 1
BINARY: Final[int] = 2

View File

@ -26,6 +26,9 @@ class FakeCursor:
self.rowcount = len(rows)
self.pos = -1
def mappings(self) -> "FakeCursor":
return self
def fetchone(self) -> Dict[str, Any]:
if len(self.__rows) != 1:
raise Exception(f"Tried to fetch one row and there are {len(self.__rows)} rows!")

View File

@ -71,8 +71,7 @@ class TestParallel(unittest.TestCase):
def test_class(self) -> None:
class Base(ABC):
def fun(self, x: int) -> int:
...
def fun(self, x: int) -> int: ...
class A(Base):
def fun(self, x: int) -> int:

View File

@ -50,7 +50,7 @@ def main() -> None:
config["database"]["read_only"] = True
if args.profile:
from werkzeug.contrib.profiler import ProfilerMiddleware
from werkzeug.middleware.profiler import ProfilerMiddleware
app.wsgi_app = ProfilerMiddleware(app.wsgi_app, profile_dir=".") # type: ignore

View File

@ -103,7 +103,7 @@ def main() -> None:
register_games()
if args.profile:
from werkzeug.contrib.profiler import ProfilerMiddleware
from werkzeug.middleware.profiler import ProfilerMiddleware
app.wsgi_app = ProfilerMiddleware(app.wsgi_app, profile_dir=".") # type: ignore

File diff suppressed because it is too large Load Diff

View File

@ -170,7 +170,7 @@ if __name__ == "__main__":
register_games()
if args.profile:
from werkzeug.contrib.profiler import ProfilerMiddleware
from werkzeug.middleware.profiler import ProfilerMiddleware
app.wsgi_app = ProfilerMiddleware(app.wsgi_app, profile_dir=".") # type: ignore

View File

@ -181,9 +181,26 @@ class StructPrinter:
raise Exception("Cannot display string as hex!")
line.append(bs.decode(self.default_encoding))
else:
# Trick python into supporting our "z" format if it has length numbers on it.
nullTerminated = False
if spec[-1] == "z":
nullTerminated = True
spec = spec[:-1] + "s"
size = struct.calcsize(prefix + spec)
chunk = self.pe.data[offset : (offset + size)]
if spec != "x":
if spec[-1] == "s":
# Support length for s/z with proper decoding.
if nullTerminated:
# Null-terminated so we should remove any nulls.
while chunk and chunk[-1:] == b"\x00":
chunk = chunk[:-1]
if dohex:
raise Exception("Cannot display string as hex!")
line.append(chunk.decode(self.default_encoding))
elif spec != "x":
if dohex:
line.append(hex(struct.unpack(prefix + spec, chunk)[0]))
else:
@ -228,7 +245,7 @@ Some examples of valid format specifiers and what they do are as follows:
*(hbb) = Decodes an array of pointers to a structure containing a short and two bytes, decoding that short and both bytes for each entry in the array.
*z = Decodes an array null-terminated string pointers.
*z = Decodes an array of null-terminated string pointers.
Ih&h = Decodes an array of structures containing an unsigned integer and two shorts, displaying the second short in hex instead of decimal.
@ -281,10 +298,11 @@ Ih&h = Decodes an array of structures containing an unsigned integer and two sho
"for details. Additionally, prefixing a format specifier with * allows dereferencing pointers. "
"Surround a chunk of format specifiers with parenthesis to dereference structures. Note that "
"structures can be arbitrarily nested to decode complex data types. For ease of unpacking C string "
'pointers, the specifier "z" is recognzied to mean null-terminated string. A & preceeding a '
"format specifier means that we should convert to hex before displaying. For the ease of decoding "
'enumerations, the specifier "#" is recognized to mean entry number. You can provide it an '
'offset value such as "+20#" to start at a certain number.'
'pointers, the specifier "z" is recognzied to mean null-terminated string. Much like the "s" specifier '
'the "z" specifier is allowed an integer prefix for inline length. Both "s" and "z" respect the '
"specified encoding. A & preceeding a format specifier means that we should convert to hex before "
'displaying. For the ease of decoding enumerations, the specifier "#" is recognized to mean entry '
'number. You can provide it an offset value such as "+20#" to start at a certain number.'
),
type=str,
default=None,
@ -294,7 +312,7 @@ Ih&h = Decodes an array of structures containing an unsigned integer and two sho
"--emulate-code",
help=(
"Hex offset pair of addresses where we should emulate x86/x64 code to "
"reconstuct a dynamic psmap structure, separated by a colon. This can "
"reconstuct a dynamic memory structure, separated by a colon. This can "
"be specified as either a raw offset into the DLL or as a virtual offset. "
"If multiple sections must be emulated you can specify this multiple times."
),
@ -306,7 +324,7 @@ Ih&h = Decodes an array of structures containing an unsigned integer and two sho
"--emulate-function",
help=(
"Hex offset address of a function that we should emulate to reconstruct a "
"dynamic psmap structure. This can be specified as either a raw offset into "
"dynamic memory structure. This can be specified as either a raw offset into "
"the DLL or as a virtual offset. If multiple functions must be emulated you "
"can specify this multiple times."
),

View File

@ -30,6 +30,7 @@ from bemani.client.popn import (
PopnMusicUsaNekoClient,
PopnMusicPeaceClient,
PopnMusicKaimeiClient,
PopnMusicUnilabClient,
)
from bemani.client.ddr import (
DDRX2Client,
@ -110,6 +111,12 @@ def get_client(proto: ClientProtocol, pcbid: str, game: str, config: Dict[str, A
pcbid,
config,
)
if game == "pnm-unilab":
return PopnMusicUnilabClient(
proto,
pcbid,
config,
)
if game == "jubeat-saucer":
return JubeatSaucerClient(
proto,
@ -370,6 +377,12 @@ def mainloop(
"old_profile_model": "M39:J:B:A",
"avs": "2.15.8 r6631",
},
"pnm-unilab": {
"name": "Pop'n Music Unilab",
"model": "M39:J:B:A:2024073100",
"old_profile_model": "M39:J:B:A",
"avs": "2.15.8 r6631",
},
"jubeat-saucer": {
"name": "Jubeat Saucer",
"model": "L44:J:A:A:2014012802",
@ -626,6 +639,7 @@ def main() -> None:
"pnm-24": "pnm-usaneko",
"pnm-25": "pnm-peace",
"pnm-26": "pnm-kaimei",
"pnm-27": "pnm-unilab",
"iidx-20": "iidx-tricoro",
"iidx-21": "iidx-spada",
"iidx-22": "iidx-pendual",

View File

@ -10,8 +10,9 @@ pushd /path/to/git/checkout
source /path/to/your/virtualenv/bin/activate
# Install dependencies, install this library.
pip install --upgrade pip
pip install . -U --force-reinstall
python3 -m pip install --upgrade pip wheel setuptools
python3 -m pip install --upgrade -r requirements.txt
python3 -m pip install --upgrade .
# Copy the WSGI files over the old ones, recompile JSX.
cp bemani/wsgi/*.wsgi /path/to/your/wsgi/files
@ -24,3 +25,4 @@ deactivate
popd
sudo service uwsgi restart && ./preload
echo "Done!"

View File

@ -17,7 +17,7 @@ server {
location ^~ /static/ {
include /etc/nginx/mime.types;
root /path/to/your/virtualenv/lib/python3.6/site-packages/bemani/frontend/;
root /path/to/your/virtualenv/lib/python3.8/site-packages/bemani/frontend/;
}
location ^~ /jsx/ {

View File

@ -5,7 +5,7 @@
# warm the cache. Note that this isn't necessary if you are compiling
# JSX files for static serving via nginx.
cd /path/to/your/virtualenv/lib/python3.6/site-packages/bemani/frontend/static
cd /path/to/your/virtualenv/lib/python3.8/site-packages/bemani/frontend/static
for url in $(find -name "*.react.js" | sed 's,^\.,https://your-domain.com/jsx,'); do
echo "Priming $url..."
curl $url --silent -H 'Cache-Control: no-cache' > /dev/null

View File

@ -1,5 +1,6 @@
setuptools
Cython
SQLAlchemy<2.0.0
SQLAlchemy
alembic
PyYAML
Flask
@ -15,11 +16,10 @@ types-PyYAML
types-Werkzeug
types-Flask
types-freezegun
types-python-dateutil
flake8==4.0.1
typed-ast
flake8
typed-ast ; python_version < "3.13"
freezegun
pyreact
pyreact @ git+https://github.com/DragonMinded/react-python@main
Flask-Caching
blinker
pycryptodome
@ -28,4 +28,4 @@ pefile
pillow
discord_webhook
iced-x86
pylibmc ; sys_platform != 'win32'
python-memcached

View File

@ -1,3 +1,3 @@
#! /bin/bash
flake8 bemani/ --ignore E203,E501,E252,E741,W503,W504,B006,B008,B009 | grep -v "migrations\/"
flake8 bemani/ --ignore E203,E501,E252,E704,E721,E741,W503,W504,B006,B008,B009 | grep -v "migrations\/"

View File

@ -11,6 +11,7 @@ declare -a arr=(
"pnm-24"
"pnm-25"
"pnm-26"
"pnm-27"
"iidx-20"
"iidx-21"
"iidx-22"