|
|
import os |
|
|
import psutil |
|
|
import logging |
|
|
from pathlib import Path |
|
|
|
|
|
|
|
|
logging.basicConfig( |
|
|
format='%(asctime)s %(levelname)-8s %(message)s', |
|
|
level=logging.INFO, |
|
|
datefmt='%m/%d %H:%M:%S') |
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
|
def log_sys_info(): |
|
|
cpu_cores = psutil.cpu_count() |
|
|
|
|
|
cpu_percent = '{:.2f}'.format(psutil.cpu_percent()) + '%' |
|
|
mem = psutil.virtual_memory() |
|
|
mem_total = '{:.2f}'.format(mem.total / 1024 / 1024 / 1024) + 'GB' |
|
|
mem_used = '{:.2f}'.format(mem.used / 1024 / 1024 / 1024) + 'GB' |
|
|
mem_percent = '{:.2f}'.format(mem.percent) + '%' |
|
|
disk = psutil.disk_usage('.') |
|
|
disk_total = '{:.2f}'.format(disk.total / 1024 / 1024 / 1024) + 'GB' |
|
|
disk_used = '{:.2f}'.format(disk.used / 1024 / 1024 / 1024) + 'GB' |
|
|
disk_percent = '{:.2f}'.format(disk.percent) + '%' |
|
|
|
|
|
logger.info('======== SYSTEM INFO =========') |
|
|
logger.info(f'CPU: {cpu_cores} cores, {cpu_percent} used') |
|
|
logger.info(f'RAM: {mem_used} / {mem_total}, {mem_percent} used') |
|
|
logger.info(f'DISK: {disk_used} / {disk_total}, {disk_percent} used') |
|
|
logger.info('==============================') |
|
|
|
|
|
|
|
|
class CHAT_DB: |
|
|
def __init__(self): |
|
|
self.prompts = {} |
|
|
self.results = {} |
|
|
self.index = 1 |
|
|
self.lockfile = '.lock' |
|
|
|
|
|
def set(self, index, prompt=None, result=None): |
|
|
assert prompt or result |
|
|
if prompt: |
|
|
if index in self.prompts: |
|
|
raise ValueError('Prompt already exists') |
|
|
self.prompts[index] = prompt |
|
|
self.index += 1 |
|
|
if result: |
|
|
self.results[index] = result |
|
|
|
|
|
def lock(self): |
|
|
if not os.path.exists(self.lockfile): |
|
|
Path(self.lockfile).touch(exist_ok=True) |
|
|
|
|
|
def unlock(self): |
|
|
if os.path.exists(self.lockfile): |
|
|
os.remove(self.lockfile) |
|
|
|
|
|
def islocked(self): |
|
|
return os.path.exists(self.lockfile) |
|
|
|
|
|
def clean(self): |
|
|
if len(self.prompts) > 100: |
|
|
self.prompts = dict(list(self.prompts.items())[-100:]) |
|
|
k = list(set(self.prompts.keys()).intersection(set(self.results.keys()))) |
|
|
self.prompts = {i: self.prompts[i] for i in k} |
|
|
self.results = {i: self.results[i] for i in k} |
|
|
log_sys_info() |
|
|
|
|
|
|
|
|
db = CHAT_DB() |
|
|
|
