Source code for skrf.util

"""

.. currentmodule:: skrf.util
========================================
util (:mod:`skrf.util`)
========================================

Holds utilities that are general conveniences.


Time-related utilities
----------------------
.. autosummary::
   :toctree: generated/

   now_string
   now_string_2_dt

   ProgressBar

Array-related functions
-----------------------
.. autosummary::
   :toctree: generated/

   find_nearest
   find_nearest_index
   has_duplicate_value
   smooth

File-related functions
----------------------
.. autosummary::
   :toctree: generated/

   get_fid
   get_extn
   basename_noext
   git_version
   unique_name
   findReplace
   dict_2_recarray

General Purpose Objects
-----------------------
.. autosummary::
   :toctree: generated/

    HomoList
    HomoDict

"""
import contextlib
import fnmatch
import os
from typing import Iterable, Tuple, List, Union, Any
import warnings
import numpy as npy
from datetime import datetime
import collections
import pprint
import re
from subprocess import Popen, PIPE
import sys
from functools import wraps
from .constants import Number, NumberLike

[docs]def now_string() -> str: ''' Return a unique sortable string, representing the current time. Nice for generating date-time stamps to be used in file-names, the companion function :func:`now_string_2_dt` can be used to read these string back into datetime objects. Returns ------- now : string curent date-time stamps. See Also -------- now_string_2_dt ''' return datetime.now().__str__().replace('-','.').replace(':','.').replace(' ','.')
[docs]def now_string_2_dt(s: str) -> datetime: ''' Converts the output of :func:`now_string` to a datetime object. Parameters ---------- s : str date-time stamps string as generated by :func:`now_string` Returns ------- dt : datetime date-time stamps See Also -------- now_string ''' return datetime(*[int(k) for k in s.split('.')])
[docs]def find_nearest(array: npy.ndarray, value: Number) -> Number: ''' Find the nearest value in array. Parameters ---------- array : numpy.ndarray array we are searching for a value in value : element of the array value to search for Returns -------- found_value : an element of the array the value that is numerically closest to `value` ''' idx = find_nearest_index(array, value) return array[idx]
[docs]def find_nearest_index(array: npy.ndarray, value: Number) -> int: ''' Find the nearest index for a value in array. Parameters ---------- array : numpy.ndarray array we are searching for a value in value : element of the array value to search for Returns -------- found_index : int the index at which the numerically closest element to `value` was found at References ---------- taken from http://stackoverflow.com/questions/2566412/find-nearest-value-in-numpy-array ''' return (npy.abs(array-value)).argmin()
def slice_domain(x: npy.ndarray, domain: tuple): ''' Returns a slice object closest to the `domain` of `x` domain = x[slice_domain(x, (start, stop))] Parameters ---------- vector : numpy.ndarray an array of values domain : tuple tuple of (start,stop) values defining the domain over which to slice Examples -------- >>> x = linspace(0,10,101) >>> idx = slice_domain(x, (2,6)) >>> x[idx] ''' start = find_nearest_index(x, domain[0]) stop = find_nearest_index(x, domain[1]) return slice(start, stop+1) # file IO
[docs]def get_fid(file, *args, **kwargs): ''' Return a file object, given a filename or file object. Useful when you want to allow the arguments of a function to be either files or filenames Parameters ---------- file : str/unicode or file-object file to open \*args, \*\*kwargs : arguments and keyword arguments to `open()` Returns ------- fid : file object ''' if isinstance(file, str): return open(file, *args, **kwargs) else: return file
[docs]def get_extn(filename: str) -> str: ''' Get the extension from a filename. The extension is defined as everything passed the last '.'. Returns None if it ain't got one Parameters ---------- filename : string the filename Returns ------- ext : string, None either the extension (not including '.') or None if there isn't one ''' ext = os.path.splitext(filename)[-1] if len(ext) == 0: return None else: return ext[1:]
[docs]def basename_noext(filename: str) -> str: ''' Get the basename and strips extension. Parameters ---------- filename : string the filename Returns ------- basename : str file basename (ie. without extension) ''' return os.path.splitext(os.path.basename(filename))[0]
# git
[docs]def git_version(modname: str) -> str: ''' Return output 'git describe', executed in a module's root directory. Parameters ---------- modname : str module name Returns ------- out : str output of 'git describe' ''' mod = __import__(modname) mod_dir = os.path.split(mod.__file__)[0] p = Popen(['git', 'describe'], stdout=PIPE, stderr=PIPE, cwd=mod_dir) try: out, er = p.communicate() except(OSError): return None out = out.strip('\n') if out == '': return None return out
[docs]def dict_2_recarray(d: dict, delim: str, dtype: List[Tuple]) -> npy.ndarray: ''' Turn a dictionary of structured keys to a record array of objects. This is useful if you save data-base like meta-data in the form or file-naming conventions, aka 'the poor-mans database' Parameters ---------- d : dict dictionnary of structured keys delim : str delimiter string dtype : list of tuple list of type, where a type is tuple like ('type_name', type) Returns ------- ra : numpy.array Examples -------- Given a directory of networks like: >>> ls a1,0.0,0.0.s1p a1,3.0,3.0.s1p a2,3.0,-3.0.s1p b1,-3.0,3.0.s1p ... you can sort based on the values or each field, after defining their type with `dtype`. The `values` field accesses the objects. >>> d = rf.read_all_networks('/tmp/') >>> delim = ',' >>> dtype = [('name', object), ('voltage', float), ('current', float)] >>> ra = dict_2_recarray(d=rf.ran(dir), delim=delim, dtype =dtype) then you can sift like you do with numpy arrays >>> ra[ra['voltage'] < 3]['values'] array([1-Port Network: 'a2,0.0,-3.0', 450-800 GHz, 101 pts, z0=[ 50.+0.j], 1-Port Network: 'b1,0.0,3.0', 450-800 GHz, 101 pts, z0=[ 50.+0.j], 1-Port Network: 'a1,0.0,-3.0', 450-800 GHz, 101 pts, z0=[ 50.+0.j], ''' split_keys = [tuple(k.split(delim)+[d[k]]) for k in d.keys()] x = npy.array(split_keys, dtype=dtype+[('values',object)]) return x
[docs]def findReplace(directory: str, find: str, replace: str, file_pattern: str): ''' Find/replace some txt in all files in a directory, recursively. This was found in [1]_ . Parameters ---------- directory : str path of a directory find : str pattern to search for replace : str string to replace with file_pattern : str file pattern for filtering. Ex: '\*.txt'. Examples -------- >>> rf.findReplace('some_dir', 'find this', 'replace with this', '*.txt') References ---------- .. [1] http://stackoverflow.com/questions/4205854/python-way-to-recursively-find-and-replace-string-in-text-files ''' for path, dirs, files in os.walk(os.path.abspath(directory)): for filename in fnmatch.filter(files, file_pattern): filepath = os.path.join(path, filename) with open(filepath) as f: s = f.read() s = s.replace(find, replace) with open(filepath, "w") as f: f.write(s)
# general purpose objects
[docs]class HomoList(collections.Sequence): ''' A Homogeneous Sequence. Provides a class for a list-like object which contains homogeneous values. Attributes of the values can be accessed through the attributes of HomoList. Searching is done like numpy arrays. Initialized from a list of all the same type >>> h = HomoDict([Foo(...), Foo(...)]) The individual values of `h` can be access in identical fashion to Lists. >>> h[0] Assuming that `Foo` has property `prop` and function `func` ... Access elements' properties: >>> h.prop Access elements' functions: >>> h.func() Searching: >>> h[h.prop == value] >>> h[h.prop < value] Multiple search: >>> h[set(h.prop==value1) & set( h.prop2==value2)] Combos: >>> h[h.prop==value].func() '''
[docs] def __init__(self, list_): self.store = list(list_)
def __eq__(self, value): return [k for k in range(len(self)) if self.store[k] == value ] def __ne__(self, value): return [k for k in range(len(self)) if self.store[k] != value ] def __gt__(self, value): return [k for k in range(len(self)) if self.store[k] > value ] def __ge__(self, value): return [k for k in range(len(self)) if self.store[k] >= value ] def __lt__(self, value): return [k for k in range(len(self)) if self.store[k] < value ] def __le__(self, value): return [k for k in range(len(self)) if self.store[k] <= value ] def __getattr__(self, name): return self.__class__( [k.__getattribute__(name) for k in self.store]) def __getitem__(self, idx): try: return self.store[idx] except(TypeError): return self.__class__([self.store[k] for k in idx]) def __call__(self, *args, **kwargs): return self.__class__( [k(*args,**kwargs) for k in self.store]) def __setitem__(self, idx, value): self.store[idx] = value def __delitem__(self, idx): del self.store[idx] def __iter__(self): return iter(self.store) def __len__(self): return len(self.store) def __str__(self): return pprint.pformat(self.store) def __repr__(self): return pprint.pformat(self.store)
[docs]class HomoDict(collections.MutableMapping): ''' A Homogeneous Mutable Mapping. Provides a class for a dictionary-like object which contains homogeneous values. Attributes of the values can be accessed through the attributes of HomoDict. Searching is done like numpy arrays. Initialized from a dictionary containing values of all the same type >>> h = HomoDict({'a':Foo(...),'b': Foo(...), 'c':Foo(..)}) The individual values of `h` can be access in identical fashion to Dictionaries. >>> h['key'] Assuming that `Foo` has property `prop` and function `func` ... Access elements' properties: >>> h.prop Access elements' functions: >>> h.func() Searching: >>> h[h.prop == value] >>> h[h.prop < value] Multiple search: >>> h[set(h.prop==value1) & set( h.prop2==value2)] Combos: >>> h[h.prop==value].func() '''
[docs] def __init__(self, dict_): self.store = dict(dict_)
def __eq__(self, value): return [k for k in self.store if self.store[k] == value ] def __ne__(self, value): return [k for k in self.store if self.store[k] != value ] def __gt__(self, value): return [k for k in self.store if self.store[k] > value ] def __ge__(self, value): return [k for k in self.store if self.store[k] >= value ] def __lt__(self, value): return [k for k in self.store if self.store[k] < value ] def __le__(self, value): return [k for k in self.store if self.store[k] <= value ] def __getattr__(self, name): return self.__class__( {k: getattr(self.store[k],name) for k in self.store}) def __getitem__(self, key): if isinstance(key, str): return self.store[key] else: c = self.__class__({k:self.store[k] for k in key}) return c #if len(c) == 1: # return c.store.values()[0] #else: # return c def __call__(self, *args, **kwargs): return self.__class__( {k: self.store[k](*args, **kwargs) for k in self.store}) def __setitem__(self, key, value): self.store[key] = value def __delitem__(self, key): del self.store[key] def __iter__(self): return iter(self.store) def __len__(self): return len(self.store) def __str__(self): return pprint.pformat(self.store) def __repr__(self): return pprint.pformat(self.store)
[docs] def copy(self): return HomoDict(self.store)
[docs] def filter_nones(self): self.store = {k:self.store[k] for k in self.store \ if self.store[k] is not None}
[docs] def filter(self, **kwargs): ''' Filter self based on kwargs This is equivalent to: >>> h = HomoDict(...) >>> for k in kwargs: >>> h = h[k ==kwargs[k]] >>> return h prefixing the kwarg value with a '!' causes a not equal test (!=) Examples ---------- >>> h = HomoDict(...) >>> h.filter(name='jean', age = '18', gender ='!female') ''' a = self for k in kwargs: if kwargs[k][0] == '!': a = a[a.__getattr__(k) != kwargs[k][1:]] else: a = a[a.__getattr__(k) == kwargs[k]] return a
[docs]def has_duplicate_value(value: Any, values: Iterable, index: int) -> Union[bool, int]: """ Check if there is another value of the current index in the list. Parameters ---------- value : Any any value in a list values : Iterable the iterable containing the values index : int the index of the current item we are checking for. Returns ------- index : bool or int returns None if no duplicate found, or the index of the first found duplicate Examples -------- >>> rf.has_duplicate_value(0, [1, 2, 0, 3, 0], -1) # -> 2 >>> rf.has_duplicate_value(0, [1, 2, 0, 3, 0], 2) # -> 4 >>> rf.has_duplicate_value(3, [1, 2, 0, 3, 0], 0) # -> 3 >>> rf.has_duplicate_value(3, [1, 2, 0, 3, 0], 3) # -> False """ for i, val in enumerate(values): if i == index: continue if value == val: return i return False
[docs]def unique_name(name: str, names: list, exclude: int = -1) -> str: """ Pass in a name and a list of names, and increment with _## as necessary to ensure a unique name. Parameters ---------- name : str the chosen name, to be modified if necessary names : list list of names (str) exclude : int, optional the index of an item to be excluded from the search. Default is -1. Returns ------- unique_name : str """ if not has_duplicate_value(name, names, exclude): return name else: if re.match("_\d\d", name[-3:]): name_base = name[:-3] suffix = int(name[-2:]) else: name_base = name suffix = 1 for num in range(suffix, 100, 1): name = "{:s}_{:02d}".format(name_base, num) if has_duplicate_value(name, names, exclude) is False: break return name
[docs]def smooth(x: npy.ndarray, window_len: int = 11, window: str = 'flat') -> npy.ndarray: """ Smooth the data using a window with requested size. Based on the function from the scipy cookbook [#]_ This method is based on the convolution of a scaled window with the signal. The signal is prepared by introducing reflected copies of the signal (with the window size) in both ends so that transient parts are minimized in the beginning and end part of the output signal. Parameters ---------- x : numpy.array the input signal window_len : int, optional the dimension of the smoothing window; should be an odd integer. Default is 11. window : str, optional the type of window from 'flat', 'hanning', 'hamming', 'bartlett', 'blackman' flat window will produce a moving average smoothing. Default is 'flat' Returns ------- y : numpy.array The smoothed signal Examples -------- >>> t = linspace(-2, 2, 0.1) >>> x = sin(t) + randn(len(t))*0.1 >>> y = smooth(x) See Also -------- numpy.hanning, numpy.hamming, numpy.bartlett, numpy.blackman, numpy.convolve scipy.signal.lfilter Note ---- `length(output) != length(input)`. To correct this: `return y[(window_len/2-1):-(window_len/2)]` instead of just `y`. References ---------- .. [#] http://scipy-cookbook.readthedocs.io/items/SignalSmooth.html """ if x.ndim != 1: raise ValueError("smooth only accepts 1 dimension arrays.") if x.size < window_len: raise ValueError("Input vector needs to be bigger than window size.") if window_len < 3: return x if window not in ['flat', 'hanning', 'hamming', 'bartlett', 'blackman']: raise ValueError("Window is one of 'flat', 'hanning', 'hamming', 'bartlett', 'blackman'") s = npy.r_[x[window_len - 1:0:-1], x, x[-2:-window_len - 1:-1]] if window == 'flat': # moving average w = npy.ones(window_len, 'd') else: w = eval('npy.' + window + '(window_len)') y = npy.convolve(w / w.sum(), s, mode='same') return y[window_len-1:-(window_len-1)]
[docs]class ProgressBar: """ A progress bar based off of the notebook/ipython progress bar from PyMC. Useful when waiting for long operations such as taking a large number of VNA measurements that may take a few minutes. Examples -------- >>> from time import sleep >>> pb = rf.ProgressBar(10) >>> for idx in range(10): >>> sleep(1) >>> pb.animate(idx) """
[docs] def __init__(self, iterations: int, label: str = "iterations"): """ Progress bar constructor. Parameters ---------- iterations : int Number of expected iterations label : str, optional Progress bar label, by default "iterations" """ self.iterations = iterations self.label = label self.prog_bar = '[]' self.fill_char = '*' self.width = 50 self.__update_amount(0)
[docs] def animate(self, iteration: int): """ Animate the progress bar. Parameters ---------- iteration : int current iteration """ print('\r', self, end='') sys.stdout.flush() self.update_iteration(iteration + 1)
[docs] def update_iteration(self, elapsed_iter: int): self.__update_amount((elapsed_iter / float(self.iterations)) * 100.0) self.prog_bar += ' %d of %s %s complete' % (elapsed_iter, self.iterations, self.label)
def __update_amount(self, new_amount: int): percent_done = int(round((new_amount / 100.0) * 100.0)) all_full = self.width - 2 num_hashes = int(round((percent_done / 100.0) * all_full)) self.prog_bar = '[' + self.fill_char * num_hashes + ' ' * (all_full - num_hashes) + ']' pct_place = (len(self.prog_bar) // 2) - len(str(percent_done)) pct_string = '%d%%' % percent_done self.prog_bar = self.prog_bar[0:pct_place] + \ (pct_string + self.prog_bar[pct_place + len(pct_string):]) def __str__(self): return str(self.prog_bar)
@contextlib.contextmanager def suppress_numpy_warnings(**kw): olderr = npy.seterr(**kw) yield npy.seterr(**olderr) def suppress_warning_decorator(msg): def suppress_warnings_decorated(func): @wraps(func) def suppressed_func(*k, **kw): show_warnings = [] with warnings.catch_warnings(record=True) as wlist: res = func(*k, **kw) for w in wlist: if not w.message.args[0].startswith(msg): show_warnings.append(w) for w in show_warnings: warnings.warn(w.message.args[0]) return suppressed_func return suppress_warnings_decorated