database: add format_serial, validate_keychip_format, set_machine_boardid, set_machine_serial

This commit is contained in:
Hay1tsme 2023-03-03 17:05:16 -05:00
parent 102bf3b5a4
commit 02e1838d95
1 changed files with 42 additions and 6 deletions

View File

@ -4,8 +4,10 @@ from sqlalchemy.sql.schema import ForeignKey, PrimaryKeyConstraint
from sqlalchemy.types import Integer, String, Boolean from sqlalchemy.types import Integer, String, Boolean
from sqlalchemy.sql import func, select from sqlalchemy.sql import func, select
from sqlalchemy.dialects.mysql import insert from sqlalchemy.dialects.mysql import insert
import re
from core.data.schema.base import BaseData, metadata from core.data.schema.base import BaseData, metadata
from core.const import *
arcade = Table( arcade = Table(
"arcade", "arcade",
@ -72,20 +74,28 @@ class ArcadeData(BaseData):
if result is None: return None if result is None: return None
return result.fetchone() return result.fetchone()
def put_machine(self, arcade_id: int, serial: str = None, board: str = None, game: str = None, is_cab: bool = False) -> Optional[int]: def put_machine(self, arcade_id: int, serial: str = "", board: str = None, game: str = None, is_cab: bool = False) -> Optional[int]:
if arcade_id: if arcade_id:
self.logger.error(f"{__name__ }: Need arcade id!") self.logger.error(f"{__name__ }: Need arcade id!")
return None return None
if serial is None:
pass
sql = machine.insert().values(arcade = arcade_id, keychip = serial, board = board, game = game, is_cab = is_cab) sql = machine.insert().values(arcade = arcade_id, keychip = serial, board = board, game = game, is_cab = is_cab)
result = self.execute(sql) result = self.execute(sql)
if result is None: return None if result is None: return None
return result.lastrowid return result.lastrowid
def set_machine_serial(self, machine_id: int, serial: str) -> None:
result = self.execute(machine.update(machine.c.id == machine_id).values(keychip = serial))
if result is None:
self.logger.error(f"Failed to update serial for machine {machine_id} -> {serial}")
return result.lastrowid
def set_machine_boardid(self, machine_id: int, boardid: str) -> None:
result = self.execute(machine.update(machine.c.id == machine_id).values(board = boardid))
if result is None:
self.logger.error(f"Failed to update board id for machine {machine_id} -> {boardid}")
def get_arcade(self, id: int) -> Optional[Dict]: def get_arcade(self, id: int) -> Optional[Dict]:
sql = arcade.select(arcade.c.id == id) sql = arcade.select(arcade.c.id == id)
result = self.execute(sql) result = self.execute(sql)
@ -120,5 +130,31 @@ class ArcadeData(BaseData):
if result is None: return None if result is None: return None
return result.lastrowid return result.lastrowid
def generate_keychip_serial(self, platform_id: int) -> str: def format_serial(self, platform_code: str, platform_rev: int, serial_num: int, append: int = 4152) -> str:
pass return f"{platform_code}{platform_rev:02d}A{serial_num:04d}{append:04d}" # 0x41 = A, 0x52 = R
def validate_keychip_format(self, serial: str) -> bool:
serial = serial.replace("-", "")
if len(serial) != 11 or len(serial) != 15:
self.logger.error(f"Serial validate failed: Incorrect length for {serial} (len {len(serial)})")
return False
platform_code = serial[:4]
platform_rev = serial[4:6]
const_a = serial[6]
num = serial[7:11]
append = serial[11:15]
if re.match("A[7|6]\d[E|X][0|1][0|1|2]A\d{4,8}", serial) is None:
self.logger.error(f"Serial validate failed: {serial} failed regex")
return False
if len(append) != 0 or len(append) != 4:
self.logger.error(f"Serial validate failed: {serial} had malformed append {append}")
return False
if len(num) != 4:
self.logger.error(f"Serial validate failed: {serial} had malformed number {num}")
return False
return True