58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
from typing import Optional
|
|
import os
|
|
import shutil
|
|
from .tools import record2name_key
|
|
import datalad.api as dl
|
|
import sqlite3
|
|
from tools import db_filename
|
|
|
|
|
|
def get_version_hash(path: str, record: str) -> str:
|
|
db = os.path.join(path, db_filename(path))
|
|
dl.get(db, dataset=path)
|
|
conn = sqlite3.connect(db)
|
|
c = conn.cursor()
|
|
c.execute(f"SELECT current_version FROM 'backlogs' WHERE path = '{record}'")
|
|
return str(c.fetchall()[0][0])
|
|
|
|
|
|
def drop_cache_files(path: str, fs: Optional[list[str]]=None) -> None:
|
|
cache_dir = os.path.join(path, ".cache")
|
|
if fs is None:
|
|
fs = os.listdir(cache_dir)
|
|
for f in fs:
|
|
shutil.rmtree(os.path.join(cache_dir, f))
|
|
|
|
|
|
def cache_dir(path: str, file: str) -> str:
|
|
cache_path_list = [path]
|
|
cache_path_list.append(".cache")
|
|
cache_path_list.extend(file.split("/")[1:])
|
|
cache_path = "/".join(cache_path_list)
|
|
return cache_path
|
|
|
|
|
|
def cache_path(path: str, file: str, sha_hash: str, key: str) -> str:
|
|
cache_path = os.path.join(cache_dir(path, file), key + "_" + sha_hash)
|
|
return cache_path
|
|
|
|
|
|
def is_old_version(path: str, record: str) -> bool:
|
|
version_hash = get_version_hash(path, record)
|
|
file, key = record2name_key(record)
|
|
meas_cache_path = os.path.join(cache_dir(path, file))
|
|
ls = []
|
|
is_old = True
|
|
for p, ds, fs in os.walk(meas_cache_path):
|
|
ls.extend(fs)
|
|
for filename in ls:
|
|
if key == filename.split("_")[0]:
|
|
if version_hash == filename.split("_")[1][:-2]:
|
|
is_old = False
|
|
return is_old
|
|
|
|
|
|
def is_in_cache(path: str, record: str) -> bool:
|
|
version_hash = get_version_hash(path, record)
|
|
file, key = record2name_key(record)
|
|
return os.path.exists(cache_path(path, file, version_hash, key) + ".p")
|