147 lines
3.9 KiB
Python
Executable File
147 lines
3.9 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
from enum import IntEnum
|
|
from flask import g
|
|
import os
|
|
import sqlite3
|
|
from typing import Optional
|
|
|
|
|
|
def get() -> sqlite3.Connection:
|
|
if 'db' not in g:
|
|
g.db = sqlite3.connect(
|
|
os.environ.get('DATABASE_PATH', 'db.sqlite'),
|
|
detect_types=sqlite3.PARSE_DECLTYPES,
|
|
)
|
|
g.db.row_factory = sqlite3.Row
|
|
cur = g.db.cursor()
|
|
cur.execute('PRAGMA foreign_keys = on')
|
|
|
|
return g.db
|
|
|
|
|
|
def init(db):
|
|
with open('schema.sql', 'r') as f:
|
|
db.executescript(f.read())
|
|
|
|
|
|
def close():
|
|
db = g.pop('db', None)
|
|
|
|
if db is not None:
|
|
db.close()
|
|
|
|
|
|
def init_app(app):
|
|
app.teardown_appcontext(close)
|
|
app.cli.add_command(init)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
db = sqlite3.connect(
|
|
os.environ.get('DATABASE_PATH', 'db.sqlite'),
|
|
detect_types=sqlite3.PARSE_DECLTYPES,
|
|
)
|
|
db.row_factory = sqlite3.Row
|
|
init(db)
|
|
|
|
|
|
class CalibrationState(IntEnum):
|
|
Empty = 0
|
|
HasData = 1
|
|
IsValidated = 2
|
|
|
|
|
|
class Calibration:
|
|
@staticmethod
|
|
def select_args() -> str:
|
|
return 'id, state'
|
|
|
|
def __init__(self, calibration_id: int, state: int):
|
|
self.id = calibration_id
|
|
self.state = CalibrationState(state)
|
|
|
|
@staticmethod
|
|
def create(db: sqlite3.Connection) -> 'Calibration':
|
|
cur = db.cursor()
|
|
response = cur.execute(
|
|
'INSERT INTO calibration(state) VALUES (?) RETURNING ' + Calibration.select_args() + ';',
|
|
[int(CalibrationState.Empty)]
|
|
)
|
|
return Calibration.from_row(response.fetchone())
|
|
|
|
@staticmethod
|
|
def from_row(row: sqlite3.Row) -> 'Calibration':
|
|
return Calibration(*row)
|
|
|
|
def save(self, db: sqlite3.Connection):
|
|
cur = db.cursor()
|
|
cur.execute(
|
|
'UPDATE calibration SET state = ? WHERE id = ?',
|
|
[int(self.state), self.id]
|
|
)
|
|
|
|
@staticmethod
|
|
def get_from_id(calibration_id: int, db: sqlite3.Connection) -> Optional['Calibration']:
|
|
cur = db.cursor()
|
|
response = cur.execute(
|
|
'SELECT ' + Calibration.select_args() + ' FROM calibration WHERE id = ?;',
|
|
[calibration_id]
|
|
)
|
|
return Calibration.from_row(response.fetchone())
|
|
|
|
|
|
class Object:
|
|
@staticmethod
|
|
def select_args() -> str:
|
|
return 'id, name, calibration_id'
|
|
|
|
def __init__(self, object_id: int, name: str, calibration: int):
|
|
self.id = object_id
|
|
self.name = name
|
|
self.calibration_id = calibration
|
|
|
|
@staticmethod
|
|
def create(name: str, db: sqlite3.Connection) -> 'Object':
|
|
cur = db.cursor()
|
|
response = cur.execute(
|
|
'INSERT INTO object(name, calibration_id) VALUES (?, ?) RETURNING ' + Object.select_args() + ';',
|
|
[name, None]
|
|
)
|
|
return Object.from_row(response.fetchone())
|
|
|
|
def save(self, db: sqlite3.Connection):
|
|
cur = db.cursor()
|
|
cur.execute(
|
|
'UPDATE object SET name = ?, calibration_id = ? WHERE id = ?',
|
|
[self.name, self.calibration_id, self.id]
|
|
)
|
|
|
|
@staticmethod
|
|
def from_row(row: sqlite3.Row) -> 'Object':
|
|
return Object(*row)
|
|
|
|
@staticmethod
|
|
def get_from_id(object_id: int, db: sqlite3.Connection) -> Optional['Object']:
|
|
cur = db.cursor()
|
|
response = cur.execute(
|
|
'SELECT ' + Object.select_args() + ' FROM object WHERE id = ?;',
|
|
[object_id]
|
|
)
|
|
return Object.from_row(response.fetchone())
|
|
|
|
@staticmethod
|
|
def all(db: sqlite3.Connection) -> list['Object']:
|
|
cur = db.cursor()
|
|
response = cur.execute(
|
|
'SELECT ' + Object.select_args() + ' FROM object;',
|
|
[]
|
|
)
|
|
return list(map(Object.from_row, response.fetchall()))
|
|
|
|
def calibration(self, db: sqlite3.Connection) -> Optional[Calibration]:
|
|
if self.calibration_id is None:
|
|
return None
|
|
|
|
return Calibration.get_from_id(self.calibration_id, db)
|