You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
209 lines
6.6 KiB
209 lines
6.6 KiB
from sqlalchemy import \
|
|
Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, Enum, \
|
|
text, func, \
|
|
create_engine, inspect, select
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy.orm import relationship, sessionmaker, Session, class_mapper, joinedload, aliased
|
|
from sqlalchemy.orm.attributes import get_history
|
|
from sqlalchemy.orm.query import Query
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.sql.expression import CTE
|
|
from sqlalchemy.sql.functions import ReturnTypeFromArgs
|
|
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
|
|
|
|
Base = declarative_base()
|
|
metadata = Base.metadata
|
|
|
|
|
|
_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:
|
|
_engine = create_engine(config.SQLALCHEMY_DATABASE_URI, echo=config.SQLALCHEMY_ECHO)
|
|
return _engine
|
|
|
|
def get_session() -> Session:
|
|
global _session
|
|
if flask_db:
|
|
return flask_db.session
|
|
if _session is None:
|
|
MOSession = sessionmaker(bind=get_engine())
|
|
_session = MOSession()
|
|
return _session
|
|
|
|
|
|
|
|
class User(Base):
|
|
__tablename__ = 'users'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
org = Column(Boolean)
|
|
token = Column(String(80), unique=True, nullable=False)
|
|
username = Column(String(80), unique=True, nullable=False)
|
|
passwd = Column(String(80), nullable=False)
|
|
|
|
def gen_token(self):
|
|
self.token = ''.join(secrets.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for i in range(15))
|
|
|
|
|
|
def __repr__(self):
|
|
return '<User %r>' % self.username
|
|
|
|
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)
|
|
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).one_or_none()
|
|
|
|
def get_logic(self) -> 'hra.game.Logic':
|
|
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) -> str:
|
|
name = self.name
|
|
if not name:
|
|
name = "<name>"
|
|
return f"{self.game_id}: {name}"
|
|
|
|
|
|
class Team(Base):
|
|
__tablename__ = 'base'
|
|
__table_args__ = (
|
|
UniqueConstraint('game_id', 'team_id'),
|
|
UniqueConstraint('user_id', 'name'),
|
|
)
|
|
|
|
game_id = Column(Integer, ForeignKey('games.game_id'), nullable=False, primary_key=True)
|
|
team_id = Column(Integer, nullable=False, primary_key=True)
|
|
user_id = Column(Integer, ForeignKey('users.id'), nullable=True)
|
|
name = Column(String(80), nullable=False)
|
|
|
|
game = relationship('Game', primaryjoin='Team.game_id == Game.game_id')
|
|
user = relationship('User', primaryjoin='Team.user_id == User.id')
|
|
|
|
def print(self):
|
|
if self.user is None:
|
|
return f"{self.team_id}: -"
|
|
return f"{self.team_id}: {self.user.username}"
|
|
|
|
def get_move(self, round_id):
|
|
return get_session().query(Move).filter_by(game_id=self.game_id, round=round_id, team_id=self.team_id).one_or_none()
|
|
|
|
|
|
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(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')
|
|
|
|
class Move(Base):
|
|
__tablename__ = 'moves'
|
|
|
|
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(Integer, ForeignKey('bigdata.id'), nullable=True)
|
|
|
|
points = Column(Integer, nullable=False, default=0)
|
|
reads_count = Column(Integer, nullable=False, default=0)
|
|
ok_pushs_count = Column(Integer, nullable=False, default=0)
|
|
warnings_pushs_count = Column(Integer, nullable=False, default=0)
|
|
err_pushs_count = Column(Integer, nullable=False, default=0)
|
|
|
|
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')
|
|
|