119 lines
3.3 KiB
Python
119 lines
3.3 KiB
Python
# coding=utf-8
|
|
from typing import List
|
|
import tinydb
|
|
from tsgrain_controller import task_queue, util, models
|
|
|
|
|
|
class RainDB:
|
|
|
|
def __init__(self, db_path: str):
|
|
self._db = tinydb.TinyDB(db_path, default=util.serializer)
|
|
|
|
self._jobs = self._db.table('jobs', cache_size=0)
|
|
self._tasks = self._db.table('tasks', cache_size=0)
|
|
self._options = self._db.table('options', cache_size=0)
|
|
|
|
def close(self):
|
|
"""Die Datenbank schließen"""
|
|
self._db.close()
|
|
|
|
def get_jobs(self) -> List[models.Job]:
|
|
"""
|
|
Gibt alle gespeicherten Bewässerungsjobs zurück
|
|
|
|
:return: Bewässerungsjobs: dict(id -> models.Job)
|
|
"""
|
|
return [
|
|
models.Job.deserialize(job_data, job_data.doc_id)
|
|
for job_data in self._jobs.all()
|
|
]
|
|
|
|
def get_job(self, job_id: int) -> models.Job:
|
|
"""
|
|
Gibt den Bewässerungsjob mit der gegebenen ID zurück
|
|
|
|
:param job_id: ID des Bewässerungsjobs
|
|
:return: Bewässerungsjob
|
|
:raise KeyError: wenn Job nicht gefunden
|
|
"""
|
|
job = self._jobs.get(None, job_id)
|
|
if job is None:
|
|
raise KeyError
|
|
|
|
return models.Job.deserialize(job, job_id)
|
|
|
|
def insert_job(self, job: models.Job) -> int:
|
|
"""
|
|
Fügt der Datenbank einen neuen Bewässerungsjob hinzu
|
|
|
|
:param job: Bewässerungsjob
|
|
:return: ID des neuen Bewässerungsjobs
|
|
"""
|
|
job_id = self._jobs.insert(util.serializer(job))
|
|
job.id = job_id
|
|
return job_id
|
|
|
|
def update_job(self, job: models.Job):
|
|
"""
|
|
Aktualisiert einen Bewässerungsjob
|
|
|
|
:param job: Bewässerungsjob
|
|
"""
|
|
self._jobs.update(util.serializer(job), None, [job.id])
|
|
|
|
def delete_job(self, job_id: int):
|
|
"""
|
|
Lösche den Bewässerungsjob mit der gegebenen ID
|
|
|
|
:param job_id: ID des Bewässerungsjobs
|
|
:raise KeyError: wenn Job nicht gefunden
|
|
"""
|
|
self._jobs.remove(None, [job_id])
|
|
|
|
def store_queue(self, queue: task_queue.TaskQueue):
|
|
"""
|
|
Speichere die aktuelle Warteschlange in der Datenbank
|
|
|
|
:param queue: Warteschlange
|
|
"""
|
|
self.empty_queue()
|
|
for task in queue.serialize():
|
|
self._tasks.insert(util.serializer(task))
|
|
|
|
def load_queue(self, queue: task_queue.TaskQueue):
|
|
"""
|
|
Lade die gespeicherten Tasks aus der Datenbank in die Warteschlange
|
|
|
|
:param queue: Warteschlange
|
|
"""
|
|
for task_data in self._tasks.all():
|
|
task = models.Task.deserialize(task_data)
|
|
queue.tasks.append(task)
|
|
self.empty_queue()
|
|
|
|
def empty_queue(self):
|
|
"""Lösche die gespeicherte Warteschlange"""
|
|
self._tasks.truncate()
|
|
|
|
def set_auto_mode(self, state: bool):
|
|
"""
|
|
Speichere den Status des Automatikmodus
|
|
|
|
:param state: Automatikstatus
|
|
"""
|
|
self._options.upsert({
|
|
'key': 'auto_mode',
|
|
'val': state
|
|
},
|
|
tinydb.Query().key == 'auto_mode')
|
|
|
|
def get_auto_mode(self) -> bool:
|
|
"""
|
|
Rufe den Status des Automatikmodus ab
|
|
|
|
:return: Automatikstatus
|
|
"""
|
|
option = self._options.get(tinydb.Query().key == 'auto_mode')
|
|
if option is None:
|
|
return False
|
|
return option.get('val', False)
|