104 lines
2.7 KiB
Python
104 lines
2.7 KiB
Python
#!/usr/bin/python3
|
|
from pathlib import Path
|
|
from utils_types import Phone, TaskList
|
|
import os
|
|
import shutil
|
|
import json
|
|
import uuid
|
|
import utils
|
|
|
|
class InfoCache:
|
|
dir: Path
|
|
|
|
def __init__(self, dir: Path) -> None:
|
|
self.dir = dir
|
|
if dir is not None and not dir.exists():
|
|
try:
|
|
os.mkdir(dir)
|
|
except Exception as e:
|
|
utils.log_error(str(e))
|
|
self.dir = None
|
|
|
|
|
|
|
|
def is_active(self) -> bool:
|
|
return self.dir is not None
|
|
|
|
|
|
def add_reference_audio(self, audio_id: int, src_path: Path):
|
|
if not self.is_active():
|
|
return
|
|
|
|
p = self.dir / f'ref_{audio_id}.wav'
|
|
if not p.exists():
|
|
shutil.copy(src_path, p)
|
|
|
|
|
|
def get_reference_audio(self, audio_id: int, dst_path: Path) -> bool:
|
|
if not self.is_active():
|
|
return False
|
|
|
|
p = self.dir / f'ref_{audio_id}.wav'
|
|
if p.exists():
|
|
shutil.copy(p, dst_path)
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def add_recorded_audio(self, src_path: Path, probe_id: str) -> Path:
|
|
if not self.is_active():
|
|
return None
|
|
|
|
p = self.dir / f'audio_{probe_id}.wav'
|
|
|
|
shutil.copy(src_path, p)
|
|
return p
|
|
|
|
|
|
def get_recorded_audio(self, probe_id: str) -> Path:
|
|
if not self.is_active():
|
|
return None
|
|
|
|
p = self.dir / f'audio_{probe_id}.wav'
|
|
if p.exists():
|
|
return p
|
|
else:
|
|
return None
|
|
|
|
def put_phone(self, phone: Phone):
|
|
if self.is_active():
|
|
with open(self.dir / f'phone_{phone.name}.json', 'wt') as f:
|
|
f.write(phone.dump())
|
|
|
|
def get_phone(self, name: str) -> Phone:
|
|
p = self.dir / f'phone_{name}.json'
|
|
if not p.exists():
|
|
utils.log(f'Phone definition at path {p} not found.')
|
|
return None
|
|
|
|
with open(p, 'rt') as f:
|
|
return Phone.make(f.read())
|
|
|
|
def put_tasks(self, name: str, tasks: TaskList):
|
|
p = self.dir / f'tasks_{name}.json'
|
|
with open(p, 'wt') as f:
|
|
f.write(json.dumps(tasks.tasks))
|
|
|
|
|
|
def get_tasks(self, name: str) -> TaskList:
|
|
p = self.dir / f'tasks_{name}.json'
|
|
# ToDo
|
|
|
|
|
|
def add_report(self, report: dict) -> str:
|
|
if not self.is_active():
|
|
return None
|
|
|
|
# Generate UUID manually and save under this name to cache dir
|
|
probe_id = uuid.uuid1().urn[9:]
|
|
with open(self.dir / f'{probe_id}.json', 'wt') as f:
|
|
f.write(json.dumps(report, indent=4))
|
|
return probe_id
|
|
|
|
|