from pydiana.scripting import cfg_reader as cfg
from pydiana.scripting import file_handler_classes as fhc
from pydiana.multiprocessing import multp_classes as multp
import os
import warnings
[docs]
class analyser():
def __init__(self,rawfolder:str,storagedir:str,anadir:str,seqs:list=None):
self.clear()
self.set_rawfolder(rawfolder)
self.set_storagedir(storagedir)
self.set_anadir(anadir)
self.set_seqs(seqs)
self.all_seqs=['wiki','reader','preprocess']
[docs]
def clear(self):
self.set_rawfolder()
self.set_storagedir()
self.set_seqs()
self.set_anadir()
self.init_dianacaller()
self.set_current_cfg()
self.wiki_manager=None
self.wiki=None
[docs]
def set_current_cfg(self,sequence=None):
seq =sequence
[docs]
def set_rawfolder(self,folder:str=None):
if folder is not None:
if folder[-1]!='/':
folder+='/'
self.rawfolder = folder
[docs]
def set_storagedir(self,folder:str=None):
if folder is not None:
if folder[-1]!='/':
folder+='/'
self.storagedir = folder
[docs]
def set_anadir(self,anadir:str=None):
if anadir is not None:
if anadir[-1]!='/':
anadir+='/'
self.anadir=anadir
[docs]
def set_seqs(self,seqs:list=[]):
self.seqs=seqs
[docs]
def get_rawfolder(self):
return self.rawfolder
[docs]
def get_storagedir(self):
return self.storagedir
[docs]
def get_seqs(self):
return self.seqs
[docs]
def get_anadir(self):
return self.anadir
[docs]
def get_current_cfg(self):
return self.current_cfg()
[docs]
def get_outfile(self):
if self.get_current_cfg() is not None:
return self.get_current_cfg().get_outfile()
else:
return None
[docs]
def init_dianacaller(self,verbosity=1,name='MultiprocessDiana'):
self.dianacaller = multp.multiprocess_diana(verbosity=verbosity,name=name)
[docs]
def init_filewiki(self,filewiki:str = 'filewiki.csv',recursive:bool=False,depth:int=-1,fields:list=None):
self.wiki = fhc.FileSelector(rawfolder=self.get_rawfolder(),
storagedir=self.get_storagedir(),
filewiki=filewiki,
recursive=recursive,
depth=depth,
fields=fields)
[docs]
def check_existance(self,path):
return os.path.exists(path)
[docs]
def make_folder(self,path):
if self.check_existance(path):
return False
os.makedirs(path)
[docs]
def make_folders(self):
#make storagedir
self.make_folder(self.get_storagedir())
print(self.get_storagedir(),"Exists? ",self.check_existance(self.get_storagedir()))
#make anadir
self.make_folder(self.get_anadir())
print(self.get_anadir(),"Exists? ",self.check_existance(self.get_anadir()))
[docs]
def init_wikimanager(self,filewiki:str):
self.wiki_manager=fhc.WikiManager(storagedir=self.get_storagedir(),filewiki =filewiki)
[docs]
def get_wikimanager(self):
return self.wiki_manager
[docs]
def failed(self):
if self.dianacaller.get_failed() is not None:
if len(self.dianacaller.get_failed())>0:
for ff in self.dianacaller.get_failed():
warnings.warn(f"Failed {self.dianacaller.get_return()[ff]}")
return True
self.dianacaller.clear()
return False
[docs]
def pre_analysis(self):
self.make_folders()
self.init_filewiki()
if self.all_seqs[0] in self.seqs:
print(f"Running {self.all_seqs[0]}")
if not self.wiki.save_file_wiki(mode='a'):
warnings.warn(f"Failed {self.all_seqs[0]}")
return False
print(f"Done {self.all_seqs[0]}")
self.init_wikimanager(self.wiki.get_filewiki())
return True
[docs]
def reader(self):
raise NotImplementedError()
return True
[docs]
def preprocess(self):
raise NotImplementedError()
return True
[docs]
def analysis(self):
return True
[docs]
def post_analysis(self):
return True
[docs]
def run(self):
if not self.pre_analysis():
warnings.warn("Failed preanalysis")
return False
if not self.analysis():
return False
if not self.post_analysis():
return False
return True