Source code for nrgpy.convert.convert_rwd

try:
    from nrgpy import logger
except ImportError:
    pass
import datetime
import time
import os
import subprocess
import shutil
import traceback
from nrgpy.utils.utilities import check_platform, windows_folder_path, linux_folder_path, affirm_directory, count_files, draw_progress_bar


[docs]class local(object): """nrgpy.convert_rwd.local - use local installation of Symphonie Data Retriever (SDR) to convert *.RWD files to *.TXT Parameters ---------- filename : str if populated, a single file is exported encryption_pin : str four digit pin, only used for encrypted files sdr_path : str r'"C:/NRG/SymDR/SDR.exe"', may be any path file_filter : str filters files on text in filename rwd_dir : str (path-like) folder to check for RWD files out_dir : str (path-like) folder to save exported TXT files into wine_folder : str '~/.wine/drive_c/', for linux installations use_site_file : bool set to True to use local site file raw_mode : bool set to True to convert raw counts and voltages progress_bar : bool set to False to see individual file conversions show_result : bool set to False to hide prints to console Returns ------- None Examples -------- Convert a folder of RWD files to Text with SymphoniePRO Desktop Software >>> from nrgpy.convert_rwd import local >>> converter = local( rwd_dir='/path/to/rwd/files', out_dir=/path/to/txt/outputs, file_filter='1234202001', # for files from January 2020 ) >>> converter.convert() Convert a folder ... on Linux; this assumes you followed the instructions in the '''SDR_Linux_README.md''' file. >>> import getpass >>> import nrgpy >>> username = getpass.getuser() >>> rwd_dir = f"/home/{username}/data/symplus3/raw" >>> txt_dir = f"/home/{username}/data/symplus3/export" >>> wine_folder = f"/home/{username}/prefix32/drive_c" >>> converter = nrgpy.convert_rwd.local( rwd_dir=rwd_dir, out_dir=txt_dir, wine_folder=wine_folder ) >>> converter.convert() """ def __init__(self, rwd_dir='', out_dir='', filename='', encryption_pin='', sdr_path=r'C:/NRG/SymDR/SDR.exe', convert_type='meas', file_filter='', wine_folder='~/.wine/drive_c/', use_site_file=False, raw_mode=False, progress_bar=True, show_result=True, **kwargs): if encryption_pin != '': self.command_switch = '/z' # noqueue with pin else: self.command_switch = '/q' # noqueue (logger params) if use_site_file: self.command_switch = '/s' # silent (site file params) if raw_mode: self.command_switch = '/r' # silent (site file params) self.filename = filename self.progress_bar = progress_bar self.encryption_pin = encryption_pin self.sdr_path = windows_folder_path(sdr_path)[:-1] self.root_folder = "\\".join(self.sdr_path.split('\\')[:-2]) self.RawData = self.root_folder + '\\RawData\\' self.ScaledData = self.root_folder + '\\ScaledData\\' self.file_filter = file_filter if 'site_filter' in kwargs and file_filter == '': self.file_filter = kwargs.get('site_filter') self.rwd_dir = rwd_dir # windows_folder_path(rwd_dir) # rwd_dir must be in Windows format, even if using Wine self.show_result = show_result self.platform = check_platform() self.wine_folder = wine_folder self.check_sdr() logger.info("conversion initialized") if self.platform == 'win32': self.out_dir = windows_folder_path(out_dir) self.file_path_joiner = '\\' else: self.file_path_joiner = '/' self.out_dir = out_dir if self.filename: self.counter = 1 self.rwd_dir = os.path.dirname(self.filename) self.file_filter = os.path.basename(self.filename) self.convert()
[docs] def check_sdr(self): """determine if SDR is installed""" if self.platform == 'win32': # do the windows check try: os.path.exists(self.sdr_path) self.sdr_ok = True except: self.sdr_ok = False print('SDR not installed. Please install SDR or check path.\nhttps://www.nrgsystems.com/support/product-support/software/symphonie-data-retriever-software') logger.error("SDR not installed, aborting") else: # do the linux check try: subprocess.check_output(['wine','--version']) except NotADirectoryError: print('System not configured for running SDR.\n Please follow instructions in SDR_Linux_README.md to enable.') logger.error('System not configured for running SDR.\n Please follow instructions in SDR_Linux_README.md to enable.') try: subprocess.check_output(['wine',self.sdr_path,'/s','test.rwd']) affirm_directory(os.path.join(self.wine_folder, "NRG/ScaledData")) self.sdr_ok = True os.remove(os.path.join(self.wine_folder, "NRG/ScaledData/test.log")) except: self.sdr_ok = False print('SDR unable to start') print(traceback.format_exc())
[docs] def convert(self): """process rwd files create list of RWD files that match filtering copy RWD files to RawData directory iterate through files """ affirm_directory(self.out_dir) self._list_files() self._copy_rwd_files() self.raw_count = len(self.rwd_file_list) self.pad = len(str(self.raw_count)) + 1 self.counter = 1 self.convert_time = time.time() self.start_time = datetime.datetime.now() for f in sorted(self.rwd_file_list): site_num = f[:4] try: self._filename = "\\".join([self.RawData+site_num, f]) self._single_file() except: print('file conversion failed on {}'.format(self._filename)) self.counter += 1 if self.raw_count > 1: txt_count = count_files(self.out_dir, self.file_filter.split(".")[0], 'txt', start_time=self.convert_time) log_count, log_files = count_files(self.out_dir, self.file_filter, 'log', show_files=True, start_time=self.convert_time) print('\n\nRWDs in : {}'.format(self.raw_count)) print('TXTs out : {}'.format(txt_count)) print('LOGs out : {}'.format(log_count)) if len(log_files) > 0: print('Log files created:') for _filename in log_files: print('\t{}'.format(_filename)) print('----------------\nDifference : {}'.format(self.raw_count - (txt_count + log_count)))
def _list_files(self): """get list of files in rwd_dir""" self.dir_paths = [] self.rwd_file_list = [] if self.platform == 'win32': walk_path = self.rwd_dir else: walk_path = linux_folder_path(self.rwd_dir) for dirpath, subdirs, files in os.walk(walk_path): self.dir_paths.append(dirpath) for x in files: if x.startswith(self.file_filter) and x.lower().endswith('rwd'): self.rwd_file_list.append(x) def _single_file(self): """process for converting a single file""" _f = self._filename if self.platform == 'linux': self.sdr_path = windows_folder_path(self.sdr_path)[:-1] _f = windows_folder_path(_f)[:-1] wine = 'wine' else: wine = '' self.cmd = [wine, '"'+self.sdr_path+'"', self.command_switch, self.encryption_pin, '"'+_f+'"'] try: if self.show_result: if self.progress_bar: draw_progress_bar(self.counter, self.raw_count, self.start_time) else: print("Converting {0}/{1} {2} ... ".format(str(self.counter).rjust(self.pad), str(self.raw_count).ljust(self.pad), _f.split("\\")[-1]), end="", flush=True) subprocess.check_output(" ".join(self.cmd), shell=True) # subprocess.run(" ".join(self.cmd), stdout=subprocess.PIPE) if not self.progress_bar and not self.show_result: print("[DONE]") try: self._copy_txt_file() except: print('unable to copy {} to text folder'.format(_f)) except: if not self.progress_bar and not self.show_result: print("[FAILED]") import traceback print(traceback.format_exc()) def _copy_rwd_files(self): """copy RWD files from self.RawData to self.rwd_dir""" for f in sorted(self.rwd_file_list): if self.file_filter in f: site_num = f[:4] site_folder = os.path.join(self.RawData, site_num) if self.platform == 'linux': site_folder = ''.join([self.wine_folder, '/NRG/RawData/', site_num]) try: affirm_directory(site_folder) except: print("couldn't create {}".format(site_folder)) pass try: # shutil.copy(os.path.join(self.rwd_dir, f), os.path.join(site_folder)) shutil.copy(self.rwd_dir + f, os.path.join(site_folder)) except: import traceback print('unable to copy file to RawData folder: {}'.format(f)) print(traceback.format_exc()) print(os.path.join(self.rwd_dir, f)) print(os.path.join(site_folder)) def _copy_txt_file(self): """copy TXT file from self.ScaledData to self.out_dir""" try: txt_file_name = os.path.basename(self._filename)[:-4] + '.txt' txt_file_path = os.path.join(self.ScaledData, txt_file_name) out_path = self.file_path_joiner.join([self.out_dir, txt_file_name]) except: print("could not do the needful") import traceback print(traceback.format_exc()) if self.platform == 'linux': out_path = linux_folder_path(self.out_dir) + txt_file_name.split("\\")[-1] txt_file_path = ''.join([self.wine_folder, '/NRG/ScaledData/', txt_file_name.split("\\")[-1]]) self.txt_file_path = txt_file_path try: shutil.copy(txt_file_path, out_path) try: os.remove(txt_file_path) except: print("{0} remains in {1}".format(txt_file_name, self.ScaledData)) except: import traceback print(traceback.format_exc()) print("Unable to copy {0} to {1}".format(txt_file_name,self.out_dir))