###################################################################### # Copyright Trimble 2018-2021 # # Various utilities to control the Spirent RF playback GSS6450: # - ping Spirent external USB drive to keep it from spinning down # - control RF switch serial port for live-sky or playback modes # - SSH into GSS6450 and copy files, start playback, etc. ###################################################################### import serial import paramiko import cryptography.utils import warnings warnings.simplefilter("ignore", cryptography.utils.CryptographyDeprecationWarning) import time from datetime import datetime import os import multiprocessing import sys import atexit import json import subprocess from collections import namedtuple from six import print_, StringIO from RunConfig import get_config_xml from functools import lru_cache import requests from lxml import etree import re # Global files created if logging is enabled # SpirentTools() = regression_progress.json & regression_diag.txt # SpirentSerial() = serial_status.json cfg = get_config_xml(None,True) # Global info on Spirent system spirent_IP = cfg.spirent_IP spirent_password = cfg.spirent_password spirent_username = cfg.spirent_username serial_port_name = cfg.spirent_serial transfer_speeds = cfg.spirent_transfer_speeds # One drive has problems if you don't ping it periodically... ping_usb_drives = cfg.spirent_ping_dirs # The following seem hardcoded on the GSS6450 remote_int_drive = '/home/spirent' # internal SSD # sftp is pretty slow with the default cipher, so choose a faster one # to get double the copy speed sftp_cmd = 'sftp -c aes256-ctr' # NOTE: could replace these with web interface calls shm_put = "/home/spirent/Projects/App/shm_put " shm_get = "/home/spirent/Projects/App/shm_get " def raw_connect_ssh(timeout=60): '''Return ssh connection to Spirent RF playback.''' ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(spirent_IP, username=spirent_username, password=spirent_password, timeout=timeout) return ssh def raw_ssh_cmd(ssh, cmd, sudo=False, timeout=120): '''Run 'cmd' over ssh connection. Return raw readlines()''' ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command(cmd, get_pty=True) if sudo: ssh_stdin.write(spirent_password + '\n') ssh_stdin.flush() if timeout is not None: t_end = time.time() + timeout while not ssh_stdout.channel.exit_status_ready(): time.sleep(0.01) if time.time() >= t_end: raise RuntimeError("SSH cmd '%s' timed out" % cmd) if ssh_stdout.channel.recv_exit_status() != 0: raise RuntimeError("SSH cmd '%s' failed" % cmd) lines = ssh_stdout.readlines() if sudo: lines = lines[2:] # strip off password stuff return lines def bg_copy( target_dir, copy_list ): '''In the background, copy files to 2nd SSD while the 1st SSD is playing back RF samples. Only runs on systems with a 2nd playback SSD. target_dir = something like /media/RAID0/Data or /media/External_SSD/Data copy_list = something like ['/net/fermion/.../file1.A.gns',...] ''' lcl_st = SpirentTools(False) target_drive = re.sub(r"/Data.*$", "", target_dir) if lcl_st.get_remote_ext_drive() == target_drive: target_device = None # no need to remount - already read/write else: lcl_st.construct_drive_info() usb_dev_index = list(lcl_st.remote_usb_info.values()).index(target_drive) target_device = list(lcl_st.remote_usb_info.keys())[usb_dev_index] # sanity check - only copy missing files local_copy_list = lcl_st.get_missing_RF_files( copy_list, target_dir ) if target_device is not None: lcl_st.do_ssh_cmd("mount -o remount,rw %s %s" % (target_device,target_drive), sudo=True) lcl_st.copy_missing_RF_files( target_dir, local_copy_list ) if target_device is not None: lcl_st.do_ssh_cmd("mount -o remount,ro %s %s" % (target_device,target_drive), sudo=True) class SpirentSerial(object): def __init__(self): self.Serial_Level_Spirent_Receivers = 0 self.Serial_Level_Antenna_Spirent = 1 if serial_port_name is None: return self.serial_port=serial.Serial(serial_port_name) self.serial_port.setRTS(self.Serial_Level_Spirent_Receivers) d = { 'level' : self.Serial_Level_Spirent_Receivers, 'status' : "Spirent <-> receivers" } with open("serial_status.json","w") as f: json.dump( d, f ) def toggle_serial(self,desired_level=None): """A serial port controls if the Spirent is connected to the antenna (for logging) or to the receivers (for playback). Input: desired_level = None -> toggle current setting Serial_Level_* -> force to given level""" if serial_port_name is None: return d = None with open("serial_status.json") as f: d = json.load( f ) if desired_level is None: if d['level'] == self.Serial_Level_Antenna_Spirent: level = self.Serial_Level_Spirent_Receivers else: level = self.Serial_Level_Antenna_Spirent else: level = desired_level if level == self.Serial_Level_Spirent_Receivers: desc = "Spirent <-> receivers" else: desc = "Antenna <-> Spirent" d = { 'level' : level, 'status' : desc } with open("serial_status.json","w") as f: json.dump( d, f ) self.serial_port.setRTS(level) print("Set %s to %d"%(serial_port_name,level)) # NOTE: don't close the serial port or the RTS setting is cleared class LogInfo(object): def __init__(self, log_to_file): """Print info to screen and optionally a file for the web/ app.""" self.info = {} self.log_to_file = log_to_file if log_to_file: if os.path.isfile("regression_diag.txt"): # Temporarily save old log for debugging. os.rename("regression_diag.txt","regression_diag.txt.bak") self.diag_f = open("regression_diag.txt","w") else: self.diag_f = sys.stdout def close(self): if self.diag_f != sys.stdout: self.diag_f.close() def raise_(self, *args ): self.print_(args) raise RuntimeError(args) def print_(self, *args ): log_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f') + ": " + " ".join(map(str,args)) print_( log_str, file=self.diag_f, flush=True ) self.info['detail'] = log_str if self.log_to_file: with open("regression_progress.json","w") as f: json.dump( self.info, f ) class SpirentPing(multiprocessing.Process): def __init__(self): '''Class to keep Spirent RF USB drive alive''' print("Start SpirentPing %d"%len(ping_usb_drives)) if len(ping_usb_drives) == 0: return multiprocessing.Process.__init__(self) self.cnt = multiprocessing.Value('i', 0) atexit.register( self.terminate ) self.start() def run(self): '''Ping USB drive to keep it awake.''' print("Start2 SpirentPing %d"%len(ping_usb_drives)) while True: self.cnt.value += 1 time.sleep(60) try: ssh = raw_connect_ssh() for drive in ping_usb_drives: x= raw_ssh_cmd(ssh, "ls -la "+drive+'/Data/') print("Pinged %s - len %d"% (drive, len(x))) f = open('ping_debug.txt','w') f.write('%d\n'%self.cnt.value) f.close() ssh.close() except: print("Couldn't ping USB drive to keep it alive...") pass class SpirentTools(object): def __init__(self,do_log): '''Class to manage Spirent RF control. If do_log is True, log diagnostics to file - see LogInfo''' self.loginfo = LogInfo(do_log) self.ssh = None self.remote_usb_info = {} self.remote_usb_drives = {} self.copy_speed = {} self.last_spin_time = None self.single_playback = [] self.dual_playback = [] self.target_dir = None self.copy_list = None self.bg_copy_proc = None def construct_drive_info(self): '''Figure out /dev/mdX <-> /media/dir_name mapping.''' if len(self.remote_usb_drives) > 0: return raw_data = self.do_ssh_cmd("lsblk -lo NAME,TYPE,UUID,MOUNTPOINT | grep raid0") #raw_data = [] # uncomment to simulate having no USB drives for line in raw_data: w=line.rstrip().split() dev='/dev/'+w[0] if len(w)==4: drive=w[3] else: drive = self.do_ssh_cmd("udisksctl info -b %s | grep Configuration | sed -e \"s/.*'dir': run as root. log=True -> Record cmd to log''' self.connect_ssh() if sudo: cmd = "sudo " + cmd if log: self.loginfo.print_("ssh cmd: %s" % cmd) data = raw_ssh_cmd( self.ssh, cmd, sudo=sudo, timeout=timeout ) if sudo: # raw_ssh_cmd() drops channel in 'sudo' mode self.ssh = None return data def get_remote_ext_drive(self): # See _cached_get_remote_ext_drive() - this just caches the last # results for up to 60 seconds. return self._cached_get_remote_ext_drive(ttl=round(time.time()/60.)) @lru_cache() def _cached_get_remote_ext_drive(self, ttl=None): # All drives in /media should be: # - USB RAID0 drives # - the removable SSD -> find and return this one # Example return: "/media/External_SSD" raw_data = self.do_ssh_cmd("lsblk -lo NAME,TYPE,MOUNTPOINT | grep /media | grep -v raid0") for line in raw_data: w = line.rstrip().split() return w[-1] raise RuntimeError("Couldn't find removable SSD drive") def get_file_info(self): # See _cached_get_file_info() - this just caches the last # results for up to 60 seconds. return self._cached_get_file_info(ttl=round(time.time()/60.)) @lru_cache() def _cached_get_file_info(self, ttl=None): '''Return dict of dicts with info on all files: db[source][filename] = [namedtuple .size, .device, .path, .locked, .play_drive] Returns: f_db[][] where: source = 'int' = internal SSD, 'ext' = external SSD, 'play_usb' = playback USB (if present) 'usb' = USB storage (not for playback) filename = base filename, e.g. 210222185526.A.gns ''' self.construct_drive_info() playback_drives = [remote_int_drive, self.get_remote_ext_drive()] if cfg.spirent_playback_drive is not None: playback_drives.append( cfg.spirent_playback_drive ) def get_file_list(drive, device): path = drive + '/Data' lines = self.do_ssh_cmd("ls -l "+path) Elem = namedtuple('elem', ['size', 'device', 'path', 'locked', 'play_drive']) file_list = {} lock_files = [] if drive == remote_int_drive: source = 'int' elif drive == self.get_remote_ext_drive(): source = 'ext' elif drive == cfg.spirent_playback_drive: source = 'play_usb' else: source = 'usb' if drive in playback_drives: play_drive = drive else: play_drive = None for l in lines: l = l.rstrip() if l.endswith('gns') or l.endswith('scn'): w = l.split() filename = w[-1] file_size = int(w[4]) file_list[filename] = Elem( size=file_size, device=device, path=path, play_drive=play_drive, locked=False ) elif l.endswith('lock'): w = l.split() file_prefix = w[-1].split('.')[0] lock_files.append(file_prefix) # Go back and set "locked" element in d[] as needed for file_prefix in lock_files: for filename in file_list.keys(): if filename.startswith(file_prefix): file_list[filename] = file_list[filename]._replace(locked=True) return source, file_list # Create f_db dict in order, so someone searching f_db.values() # will find fastest versions of RF files first. f_db = {} f_db['int'] = {} f_db['ext'] = {} f_db['play_usb'] = {} f_db['usb'] = {} for dev,drive in self.remote_usb_info.items(): src, info = get_file_list(drive,dev) if not src in f_db: f_db[src] = {} f_db[src].update(info) src, info = get_file_list(self.get_remote_ext_drive(),'ssd') f_db[src] = info src,info = get_file_list(remote_int_drive,'ssd') f_db[src] = info return f_db def quick_stop(self): '''Stop any Spirent RF playback, but don't wait for a response. When this function exits the playback may still be trying to stop. ''' self.do_ssh_cmd(shm_put+"-wS") def stop_any_spirent_test(self): '''Stop any Spirent RF playback and wait until it is fully stopped.''' self.loginfo.print_("Stop any playback and wait until stopped") self.do_ssh_cmd(shm_put+"-wS") n = 0 while True: time.sleep(1) info = self.do_ssh_cmd(shm_get+"-m") if info[0].startswith("-m STOPPED"): break n += 1 time.sleep(1) if n >= 20: self.loginfo.raise_("Couldn't stop playback") time.sleep(5) # without this delay we seem to get an error? def switch_spirent_playback_drive(self, new_drive): '''Switch to another playback drive, e.g., /media/GSS6450''' self.loginfo.print_("Switch media and wait 15s for update") if new_drive == remote_int_drive: self.do_ssh_cmd(shm_put+"-D"+new_drive+" -M1 -wM") else: # Everything but internal uses "-M2" self.do_ssh_cmd(shm_put+"-D"+new_drive+" -M2 -wM") time.sleep(15) info = self.do_ssh_cmd(shm_get+"-D") if not info[0].startswith("Datapath: "+new_drive): self.loginfo.raise_("Couldn't switch media1: " + info[0]) info = self.do_ssh_cmd(shm_get+"-M") if new_drive == remote_int_drive: target_result = "-M 1" else: target_result = "-M 2" if not info[0].startswith(target_result): self.loginfo.raise_("Couldn't switch media2: " + "".join(info)) def get_free_bytes( self, check_dir, cleanup ): '''Get free space on 'check_dir' after removing all non-locked files. ''' info = self.do_ssh_cmd("df -B 1 -P %s | awk 'NR==2 {print $4}'" % check_dir) free_bytes = int(info[0]) files = self.do_ssh_cmd("find %s -type f" % check_dir) files = self.drop_locked_files( files ) if not cleanup: return free_bytes for f in files: f_len = int(self.do_ssh_cmd("stat --printf='%%s' %s"%f.rstrip())[0]) free_bytes += f_len return free_bytes def make_space_for_files( self, free_bytes_needed, check_dir, fake=False ): '''Make sure we have at least 'free_bytes_needed' bytes in directory 'check_dir'. If fake is True, don't actually delete any files. For example, to free ~10KB in the main playback directory: make_space_for_files( 10000, playback_drive+'/Data' )''' info = self.do_ssh_cmd("df -B 1 -P %s | awk 'NR==2 {print $4}'" % check_dir) free_bytes = int(info[0]) toGB = 1024*1024*1024.0 self.loginfo.print_("Free space %d(%f[GB]). Need %d(%f[GB])" % (free_bytes, free_bytes/toGB, free_bytes_needed, free_bytes_needed/toGB)) if free_bytes < free_bytes_needed: files = self.do_ssh_cmd("find %s/*.*" % check_dir) files = self.drop_locked_files( files ) db = {} for f in files: f_len = int(self.do_ssh_cmd("stat --printf='%%s' %s"%f.rstrip())[0]) f_base = f[:f.index('.')] db.setdefault(f_base,0) db[f_base] += f_len sorted_db = sorted(db.items(), key=lambda kv:kv[1]) print(sorted_db) for f,f_len in sorted_db: if fake: print("fake: rm %s.*" % f) else: self.do_ssh_cmd("rm %s.*" % f) free_bytes += f_len if free_bytes > free_bytes_needed: break if free_bytes < free_bytes_needed: self.loginfo.print_("After cleanning. Free space %d(%f[GB]). Need %d(%f[GB])" % (free_bytes, free_bytes/toGB, free_bytes_needed, free_bytes_needed/toGB)) self.loginfo.raise_("Couldn't free up enough space") self.loginfo.print_("After: free space %d, need %d" % (free_bytes, free_bytes_needed)) def compare_local_remote_files(self, local_file_list, check_dir ): '''Compare 'local_file_list' to files in 'check_dir' and return list of missing files. Example: compare_local_remote_files(['/mnt/data_drive/samples_2017_09_11_DHS/2017_09_11/conv/20170911_test01_L1.scn'],playback_drive+'/Data') ''' self.connect_ssh() remote_info = self.get_file_info() def get_remote_len(filename,path): for x in remote_info.values(): if filename in x and x[filename].path==path: return x[filename].size return -1 local_files_to_copy = [] for local_file in local_file_list: local_stat = None if os.path.isfile(local_file): local_stat = os.stat(local_file) copy_file = True remote_filename = os.path.basename(local_file) remote_len = get_remote_len( remote_filename, check_dir ) if local_stat == None or remote_len == local_stat.st_size: copy_file = False else: if remote_len < 0: self.loginfo.print_("Remote file {} does not exist".format(remote_filename)) else: self.loginfo.print_("Remote file {} is the wrong size".format(remote_filename)) if copy_file: local_files_to_copy.append(local_file) return local_files_to_copy def drop_locked_files( self, files ): '''files = list of files like ['/dir1/file.A.gns\n','/dir2/file2.scn\n',...] We can lock scenarios by manually adding a ".lock" file. For example: /dir1/../file.lock protects: /dir1/../file.* This function removes all locked files from the return value so we don't consider them for deletion. ''' locked_files = {} for full_path in files: filename = full_path.rstrip().split('/')[-1] if filename.endswith('.lock'): file_no_ext = filename[:-5] locked_files[file_no_ext] = 1 out_files = [] for full_path in files: filename = full_path.rstrip().split('/')[-1] file_no_ext = filename.split('.')[0] if file_no_ext in locked_files: continue out_files.append( full_path ) return out_files def _get_playback_info( self, info, f_db ): '''For given scenario(info), return info like where the RF file is and how long it will take to copy. Inputs: info = ScenarioInfo() f_db = from self.get_file_info() Returns namedtuple: (.locked = is the file locked on the drive and can't be deleted? .play_drive = what drive is the RF file on? (e.g., /media/RAID0/Data) .len_bytes = length of RF data [bytes] .copy_secs = approx time to copy RF data [secs] .info = ScenarioInfo() ) ''' Elem = namedtuple('elem', ['locked', 'play_drive', 'len_bytes', 'copy_secs', 'info']) try: file_list_len = [os.stat(f).st_size for f in info.file_list] except: file_list_len = info.file_len_list for f_test in f_db.values(): found = True locked = False play_drive = None copy_secs = 0. len_bytes = 0 for filename,server_len in zip(info.file_list,file_list_len): short_filename = os.path.basename(filename) try: f_test_len = f_test[short_filename].size except: f_test_len = -1 if server_len != f_test_len: found = False break locked = f_test[short_filename].locked play_drive = f_test[short_filename].play_drive len_bytes += server_len if play_drive is None: copy_secs += server_len / self.copy_speed[f_test[short_filename].device] if found: return Elem(locked,play_drive,len_bytes,copy_secs,info) # If we get here, the file is not cached on the USB drive and # must be copied over the network. network_copy_secs = sum( file_list_len ) / self.copy_speed['network'] return Elem(False,None,sum( file_list_len ),network_copy_secs,info) def _do_schedule( self, playback_list ): '''Input: playback_list=list of get_file_info() elements Return: (list of commands for playback, approx total copy time [secs]) commands are a namedtuple with: .play_from = drive to start playback on, e.g. /media/External_SSD .play_list = elements from get_file_info() .copy_to = drive to start copy, e.g, /media/some_other_SSD .copy_list = elements from get_file_info() Try to schedule the fastest playback when we have multiple SSD drives. The scheduling algorithm is really a guess, so please feel free to improve. Here's an example of what 2 SSDs could do in parallel: Scenario 1: play time 1 hour, copy time 1 hour Scenario 2: play time 0.5 hours, copy time 0 hours (already on SSD2) time 0: start copy of Scenario 1 to SSD1 start play of Scenario 2 from SSD2 time 0.5 hour: Scenario 2 playback finishes time 1 hour: start Scenario 1 playback from SSD1 ''' # Partition data into: # on_int_ssd = on internal SSD. Should not copy to this drive. # on_ext_ssd = on external SSD. # on_play_usb = on playback SSD USB. # on_usb = on slow HDD USB. Can only copy from here, not playback on_int_ssd = [] on_ext_ssd = [] on_play_usb = [] on_usb = [] remote_ext_drive = self.get_remote_ext_drive() play_usb_drive = cfg.spirent_playback_drive for item in playback_list: if item.play_drive == remote_int_drive: on_int_ssd.append(item) elif item.play_drive == remote_ext_drive: on_ext_ssd.append(item) elif item.play_drive == play_usb_drive: on_play_usb.append(item) else: on_usb.append(item) # Sort items already on playback drives, putting locked items last def sort_p(x): if x.locked: return 1e9 + x.info.play_secs else: return x.info.play_secs sorted_on_int_ssd = sorted(on_int_ssd,key=sort_p) sorted_on_ext_ssd = sorted(on_ext_ssd,key=sort_p) sorted_on_play_usb = sorted(on_play_usb,key=sort_p) # Try both a forward and reverse time sort because either could win. # It'd be nice to figure out a more robust algorithm. def sort_c(x): return x.copy_secs sorted_on_usb = sorted(on_usb, key=sort_c) rev_sorted_on_usb = sorted(on_usb, key=sort_c, reverse=True) to_do, t = self._do_schedule_1pass(list(sorted_on_int_ssd), list(sorted_on_ext_ssd), list(sorted_on_play_usb), sorted_on_usb) rev_to_do, rev_t = self._do_schedule_1pass(list(sorted_on_int_ssd), list(sorted_on_ext_ssd), list(sorted_on_play_usb), rev_sorted_on_usb) if t <= rev_t: sel_to_do, sel_t = to_do, t else: sel_to_do, sel_t = rev_to_do, rev_t secs_to_hrs = 1./60./60. self.loginfo.print_('Added copy hours: %.1f'%(sel_t*secs_to_hrs)) for item in sel_to_do: self.loginfo.print_('Play from: %s'%item.play_from) for elem in item.play_list: self.loginfo.print_(' %s %.1f hrs'% (elem.info.fileName, elem.info.play_secs*secs_to_hrs)) if len(item.copy_list) > 0: self.loginfo.print_('Copy to: %s'%item.copy_to) for elem in item.copy_list: self.loginfo.print_(' %s'%elem.info.fileName) return sel_to_do, sel_t def _do_schedule_1pass( self, on_int_ssd, on_ext_ssd, on_play_usb, sorted_on_usb ): '''heuristic to find minimal copy/play time. Just sort by length and add copies as long as we don't overflow the current playback time too much. Inputs: on_int_ssd = list of get_file_info() already on internal SSD on_ext_ssd = list of get_file_info() already on external SSD on_play_usb = list of get_file_info() already on USB playback SSD sorted_on_usb = list of get_file_info() on slow USB storage See _do_schedule() for return values. ''' Item = namedtuple('item', ['play_from', 'play_list', 'copy_to', 'copy_list']) to_do = [] time_int = sum([x.info.play_secs for x in on_int_ssd]) time_ext = sum([x.info.play_secs for x in on_ext_ssd]) time_usb = sum([x.info.play_secs for x in on_play_usb]) remote_ext_drive = self.get_remote_ext_drive() remote_play_usb_drive = cfg.spirent_playback_drive free_bytes_remote_ext_drive = self.get_free_bytes(remote_ext_drive + '/Data',False) free_bytes_play_usb_drive = self.get_free_bytes(remote_play_usb_drive + '/Data',False) # Starting from internal may not be optimal, but otherwise when do # we play internal SSD scenarios? if time_int > 0: curr_drive = remote_int_drive elif time_ext >= time_usb: curr_drive = remote_ext_drive else: curr_drive = remote_play_usb_drive idx = 0 added_copy_secs = 0. base_free_bytes_remote_ext_drive = self.get_free_bytes(remote_ext_drive + '/Data',True) base_free_bytes_play_usb_drive = self.get_free_bytes(remote_play_usb_drive + '/Data',True) while len(on_int_ssd + on_ext_ssd + on_play_usb) > 0 or idx < len(sorted_on_usb): extra_play_secs = 0. if curr_drive == remote_int_drive: to_play = on_int_ssd on_int_ssd = [] else: # Any non-locked files could get removed after a playback, so update space if curr_drive == remote_ext_drive: source_files = on_ext_ssd free_bytes_remote_ext_drive = base_free_bytes_remote_ext_drive else: source_files = on_play_usb free_bytes_play_usb_drive = base_free_bytes_play_usb_drive # Always play all non-locked files to_play = [] while len(source_files) > 0: if source_files[0].locked: break to_play.append(source_files.pop(0)) if len(to_play) == 0 and len(source_files) > 0: # Always be sure to play at least 1 file to_play.append(source_files.pop(0)) extra_play_secs = sum([x.info.play_secs for x in source_files]) play_secs = sum([x.info.play_secs for x in to_play]) if play_secs == 0.: if idx < len(sorted_on_usb): # We must play something the first time. Pick first # scenario... to_play = [sorted_on_usb[idx]] added_copy_secs += to_play[0].copy_secs idx += 1 else: # Nothing to play from USB. Switch playback drive if curr_drive == remote_play_usb_drive: curr_drive = remote_ext_drive else: curr_drive = remote_play_usb_drive continue to_copy = [] if curr_drive == remote_play_usb_drive: # background copy to a drive not being used for playback copy_dest = remote_ext_drive free_bytes = free_bytes_remote_ext_drive else: copy_dest = remote_play_usb_drive free_bytes = free_bytes_play_usb_drive next_copy_secs = 0. curr_copy_secs = 0. next_copy_used_bytes = 0 while idx < len(sorted_on_usb): next_copy_secs += sorted_on_usb[idx].copy_secs next_copy_used_bytes += sorted_on_usb[idx].len_bytes out_of_space = (next_copy_used_bytes > free_bytes) if not out_of_space: while next_copy_secs > play_secs and extra_play_secs > 0: if curr_drive == remote_ext_drive: to_play.append( on_ext_ssd.pop(0) ) else: to_play.append( on_play_usb.pop(0) ) extra_play_secs -= to_play[-1].info.play_secs play_secs += to_play[-1].info.play_secs out_of_time = (next_copy_secs > play_secs + 0.1*60*60) if (len(to_copy) >= 1 and out_of_time) or out_of_space: break if out_of_space: raise RuntimeError("Out of space",sorted_on_usb[idx]) curr_copy_secs += sorted_on_usb[idx].copy_secs to_copy.append(sorted_on_usb[idx]) if copy_dest == remote_play_usb_drive: on_play_usb.insert(0,sorted_on_usb[idx]) elif copy_dest == remote_ext_drive: on_ext_ssd.insert(0,sorted_on_usb[idx]) idx = idx + 1 if curr_copy_secs > play_secs: added_copy_secs += curr_copy_secs - play_secs to_do.append( Item(curr_drive, to_play, copy_dest, to_copy) ) curr_drive = copy_dest return to_do, added_copy_secs def _rm_non_playback_files( self, playback_list, fake=False ): '''Given a list of files we're going to play, remove any non-locked files from the playback drives we're not going to play. This gives us maximum copy space to use. playback_list = list of _get_playback_info() results ''' short_filenames_to_play = [] for item in playback_list: for filename in item.info.file_list: if item.copy_secs > 1e-3: # Partially-copied files are not preserved continue short_filenames_to_play.append( os.path.basename(filename) ) # Make sure 'play_usb' drive is mounted RW. 'ext' is always 'RW'. target_drive = cfg.spirent_playback_drive self.construct_drive_info() usb_dev_index = list(self.remote_usb_info.values()).index(target_drive) target_device = list(self.remote_usb_info.keys())[usb_dev_index] self.do_ssh_cmd("mount -o remount,rw %s %s" % (target_device,target_drive), sudo=True) f_db = self.get_file_info() for src in ['ext','play_usb']: for filename,info in f_db[src].items(): if info.locked: continue if filename not in short_filenames_to_play: if fake: print("TODO: rm %s/%s"%(info.path,filename)) else: self.do_ssh_cmd("rm %s/%s"%(info.path,filename)) # Put 'play_usb' drive back to read-only. self.do_ssh_cmd("mount -o remount,ro %s %s" % (target_device,target_drive), sudo=True) def prepare_runs( self, sample_file_list ): '''Given all scenarios, sort it to produce the best runtime. Return approximate total copy time [secs]. (not playback time!) Only 1 playback drive: -> just playback files on SSDs first. 2 playback drives: -> We're going to be copying while playing back, so figure out how to copy only while doing the playback. ''' f_db = self.get_file_info() playback_list = [] for info in sample_file_list: playback_list.append( self._get_playback_info( info, f_db ) ) if cfg.spirent_playback_drive is None or len(playback_list) <= 1: if len(playback_list) == 1 and cfg.spirent_playback_drive is not None: self.loginfo.print_("Dual_playback system with 1 scenario to play follows single_playback routine without removing unnecessary unlocked files.") # just put things already playable first playback_list.sort( key=lambda k:k.copy_secs ) copy_secs = 0. for elem in playback_list: copy_secs += elem.copy_secs self.single_playback = playback_list else: # Clean playback drives of stuff we're not going to play, # otherwise we may not have room to do copies during scheduling. self._rm_non_playback_files( playback_list ) # do sorting to minimize time when we ping-pong back and forth self.dual_playback, copy_secs = self._do_schedule( playback_list ) return copy_secs def get_sorted_scenario_groups(self): '''Called after prepare_runs(). Return list of [ScenarioInfo] in the order we will run them, e.g.: [ [ScenarioInfo #1, ScenarioInfo #2], [ScenarioInfo #3], [ScenarioInfo #4] ] ''' if len(self.single_playback) > 0: return [[x.info] for x in self.single_playback] else: play_list = [] for item in self.dual_playback: play_list.append( [elem.info for elem in item.play_list] ) return play_list def calc_num_files_to_copy(self): '''Called after prepare_runs(). Figure out how many files need copying for next run. If 0, then we're ready to play right now with no copying. ''' if len(self.single_playback) > 0: elem = self.single_playback[0] file_list = elem.info.file_list playback_drive = elem.play_drive if playback_drive is None: playback_drive = self.get_remote_ext_drive() else: curr_item = self.dual_playback[0] file_list = [] for elem in curr_item.play_list: file_list.extend( elem.info.file_list ) playback_drive = curr_item.play_from self.switch_spirent_playback_drive( playback_drive ) self.target_dir = playback_drive+'/Data' self.copy_list = self.get_missing_RF_files( file_list, self.target_dir ) return len(self.copy_list) def copy_needed_files(self): '''Called after prepare_runs(). Copy the RF files needed for next run. USB drives is ReadOnly most of time If self.target_dir is cfg.spirent_playback_drive on usb drive, eg. /media/RAID5, remount with RW permission before copying missing files ''' if cfg.spirent_playback_drive is not None and self.target_dir == (cfg.spirent_playback_drive+'/Data'): target_drive = cfg.spirent_playback_drive # Find target_device. eg. '/dev/md5' for target_drive = '/media/RAID5' self.construct_drive_info() usb_dev_index = list(self.remote_usb_info.values()).index(target_drive) target_device = list(self.remote_usb_info.keys())[usb_dev_index] self.loginfo.print_("Set %s %s to Read&Write during copying missing files" %(target_device,target_drive)) self.do_ssh_cmd("mount -o remount,rw %s %s" %(target_device,target_drive), sudo=True) self.loginfo.print_("Copy missing RF files to %s for %s:" %( self.target_dir, str(self.copy_list) )) self.copy_missing_RF_files( self.target_dir, self.copy_list ) self.loginfo.print_("Set %s %s to ReadOnly during copying missing files" %(target_device,target_drive)) self.do_ssh_cmd("mount -o remount,ro %s %s" %(target_device,target_drive), sudo=True) else: self.loginfo.print_("Copy missing RF files to %s for %s:" %( self.target_dir, str(self.copy_list) )) self.copy_missing_RF_files( self.target_dir, self.copy_list ) def bg_copy_next_run_in_advance(self): '''Called after prepare_runs(). If we have 2 playback drives: Kick off a task in the background to copy next scenario to playback drive (which runs while current scenario is playing!). Return bg_copy pid Otherwise: do nothing. Return None ''' if len(self.dual_playback) == 0: return None # Figure out what files to copy curr_item = self.dual_playback[0] target_dir = curr_item.copy_to + '/Data' copy_list = [] for elem in curr_item.copy_list: missing_files = self.get_missing_RF_files( elem.info.file_list, target_dir ) copy_list.extend( missing_files ) if len(copy_list) == 0: # Nothing to copy, so exit early return # In background, copy files self.loginfo.print_("Start background copy of files to %s:"%target_dir) for filename in copy_list: self.loginfo.print_(" %s"%filename) proc = multiprocessing.Process(target=bg_copy, args=(target_dir,copy_list,)) proc.start() self.bg_copy_proc = proc return proc.pid def done_with_playback_group(self): '''Finished a playback item in prepare_runs(). Remove it from list to run. ''' if len(self.single_playback) > 0: self.single_playback.pop(0) else: self.dual_playback.pop(0) if self.bg_copy_proc is not None: self.loginfo.print_("Wait for background copy to finish..") self.bg_copy_proc.join() # If bg_copy fails the system will try again later. We # don't need to exit, but it is nice to record what happened. self.loginfo.print_(" background copy exit code", self.bg_copy_proc.exitcode) self.bg_copy_proc = None self.loginfo.print_(" background copy finished") def spirent_copy(self, filename, src_dir, dest_dir): """Copy file from one path to another - all on the Spirent""" if src_dir == dest_dir: raise RuntimeError("Can't copy file on top of itself",src_dir,dest_dir) filename = os.path.basename(filename) # dd seems to run a little faster than cp cmd = "dd bs=256k if='%s' of='%s'" % (src_dir+'/'+filename, dest_dir+'/'+filename) self.do_ssh_cmd(cmd, timeout=None) def get_usb_path_from_filename( self, full_filename ): """Given a full path like /mnt/data_drive/sample_123/123.A.gns, find the USB drive the file resides on (e.g., remote_usb_drives[0]+'/Data') """ self.construct_drive_info() filename = os.path.basename(full_filename) filename_len = os.stat(full_filename).st_size for drive in self.remote_usb_drives: try: f_local = "%s/Data/%s" % (drive,filename) self.do_ssh_cmd("ls " + f_local) # Make sure file is the right size (in case we have a corrupt USB file) f_local_len = int(self.do_ssh_cmd("stat --printf='%%s' %s"%f_local)[0]) if filename_len != f_local_len: continue return drive+'/Data' except: pass raise RuntimeError("Can't find %s on USB drive"%filename) def copy_usb_to_server( self, full_filename ): """full_filename is the name of a (missing) file on a server like fermion.eng.trimble.com (e.g, /mnt/data_drive/sample_123/123.A.gns). Call this if 123.A.gns only resides on the Spirent USB drive and you want to copy it to fermion. """ filename = os.path.basename(full_filename) local_dir = os.path.dirname(full_filename) if not os.path.isdir(local_dir): os.mkdir(local_dir) remote_usb_path = self.get_usb_path_from_filename( filename ) cmd = "sshpass -p %s %s %s@%s:%s" % \ (spirent_password, sftp_cmd, spirent_username, spirent_IP, remote_usb_path+'/'+filename) cmd = "cd %s && %s" % (local_dir, cmd ) subprocess.check_call(cmd, shell=True) def get_usb_device_with_most_space( self ): """Return USB path with the most free space, e.g.: /dev/md0 However, don't include SSD playback drive. """ self.construct_drive_info() if len(self.remote_usb_info.keys()) == 0: return None max_space = 0 max_dev = list(self.remote_usb_info.keys())[0] for dev,drive in self.remote_usb_info.items(): if drive == cfg.spirent_playback_drive: # reserve playback SSD for user-specified data continue info = self.do_ssh_cmd("df -P %s | grep / | awk '{print $4}'" % drive) free_space = int(info[0]) if free_space > max_space: max_space = free_space max_dev = dev return max_dev def get_usb_path_with_most_space( self ): """Return USB path with the most free space, e.g.: /media/c2018175-00e1-47a6-bf4e-560808863bd2/Data """ self.construct_drive_info() dev = self.get_usb_device_with_most_space() return self.remote_usb_info[dev] + '/Data' def copy_internal_ssd_to_usb( self, full_filenames ): """full_filenames is a list of files on the Spirent internal SSD like ['/home/spirent/Data/123.A.gns']. Call this if 123.A.gns only resides on the Spirent internal SSD drive and you want to copy it to the USB drive. """ dest_path = self.get_usb_path_with_most_space() for filename in full_filenames: self.spirent_copy( filename, remote_int_drive+'/Data', dest_path ) def copy_external_ssd_to_usb( self, full_filenames ): """full_filenames is a list of files on the Spirent external SSD like ['/home/GSS6450_Data/Data/123.A.gns']. Call this if 123.A.gns only resides on the Spirent internal SSD drive and you want to copy it to the USB drive. """ dest_path = self.get_usb_path_with_most_space() remote_ext_drive = self.get_remote_ext_drive() for filename in full_filenames: self.spirent_copy( filename, remote_ext_drive+'/Data', dest_path ) def copy_server_to_dir( self, filename, target_dir ): """filename = a file on a server like fermion.eng.trimble.com (e.g, '/mnt/data_drive/sample_123/123.A.gns'). target_dir = path on Spirent, e.g. /media/GSS6450_Data/ Call this if 123.A.gns only resides on the server and you want to copy it to a specific path. """ self.loginfo.print_("Copy from server %s to Spirent dir %s" % (filename,target_dir)) cmd = "echo 'put %s' | sshpass -p %s %s %s@%s:%s" % \ (filename, spirent_password, sftp_cmd, spirent_username, spirent_IP, target_dir) subprocess.check_call(cmd, shell=True) def copy_server_to_usb( self, full_filenames ): """full_filenames is a list of files on a server like fermion.eng.trimble.com (e.g, ['/mnt/data_drive/sample_123/123.A.gns']). Call this if 123.A.gns only resides on the server and you want to copy it to the Spirent USB drive. """ dest_path = self.get_usb_path_with_most_space() for filename in full_filenames: cmd = "echo 'put \"%s\"' | sshpass -p %s %s %s@%s:%s" % \ (filename, spirent_password, sftp_cmd, spirent_username, spirent_IP, dest_path) subprocess.check_call(cmd, shell=True) def get_missing_RF_files(self, local_file_list, target_dir): """Given a list of local PC files(local_file_list) and a remote Spirent RF directory(target_dir), get missing files for 'target_dir' (e.g., playback_drive + '/Data') """ # Check to see what files already exist local_files_to_copy = self.compare_local_remote_files( local_file_list, target_dir ) return local_files_to_copy def copy_missing_RF_files(self, target_dir, local_files_to_copy): """Call this after get_missing_RF_files(). target_dir = same as get_missing_RF_files() local_files_to_copy = return value from get_missing_RF_files() """ # do we have enough space to copy all files in local_files_to_copy? local_total_size = sum([os.stat(x).st_size for x in local_files_to_copy]) self.make_space_for_files( local_total_size, target_dir ) # Copy all needed files to GSS6450 # local copy is 2-2.5x faster than network copy, but allow fallback... for local_file in local_files_to_copy: try: remote_usb_dir = self.get_usb_path_from_filename( local_file ) self.spirent_copy( local_file, remote_usb_dir, target_dir ) except: self.copy_server_to_dir( local_file, target_dir ) def get_mount_usb_status(self): '''Return dictionary of USB drive status, e.g.: {'/dev/md0':'Unmounted', '/dev/md1':'Unmounted'} Status words: Unconnected = not detected. Is cable plugged in? MountedRO = mounted read-only MountedRW = mounted read-write Unmounted = not mounted UnmountedWrongPath = not mounted at the correct location ''' self.construct_drive_info() info_mnt = {} for raid_dev in self.remote_usb_info.keys(): info_mnt[raid_dev] = 'Unconnected' short_dev = raid_dev.strip('/dev/') try: tmp = self.do_ssh_cmd("lsblk -l | grep raid0 | grep %s" % short_dev, log=False) except RuntimeError: continue info_mnt[raid_dev] = 'Unmounted' try: tmp = self.do_ssh_cmd("mount | grep %s" % raid_dev, log=False) if len(tmp) > 0: if tmp[0].find(self.remote_usb_info[raid_dev]) < 0: info_mnt[raid_dev] = 'UnmountedWrongPath' elif tmp[0].find("(ro") > 0: info_mnt[raid_dev] = 'MountedRO' else: info_mnt[raid_dev] = 'MountedRW' except RuntimeError: continue return info_mnt def mount_usb(self, mount, read_only=True): """Input: mount = True -> mount all USB drives on GSS6450 False -> unmount read_only = True -> mount as read-only? (safer) False -> mount read-write Mount or unmount GSS6450 backup USB drives. """ self.construct_drive_info() usb_status = self.get_mount_usb_status() for usb_device, status in usb_status.items(): if status == 'UnmountedWrongPath': # One USB drive sometimes comes up mounted in the wrong # spot, so undo that error before proceeding. self.do_ssh_cmd("umount " + usb_device, sudo=True) status = 'Unmounted' usb_drive = self.remote_usb_info[usb_device] if mount: if read_only: if status=='MountedRW': self.do_ssh_cmd("mount -o remount,ro %s %s" % (usb_device,usb_drive), sudo=True) elif status=='Unmounted': self.do_ssh_cmd("udisksctl mount -b %s" % usb_device, sudo=True) self.do_ssh_cmd("mount -o remount,ro %s %s" % (usb_device,usb_drive), sudo=True) else: if status=='MountedRO': self.do_ssh_cmd("mount -o remount,rw %s %s" % (usb_device,usb_drive), sudo=True) elif status=='Unmounted': self.do_ssh_cmd("udisksctl mount -b %s" % usb_device, sudo=True) else: if status=='MountedRO' or status=='MountedRW': self.do_ssh_cmd("umount " + usb_drive, sudo=True) def reset_offset(self): '''Clear any playback offset''' self.do_ssh_cmd(shm_put+"-o0") def supports_continuous( self ): '''Our older GSS6450 doesn't support continuous playback, so test for the feature.''' raw_data = self.do_ssh_cmd(shm_put+"| grep Continuous | wc -l") if raw_data[0].startswith('0'): return False else: return True def scn_is_continuous( self, scn_file_path ): raw_data = self.do_ssh_cmd("grep 'Continuous Record Mode' %s | wc -l"%scn_file_path) if raw_data[0].startswith('0'): return False else: return True def start_spirent_test(self,test_file_path, offset_sec=0): '''Start playback of file 'test_file_path'. Must be full pathname, e.g.: /media/GSS6450_Data/Data/20170912.scn ''' remote_filename = os.path.basename(test_file_path) flag_chain = '' if self.supports_continuous(): if self.scn_is_continuous(test_file_path): flag_chain="--ContinuousPlayback 1" else: flag_chain="--ContinuousPlayback 0" self.loginfo.print_("Trying to start playback of %s (%s)"% (remote_filename,flag_chain)) if offset_sec > 0: self.do_ssh_cmd("%s-o%d -f%s -wP %s"% (shm_put, offset_sec, remote_filename, flag_chain) ) else: self.do_ssh_cmd("%s-f%s -wP %s"% (shm_put, remote_filename, flag_chain)) n = 0 while True: info = self.do_ssh_cmd(shm_get+"-m") if info[0].startswith("-m PLAYING"): self.loginfo.print_("Started playback") break n += 1 time.sleep(1) if n >= 20: self.loginfo.raise_("Couldn't start playback") def spirent_test_status(self): '''Returns status of GSS6450 playback: PLAYING = currently running STOPPED = done ERROR_RECOVERED = error. Glitch with hard drive. Stopped, but you probably want to try again. ERROR = unknown error. Not sure how to recover... ''' t1=time.time() try: r = requests.get( "http://%s/cgi-bin/realtime_info.cgi"%spirent_IP, timeout=10 ) data = etree.fromstring( r.content ) status = data.find('status') except: self.loginfo.print_("Error connecting to GSS6450") status = None dt = time.time() - t1 if dt > 30: self.loginfo.print_("Long time to get GSS6450 status! %.1f [s]"%dt) if status is None: return "ERROR" elif status.text == 'STOPPED': self.loginfo.print_("Done with test") return "STOPPED" elif status.text == 'PLAYING': return "PLAYING" elif status.text == 'ERROR': self.loginfo.print_("Error with status!") # try to recover try: r = requests.get( "http://%s/cgi-bin/stop.cgi"%spirent_IP, timeout=30 ) time.sleep(10) r = requests.get( "http://%s/cgi-bin/realtime_info.cgi"%spirent_IP, timeout=10 ) data = etree.fromstring( r.content ) status = data.find('status') if status.text == 'STOPPED': return "ERROR_RECOVERED" except: self.loginfo.print_("Couldn't recover") return "ERROR"