91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
import json
|
|
import logging
|
|
from mqtt.database import get_db
|
|
|
|
logger = logging.getLogger("builder.database")
|
|
|
|
|
|
async def insert_built_melody(melody_id: str, name: str, pid: str, steps: str) -> None:
|
|
db = await get_db()
|
|
await db.execute(
|
|
"""INSERT INTO built_melodies (id, name, pid, steps, assigned_melody_ids)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
(melody_id, name, pid, steps, json.dumps([])),
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def update_built_melody(melody_id: str, name: str, pid: str, steps: str) -> None:
|
|
db = await get_db()
|
|
await db.execute(
|
|
"""UPDATE built_melodies
|
|
SET name = ?, pid = ?, steps = ?, updated_at = datetime('now')
|
|
WHERE id = ?""",
|
|
(name, pid, steps, melody_id),
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def update_binary_path(melody_id: str, binary_path: str) -> None:
|
|
db = await get_db()
|
|
await db.execute(
|
|
"""UPDATE built_melodies
|
|
SET binary_path = ?, updated_at = datetime('now')
|
|
WHERE id = ?""",
|
|
(binary_path, melody_id),
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def update_progmem_code(melody_id: str, progmem_code: str) -> None:
|
|
db = await get_db()
|
|
await db.execute(
|
|
"""UPDATE built_melodies
|
|
SET progmem_code = ?, updated_at = datetime('now')
|
|
WHERE id = ?""",
|
|
(progmem_code, melody_id),
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def update_assigned_melody_ids(melody_id: str, assigned_ids: list) -> None:
|
|
db = await get_db()
|
|
await db.execute(
|
|
"""UPDATE built_melodies
|
|
SET assigned_melody_ids = ?, updated_at = datetime('now')
|
|
WHERE id = ?""",
|
|
(json.dumps(assigned_ids), melody_id),
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def get_built_melody(melody_id: str) -> dict | None:
|
|
db = await get_db()
|
|
rows = await db.execute_fetchall(
|
|
"SELECT * FROM built_melodies WHERE id = ?", (melody_id,)
|
|
)
|
|
if not rows:
|
|
return None
|
|
row = dict(rows[0])
|
|
row["assigned_melody_ids"] = json.loads(row["assigned_melody_ids"] or "[]")
|
|
return row
|
|
|
|
|
|
async def list_built_melodies() -> list[dict]:
|
|
db = await get_db()
|
|
rows = await db.execute_fetchall(
|
|
"SELECT * FROM built_melodies ORDER BY updated_at DESC"
|
|
)
|
|
results = []
|
|
for row in rows:
|
|
r = dict(row)
|
|
r["assigned_melody_ids"] = json.loads(r["assigned_melody_ids"] or "[]")
|
|
results.append(r)
|
|
return results
|
|
|
|
|
|
async def delete_built_melody(melody_id: str) -> None:
|
|
db = await get_db()
|
|
await db.execute("DELETE FROM built_melodies WHERE id = ?", (melody_id,))
|
|
await db.commit()
|