Browse Source

Strategická: Reforma db a automatická testovací hra

master
Jiří Kalvoda 2 years ago
parent
commit
9a06487a48
  1. 19
      server/bin/create_game
  2. 13
      server/bin/create_user
  3. 86
      server/hra/db.py
  4. 87
      server/hra/lib.py
  5. 6
      server/hra/web/api.py
  6. 2
      server/hra/web/game.py
  7. 16
      server/hra/web/pages.py

19
server/bin/create_game

@ -2,12 +2,15 @@
from hra.game import logic_by_mode
import hra.db as db
import hra.lib as lib
from datetime import datetime
import sys
from sqlalchemy import exc, update
ses = db.get_session()
name="Hlavní hra"
mode = "occupy"
teams_count = 16
configuration = {
@ -49,22 +52,8 @@ configuration = {
"............xx................"
]
}
g = db.Game(game_mode=mode, configuration=configuration, teams_count=teams_count)
ses.add(g)
ses.commit()
g.lock()
s = db.State(game_id=g.game_id, round=0, state=g.get_logic().zero_state())
ses.add(s)
for i in range(teams_count):
t = db.Team(team_id=i, game_id=g.game_id, name="")
ses.add(t)
g.current_round = 0
g.working_on_next_state = False
g = lib.create_game(mode=mode, teams_count=teams_count, configuration=configuration, name=name)
ses.commit()
print(f"Přidána hra {g.game_id}. ")

13
server/bin/create_user

@ -6,6 +6,7 @@ from sqlalchemy import exc, update
import sys
import hra.db as db
import hra.lib as lib
parser = argparse.ArgumentParser()
parser.add_argument("username", help="Username")
@ -14,13 +15,11 @@ parser.add_argument("--org", help="Přidělí org prává", action='store_true')
args = parser.parse_args()
u = db.User(org=args.org, username=args.username, passwd=hash_passwd(args.passwd))
u.gen_token()
try:
db.get_session().add(u)
db.get_session().commit()
except exc.IntegrityError:
print("Uživatelské jméno již existuje")
sys.exit(1)
u = lib.create_user(args.username, args.passwd, org=args.org)
except lib.UsernameExist:
print("Uživatelské jméno již existuje")
sys.exit(1)
db.get_session().commit()
print("Přidán nový uživatel.")
print(u.token)

86
server/hra/db.py

@ -1,5 +1,5 @@
from sqlalchemy import \
Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, \
Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, Enum, \
text, func, \
create_engine, inspect, select
from sqlalchemy.engine import Engine
@ -14,6 +14,8 @@ from sqlalchemy.sql.sqltypes import Numeric
from typing import Any, Optional, List, Tuple
import secrets
import string
from enum import Enum as PythonEnum, auto
import hra.config as config
import hra.game
@ -26,6 +28,33 @@ _engine: Optional[Engine] = None
_session: Optional[Session] = None
flask_db: Any = None
class MOEnum(str, PythonEnum):
"""MOEnum je varianta PythonEnum, ve které se automaticky přidělované
hodnoty jmenují stejně jako klíče a funguje serializace do JSONu."""
def _generate_next_value_(name, start, count, last_values):
return name
@classmethod
def choices(enum) -> List[Tuple[str, str]]:
out = []
for item in enum:
out.append((item.name, item.friendly_name()))
return out
def friendly_name(self) -> str:
return str(self)
@classmethod
def coerce(enum, name):
if isinstance(name, enum):
return name
try:
return enum[name]
except KeyError:
raise ValueError(name)
def get_engine() -> Engine:
global _engine
if _engine is None:
@ -62,31 +91,63 @@ class User(Base):
def print(self):
return self.username + (" (org)" if self.org else "")
class BigData(Base):
__tablename__ = 'bigdata'
id = Column(Integer, primary_key=True)
data = Column(JSONB, nullable=False)
def get_big_data(id):
return get_session().query(BigData).filter_by(id=id).one_or_none().data
def new_big_data(d):
o = BigData(data=d)
get_session().add(o)
get_session().flush()
return o.id
class StepMode(MOEnum):
none = auto()
org = auto()
user = auto()
automatic = auto()
class Game(Base):
__tablename__ = 'games'
game_id = Column(Integer, primary_key=True)
configuration = Column(JSONB, nullable=False)
name = Column(String(80), nullable=True)
configuration = Column(Integer, ForeignKey('bigdata.id'), nullable=False)
step_mode = Column(Enum(StepMode, name='step_mode'), nullable=False, default=StepMode.none)
step_every_s = Column(Integer, nullable=False, default=60)
game_mode = Column(String(80), nullable=False)
teams_count = Column(Integer, nullable=False)
working_on_next_state = Column(Boolean, nullable=False, default=True)
current_round = Column(Integer, default=-1)
def get_configuration(self):
return get_big_data(self.configuration)
def current_state(self, none_if_working=True) -> Optional['State']:
if none_if_working and self.working_on_next_state is True:
return None
return get_session().query(State).filter_by(game_id=self.game_id, round=self.current_round).order_by(State.round.desc()).first()
return get_session().query(State).filter_by(game_id=self.game_id, round=self.current_round).one_or_none()
def get_logic(self) -> 'hra.game.Logic':
return hra.game.logic_by_mode[self.game_mode](self.teams_count, self.configuration)
return hra.game.logic_by_mode[self.game_mode](self.teams_count, self.get_configuration())
def lock(self) -> 'Game':
ses = get_session()
ses.expire_all()
return ses.query(Game).filter_by(game_id=self.game_id).with_for_update().first()
def print(self):
return f"{self.game_id}: <name>"
def print(self) -> str:
name = self.name
if not name:
name = "<name>"
return f"{self.game_id}: {name}"
class Team(Base):
@ -108,9 +169,13 @@ class Team(Base):
class State(Base):
__tablename__ = 'states'
create_time = Column(DateTime, nullable=False)
game_id = Column(Integer, ForeignKey('games.game_id'), primary_key=True, nullable=False)
round = Column(Integer, primary_key=True)
state = Column(JSONB)
state = Column(Integer, ForeignKey('bigdata.id'), nullable=False)
def get_state(self):
return get_big_data(self.state)
game = relationship('Game', primaryjoin='State.game_id == Game.game_id')
@ -120,6 +185,11 @@ class Move(Base):
game_id = Column(Integer, ForeignKey('games.game_id'), primary_key=True, nullable=False)
round = Column(Integer, primary_key=True)
team_id = Column(Integer, primary_key=True)
move = Column(JSONB)
move = Column(Integer, ForeignKey('bigdata.id'), nullable=True)
def get_move(self):
if self.move is None:
return None
return get_big_data(self.move)
game = relationship('Game', primaryjoin='Move.game_id == Game.game_id')

87
server/hra/lib.py

@ -1,6 +1,9 @@
import json
import hra.config as config
import hra.db as db
from hra.util import hash_passwd
from datetime import datetime
from sqlalchemy import exc, update
class DuplicitMakeStep(Exception):
pass
@ -11,6 +14,7 @@ def game_step(game_id: int):
ses.expire_all()
game = ses.query(db.Game).filter_by(game_id=game_id).with_for_update().one_or_none()
assert game is not None
time = datetime.now()
if game.working_on_next_state:
ses.commit()
raise DuplicitMakeStep()
@ -19,17 +23,25 @@ def game_step(game_id: int):
old_round_id = game.current_round
new_round_id = old_round_id + 1
old_state = ses.query(db.State).filter_by(game_id=game.game_id, round=old_round_id).one_or_none()
old_state = ses.query(db.State).filter_by(game_id=game.game_id, round=old_round_id).one_or_none().get_state()
moves = [None for _ in range(game.teams_count)]
for i in ses.query(db.Move).filter_by(game_id=game.game_id, round=old_round_id).all():
moves[i.team_id] = i.move
moves[i.team_id] = i.get_move()
print(old_state)
print(moves)
print(old_round_id)
print()
ses.commit()
x, points = game.get_logic().step(old_state.state, moves, old_round_id)
x, points = game.get_logic().step(old_state, moves, old_round_id)
print(x)
print(points)
new_state = db.State(game_id=game.game_id, round=new_round_id, state=x)
new_state = db.State(game_id=game.game_id, round=new_round_id, state=db.new_big_data(x), create_time=time)
ses.add(new_state)
ses.expire_all()
@ -48,3 +60,70 @@ def game_restore_broken(game_id: int) -> None:
ses.commit()
def create_game(mode, teams_count, configuration={}, test_for=None, name=None, step_mode=db.StepMode.none):
ses = db.get_session()
g = db.Game(game_mode=mode, configuration=db.new_big_data(configuration), teams_count=teams_count, name=name, step_mode=step_mode)
ses.add(g)
ses.flush()
g.lock()
s = db.State(game_id=g.game_id, round=0, state=db.new_big_data(g.get_logic().zero_state()), create_time=datetime.now())
ses.add(s)
if test_for is not None:
for i in range(teams_count):
t = db.Team(team_id=i, game_id=g.game_id, name=f"test_{i}", user_id=test_for.id)
ses.add(t)
else:
for i in range(teams_count):
t = db.Team(team_id=i, game_id=g.game_id, name="")
ses.add(t)
g.current_round = 0
g.working_on_next_state = False
return g
def create_test_game(user):
mode = "occupy"
teams_count = 4
configuration = {
"teams_width": 2,
"teams_height": 2,
"width_per_team": 10,
"height_per_team": 10,
"hills": [
".xx....x..",
".xxx...xx.",
"...x......",
".......xx.",
".......xx.",
".......x..",
"..........",
"..........",
"..........",
"..........",
],
}
return create_game(mode=mode, teams_count=teams_count, configuration=configuration, test_for=user, name=f"Testovací hra uživatele {user.username}", step_mode=db.StepMode.user)
class UsernameExist(Exception):
pass
def create_user(username, passwd, org=False, test_game=True):
u = db.User(org=org, username=username, passwd=hash_passwd(passwd))
u.gen_token()
try:
db.get_session().add(u)
db.get_session().flush()
except exc.IntegrityError:
raise UsernameExist()
if test_game:
create_test_game(u)
return u

6
server/hra/web/api.py

@ -51,7 +51,7 @@ def api_state():
"status": "ok",
"round": state.round,
"team_id": team_id,
"state": game.get_logic().personalize_state(state.state, team_id, state.round),
"state": game.get_logic().personalize_state(state.get_state(), team_id, state.round),
})
@app.route("/api/action", methods=['POST'])
@ -72,7 +72,7 @@ def api_action():
return to_late()
try:
warnings = game.get_logic().validate_move(state.state, team_id, j, round_id)
warnings = game.get_logic().validate_move(state.get_state(), team_id, j, round_id)
except Exception as e:
return json.dumps({
"error": "error",
@ -91,7 +91,7 @@ def api_action():
move = db.Move(game_id=game.game_id, team_id=team_id, round=round_id)
db.get_session().add(move)
move.move = j
move.move = db.new_big_data(j)
db.get_session().commit()

2
server/hra/web/game.py

@ -43,7 +43,7 @@ class WLogic:
@add_wlogic
class Occupy(WLogic):
def view(self, state: db.State, team: Optional[db.Team]):
s = state.state
s = state.get_state()
if team is not None:
s = self.logic.personalize_state(s, team.team_id, state.round)
b = BasePage()

16
server/hra/web/pages.py

@ -17,6 +17,7 @@ import hra.db as db
from hra.web import app
import hra.web.jinja_mac as jinja_mac
from hra.util import hash_passwd
import hra.lib as lib
@html.WrapAfterBuilder_decorator
@ -87,7 +88,7 @@ def right_for_game(game):
return True
if g.user is None:
return False
return db.get_session().query(db.Team).filter_by(game_id=game_id, user=g.user.id).one_or_none() is not None
return db.get_session().query(db.Team).filter_by(game_id=game.game_id, user_id=g.user.id).count() > 0
def right_for_team(team):
@ -134,17 +135,14 @@ class LoginForm(FlaskForm):
def registration():
f = RegistrationForm()
if f.validate_on_submit():
u = db.User(org=False, username=f.username.data, passwd=hash_passwd(f.passwd.data))
u.gen_token()
try:
db.get_session().add(u)
db.get_session().commit()
except exc.IntegrityError:
lib.create_user(f.username.data, f.passwd.data)
except lib.UsernameExist:
flash("Uživatelské jméno již existuje")
else:
db.get_session().commit()
flash("Přidán nový uživatel.", 'success')
return redirect("login")
b = BasePage()
b(jinja_mac.quick_form(f, form_type='horizontal'))
return b.print_file()
@ -285,12 +283,12 @@ def web_org_games():
b.h2("Hry")
with b.p().table(_class="data full"):
with b.thead():
b.line().th()("Id")
b.line().th()("Id: Jméno")
b.line().th()("Kolo")
b.line().th()("Akce")
for g in games:
with b.tr():
b.line().td()(g.game_id)
b.line().td()(g.print())
with b.line().td():
if g.working_on_next_state:
b.b()(g.current_round, "++")

Loading…
Cancel
Save