# -*- coding: utf-8 -*- #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # file sfc.py # purpose Skaperen's Function Collection (and other useful stuff) # # legal Copyright © 2018, 2025, by Philip D. Howard - all rights reserved # license MIT-like, not GPL, see variable name "license" below near line 28 # summary you are obligated to be truthful, not to provide source code # # author 11054987560151472272755686915985840251291393453694611309 # (provu igi la numeron al duuma) # # usage import sfcdebug (optional to enable debugging by default) # import sfc # # or import sfcdebug (optional to enable debugging by default) # from sfc import * (dangerous unless you know what sfc.py defines) # # requires Python 3.6 or higher # # URLs http://ipal.net/python/sfc.py # http://ipal.net/python/sfcdebug.py (separate module to enable debug default mode) # # note developed primarily for BSD and Linux but may also work elsewhere # note tested on Linux (Xubuntu 18.04.5 with Python 3.6, Xubuntu 20.04.6 with Python 3.8) # note code in this file assumes a text screen width of at least 128 character columns #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- """Skaperen's Function Collection (and other useful stuff)""" license = """ Copyright © 2018, 2025 by Phil D. Howard - all rights reserved Permission to use, copy, modify, and/or distribute this software for any purpose with or without fee is hereby granted, provided that the above copyright notice and this permission notice appear in all copies. THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA, OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE, OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. The author may be contacted by decoding the number 11054987560151472272755686915985840251291393453694611309 (provu igi la numeron al duuma) """ _debugmode = True #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- import base64 import collections import copy as mod_copy import cmath import curses import curses.ascii import decimal import fcntl import inspect import io import itertools import ipaddress import json import lzma import math import numbers import os import pathlib import random import signal import socket import stat as mod_stat import struct import subprocess import sys import re import termios import textwrap import time import uuid # check version before trying to do individual imports or anything else if sys.version_info.major<3:raise RuntimeError('Python versions 1.X-2.X are not supported') if sys.version_info.minor<6:raise RuntimeError('Python versions 3.0-3.5 are not supported') if sys.version_info.major>3:raise RuntimeError('Python major versions above 3 are not known') from base64 import b85decode,b85encode,b64decode,b64encode,b32decode,b32encode,b16decode,b16encode from collections import Counter,namedtuple from collections.abc import Iterable from copy import copy,deepcopy from decimal import * from fcntl import ioctl from inspect import currentframe from io import IOBase from itertools import chain from math import * from numbers import Number,Number as Num from os import chdir,chmod,chown,dup,dup2,environ,environb,fchdir,fspath,getcwd,getegid,geteuid,getgid from os import getpid,getuid,getgroups,link,listdir,lstat,mkdir,path,read,readlink,rename,rmdir,stat from os import unlink,utime,write from os.path import exists,expanduser,isdir,join,lexists,sep,split from pathlib import Path,PurePath from random import randint,randrange from socket import inet_pton,AF_INET,AF_INET6 from stat import S_ISBLK,S_ISCHR,S_ISDIR,S_ISDOOR,S_ISFIFO,S_ISLNK,S_ISPORT,S_ISREG,S_ISSOCK,S_ISWHT from stat import filemode from struct import pack,unpack from subprocess import call,DEVNULL,PIPE,Popen,run,STDOUT from sys import argv,maxsize,stderr,stdin,stdout,version_info from re import sub from termios import TIOCGWINSZ from textwrap import TextWrapper from time import gmtime,localtime,strftime,time as secs from types import MethodType from uuid import uuid4 # what time sfc module was loaded sfc_load_time = secs() # set some lower case aliases dec = Decimal decimal = Decimal num = Number number = Number path = Path popen = Popen purepath = PurePath textwrapper = TextWrapper # set some missing types NoneType = type(None) NotImplementedType = type(NotImplemented) EllipsisType = type(Ellipsis) # define which characters are used for decimal point and separator decpnt = str(3/2)[1] decsep = ',' if decpnt == '.' else '.' # get some current system info (POSIX) # this could fail on some non-POSIX systems cwd = getcwd() pid = getpid() egid = getegid() rgid = getgid() euid = geteuid() ruid = getuid() groups = getgroups() # define short ways to express combo exceptions # this name is not exported from this module _tv = (TypeError,ValueError) # strftime formats for date stamp and time stamp and date/time expression ds_fmt = '%Y%m%d' ts_fmt = '%Y%m%d-%H%M%S' dt_fmt = '%Y-%m-%dT%H:%M:%S' # sets for no and yes in a few European languages (not exported from this module) _n={'dim','ei','ez','ikke','ingen','inno','innò','n','nao','não','ne','nee','nei','nein','no','non',} _y={'ano','bai','da','ie','iè','ja','jah','jawohl','jes','jo','kylla','oui','si','sí','sì','sim','y','yes',} # assign mappings and sets used by the AWS region abbreviation conversion functions region_to_reg() and reg_to_region() map0 = {'a':'ap','c':'ca','e':'eu','s':'sa','u':'us'} map1 = {'n':'north','e':'east','s':'south','w':'west','c':'central'} map2 = {'se':'southeast','sw':'southwest','ne':'northeast','nw':'northeast'} map3 = {'af':'af','ap':'ap','ca':'ca','eu':'eu','me':'me','sa':'sa','ug':'us-gov','us':'us'} set0 = {'af','ap','ca','eu','me','sa','us'} set1 = {'east','north','south','west','central'} set2 = {'southeast','southwest','northeast','northwest'} # power of 1000 metric table: metric_1000_table = dict( s = 500, # ???? k = 10**3, # kilo b = 4000, # block m = 10**6, # mega g = 10**9, # giga t = 10**12, # tera p = 10**15, # peta e = 10**18, # exa z = 10**21, # zetta y = 10**24, # yotta x = 10**27, # xenna w = 10**30, # weka v = 10**33, # vendeka ) # power of 1024 metric table: metric_1024_table = dict( s = 512, # sector k = 2**10, # kilo b = 4096, # block m = 2**20, # mega g = 2**30, # giga t = 2**40, # tera p = 2**50, # peta e = 2**60, # exa z = 2**70, # zetta y = 2**80, # yotta x = 2**90, # xenna w = 2**100, # weka v = 2**110, # vendeka ) # base 64 digit set in order of values 0..64 (same set as YouTube, not same index) digits64 = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_' # the characters on YouTube are obvious, but their index values are unknown # the indexes do not really matter to keep each index separate # base 62 digit set in order of values 0..62 (a commonly used base) digits62 = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ' # base 57 digit set in order of values 0..56 (some letters skipped) digits57 = '0123456789abcdefghjkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ' # base 57 is the smallest base that can resolve a 64 bit value in 11 digits assert 56**11 < 2**64 < 57**11 # 57**11 == 0x11e61719e79566a29 (more than enough codes) # 56**11 == 0x0ebb7392e00000000 (not enough codes) ascii_control_code_to_esc = { # letter that follows backslash, number indexed 0:'0', # nul 7:'a', # bel 8:'b', # bs 9:'t', # tab 10:'n', # nl 11:'v', # vtab 12:'f', # ff 13:'r', # cr 27:'e', # esc } tty_ctrl = ascii_control_code_to_esc # reverse table escap sequence letter to code ('r' to 13) ascii_control_esc_to_code = {v:k for k,v in ascii_control_code_to_esc.items()} ascii_control_code_to_short = ( 'nul', 'soh', 'stx', 'etx', 'eot', 'enq', 'ack', 'bel', 'bs', 'tab', 'nl', 'vtab', 'ff', 'cr', 'so', 'si', 'dle', 'dc1', 'dc2', 'dc3', 'dc4', 'nak', 'syn', 'etb', 'can', 'em', 'sub', 'esc', 'fs', 'gs', 'rs', 'us', ) # test that all 32 codes are included above assert len(ascii_control_code_to_short)==32 # reverse the above tuple to a dict of: control code short name to index ascii_control_short_to_code = {ascii_control_code_to_short[x]:x for x in range(32)} # tuple of control code index to long name ascii_control_code_to_long = ( "null character", "start of heading", "start of text", "end of text", "end of transmission", "enquiry", "acknowledge", "bell", "backspace", "horizontal tab", "new line", "vertical tab", "form feed", "carriage return", "shift out", "shift in", "data link escape", "device control 1", "device control 2", "device control 3", "device control 4", "negative ack", "synchronous idle", "end of transmission block", "cancel", "end of medium", "substitute", "escape", "file separator", "group separator", "record separator", "unit separator", ) # test that all 32 codes are included above assert len(ascii_control_code_to_long) == 32 # revese the above tuple to a dict of: control code long name to index ascii_control_long_to_code = {ascii_control_code_to_long[x]:x for x in range(32)} # map between tables (probably never needed) ascii_control_short_to_long = {ascii_control_code_to_short[x]:ascii_control_code_to_long[x]for x in range(32)} ascii_control_long_to_short = {ascii_control_code_to_long[x]:ascii_control_code_to_short[x]for x in range(32)} #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # for multiple types, test logic is (IS type1 OR IS type2 ...) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- from adict import adict as _adict from collections.abc import Iterable as _iter from decimal import Decimal as _dec from io import IOBase as _iob from numbers import Number as _num def is_a (x):return isinstance(x,_adict) def is_ad (x):return isinstance(x,(_adict,dict)) def is_adict (x):return isinstance(x,_adict) def is_b (x):return isinstance(x,bytes) def is_ba (x):return isinstance(x,bytearray) def is_bb (x):return isinstance(x,(bytes,bytearray)) def is_bi (x):return isinstance(x,(bool,int)) def is_bif (x):return isinstance(x,(bool,int,float)) def is_bifc (x):return isinstance(x,(bool,int,float,complex)) def is_bifcd (x):return isinstance(x,(bool,int,float,complex,_dec)) def is_bool (x):return isinstance(x,bool) def is_bytearray (x):return isinstance(x,bytearray) def is_bytes (x):return isinstance(x,bytes) def is_complex (x):return isinstance(x,complex) def is_decimal (x):return isinstance(x,_dec) def is_da (x):return isinstance(x,(dict,_adict)) def is_dict (x):return isinstance(x,dict) def is_ds (x):return isinstance(x,(dict,set)) def is_fc (x):return isinstance(x,(float,complex)) def is_fcd (x):return isinstance(x,(float,complex,_dec)) def is_fd (x):return isinstance(x,(float,_dec)) def is_float (x):return isinstance(x,float) def is_frozenset (x):return isinstance(x,frozenset) def is_id (x):return (x).isidentifier() def is_if (x):return isinstance(x,(int,float)) def is_ifc (x):return isinstance(x,(int,float,complex)) def is_ifcd (x):return isinstance(x,(int,float,complex,_dec)) def is_ifd (x):return isinstance(x,(int,float,_dec)) def is_iio (x):return isinstance(x,(int,_iob)) def is_int (x):return isinstance(x,int) def is_iobase (x):return isinstance(x,_iob) def is_iter (x):return isinstance(x,_iter) def is_lio (x):return isinstance(x,(list,_iob)) def is_list (x):return isinstance(x,list) def is_ls (x):return isinstance(x,(list,set)) def is_lt (x):return isinstance(x,(list,tuple)) def is_ltsf (x):return isinstance(x,(list,tuple,set,frozenset)) def is_ltsfd (x):return isinstance(x,(list,tuple,set,frozenset,dict)) def is_num (x):return isinstance(x,_num) def is_number (x):return isinstance(x,_num) def is_sb (x):return isinstance(x,(str,bytes)) def is_sbi (x):return isinstance(x,(str,bytes,int)) def is_sbb (x):return isinstance(x,(str,bytes,bytearray)) def is_sbbi (x):return isinstance(x,(str,bytes,bytearray,int)) def is_seq (x):return isinstance(x,_sequ) def is_set (x):return isinstance(x,set) def is_sf (x):return isinstance(x,(set,frozenset)) def is_si (x):return isinstance(x,(str,int)) def is_str (x):return isinstance(x,str) def is_tf (x):return isinstance(x,(tuple,frozenset)) def is_tuple (x):return isinstance(x,tuple) #-------.-------.-------.-------.-------.-------.-------.-------.-------.------- #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # setting any one of these environment variables to a string that eval() returns a true value for will enable debug mode # # SFCdebug SFCdebugMODE SFCdebugging # SFCdebugmode # sfcdebug sfcdebugmode sfcdebugging # SFCDEBUG SFCDEBUGMODE SFCDEBUGGING # # any one variable set true overrides every other variable set false (big or) # importing module sfcdebug will enable debug mode unless any one of those environment variables is set to any string #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- _debugvars = [y for x in ('SFCdebug','SFCdebugMODE','SFCdebugmode','SFCdebugging') for y in (x,x.upper(),x.lower())] _debugmode = any(eval(environ.get(x,'0'))for x in _debugvars) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- _zopen_default_compresslevel = 6 # when zopen() compresses a file and no level is specified #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- class adict(dict): """Class that is a dictionary with items usable like attributes. purpose class that is a dictionary with items usable like attributes so code can look like: obj1.foo = obj2.bar instead of: obj1['foo'] = obj2['bar'] although the latter may also be used requires python 3.6 or later (uses f-strings) warning this class will fail in cases where keys have values the same as existing methods warning adict_obj['_Iterable__'] = value # bad idea init usage adict_obj = adict() # empty adict_obj = adict(dictionary) adict_obj = adict(iterable) adict_obj = adict(dictionary,key=value...) adict_obj = adict(iterable,key=value...) adict_obj = adict(key=value...) methods _cpwo, _del, _get, _pop, plus all dict methods attr usage adict_obj.name # name must be valid identifier dict usage adict_obj[key] note attribute usage is like string keys that are limited to what can be a valid identifier while dictionary usage can use any hashable value as a key """ def __init__(self,*args,**kwargs): """Create a dictionary that works like attributes.""" n = 0 for arg in args: if isinstance(arg,(adict,dict)): self.update(arg) elif isinstance(arg,Iterable): for a in arg: self.update(**dict(a)) else: raise TypeError(f'argument #{n} is not a dictionary or iterable') n += 1 self.update(**kwargs) def __delattr__(self,key): """Delete any attribute or iterable of keys.""" if isinstance(key,Iterable): for k in key: del self[k] else: del self[key] return self def __getattr__(self,key): """Get any attribute.""" return self[key] def __setattr__(self,key,value): """Set any attribute.""" self[key] = value return self def _cpwo(self,keys=[]): """Copy self but without a key or any key in iterable.""" new = adict(self) if not isinstance(keys,Iterable): keys = [keys] for key in keys: new.pop(key,None) return new def _del(self,key): """Delete any key or iterable of keys.""" if isinstance(key,Iterable): for k in key: del self[k] else: del self[key] return self def _get(self,keys,default=None): """Get a value based on a key or iterable of keys or return default.""" if not isinstance(keys,Iterable): keys = [keys] for key in iter(keys): if key in self: return self[key] return default def _pop(self,keys,default=None): """Pop a value based on a key or iterable of keys or return default.""" if not isinstance(keys,Iterable): keys = [keys] for key in iter(keys): if key in self: value = self[key] del self[key] return value return default #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def allin(pats,obj): """Is all of pats in obj?""" return all(x in obj for x in pats) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def all_args_same_type(*args): """Return True if each argument is the same type.""" if len(args)<2: return False ty = type(args[0]) for arg in args[1:]: if not isinstance(arg,ty): return False return True #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def all_items_same_type(items=[]): """Return True if each item in sequence/set items is the same type.""" if len(items)<2: return False ty = type(items[0]) for item in items[1:]: if not isinstance(item,ty): return False return True #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def anyin(pats,obj): """Is any of pats in obj?""" return any(x in obj for x in pats) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def aprint(*args,**kwargs): """Like print() but bytes and bytearray are printed like string.""" if not args: raise ValueError('no args') for x in ('end','sep'): if x in kwargs: y=kwargs['end'] # when x == 'sep' then 'end' may not be in kwargs if is_str(y): pass elif is_bb(y): y=tostr(y) else: y=repr(y) #xyzzy kwargs[x]=y a=[] for x in args: if is_str(x): a.append(x) elif is_bb(x): a.append(tostr(x)) else: a.append(repr(x)) return print(*a,**kwargs) bprint = aprint #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # function aws_region_to_reg # purpose convert a region from full name form (9 to 14 characters) to abbreviated form (4 characters) # note this function tries to be versatile enough to handle most future regions # note these abbreviated forms may not match what others use #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def aws_region_to_reg(arg): """Convert an AWS region name from full to abbreviated.""" if not is_str(arg): return '' if len(arg)<5: return arg parts = arg.split('-') if len(parts)==3: # us-east-1 -> use1 if parts[0] in set0: if parts[1] in set1: return parts[0][:2] + parts[1][0] + parts[2] if parts[1] in set2: return parts[0][0] + parts[1][0] + parts[1][5] + parts[2] return '' if len(parts)==4: # us-gov-west-1 -> ugw1 return ''.join(x[0] for x in parts) return '' assert aws_region_to_reg('us-gov-west-1') == 'ugw1' assert aws_region_to_reg('sa-south-1') == 'sas1' assert aws_region_to_reg('ap-southwest-4') == 'asw4' assert aws_region_to_reg('eu-central-6') == 'euc6' assert aws_region_to_reg('us-east-2') == 'use2' assert aws_region_to_reg('af-south-3') == 'afs3' assert aws_region_to_reg('me-south-2') == 'mes2' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # function aws_reg_to_region # purpose convert a region from abbreviated form (4 characters) to full name form (9 to 14 characters) # note this function tries to be versatile enough to handle most future regions # note these abbreviated forms may not match what others use #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def aws_reg_to_region(arg): """Convert an AWS region name from abbreviated to full.""" if not is_str(arg): return '' if len(arg) in range(9,15): return arg if arg[0:2] in map3 and arg[2] in map1: return map3[arg[0:2]]+'-'+map1[arg[2]]+'-'+arg[3] if arg[0]=='a' and arg[1:3] in map2: return 'ap-'+map2[arg[1:3]]+'-'+arg[3] return '' assert aws_reg_to_region('ugw2') == 'us-gov-west-2' assert aws_reg_to_region('sas1') == 'sa-south-1' assert aws_reg_to_region('asw4') == 'ap-southwest-4' assert aws_reg_to_region('euc6') == 'eu-central-6' assert aws_reg_to_region('use2') == 'us-east-2' assert aws_reg_to_region('afs2') == 'af-south-2' assert aws_reg_to_region('mee1') == 'me-east-1' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def be(arg,begin=None,end=None): """Test if sequence {arg} Begins with {begin} and Ends with {end}.""" return (begin and arg[:len(begin)]==begin) and (end and arg[-len(end):]==end) assert be('fooxyzbar',begin='foo',end="bar") assert be('fooxyzbar','foo',"bar") #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def between(a,b,c): """Confine b (arg 2) to >= a (arg 1) or <= c (arg 3).""" if a < c: if b < a: b = a if b > c: b = c return b else: return None assert between('bar','ddd','foo') == 'ddd' assert between(3,5,7) == 5 assert between(10,6,20) == 10 assert between(10,36,20) == 20 assert between(99,44,11) is None #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def box(text='empty box',name=None,padh=1,padv=0): """Return given list of lines text with an ASCII box around it.""" if is_fd(padh): padh = int(padh) if not is_int(padh): error_message = f'padding parameter padh is not int or float' raise TypeError(error_message) if is_fd(padv): padv = int(padv) if not is_int(padv): error_message = f'padding parameter padv is not int or float' raise TypeError(error_message) if name is None: name = '' if is_bb(name): name = tostr(name) if not is_str(name): raise TypeError(f'name is {typestr(name)}, not str or bytes') if is_tuple(text): return tuple(box(list(text))) if is_str(text): return '\n'.join(box(text.splitlines()))+'\n' if is_bytes(text): return tobytes(box(tostr(text))) if is_bytearray(text): return tobytearray(box(tostr(text))) if not is_list(text): raise TypeError(f'argument #0 (text) type is {tn(text)}, not list') width = max([0,]+[len(x) for x in text]) if name: name = '<'+name+'>' text = ['|'+' '*padh+x.ljust(width,' ')+' '*padh+'|' for x in text] a = '+'+'-'*padh+'-'*width+'-'*padh+'+' b = '|'+' '*padh+' '*width+' '*padh+'|' r = [a] for n in range(padv): r += [b] r += text for n in range(padv): r += [b] r += [a] return r #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def camel_to_snake(c): """Convert a name from CamelForm to snake_form (str and bytes only).""" if is_bytes(c): return tobytes(camel_to_snake(tostr(c))) if is_bytearray(c): return tobytearray(camel_to_snake(tostr(c))) return sub('([a-z0-9])([A-Z])',r'\1_\2',sub('(.)([A-Z][a-z]+)',r'\1_\2',c)).lower() assert camel_to_snake('CamelToSnake') == 'camel_to_snake' assert camel_to_snake("CamelForm") == "camel_form" #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def cd(path=None,user=None): """Change the current directory to the specified directory of the specified user.""" # a relative path is relative to the current directory or the home directory # if user is not specified then the current user is used. path = get_user_dir(path,user) try: chdir(path) return 0 except OSError: return 1 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def chjust(st,wl=9,wr=9,pl='',pr='',ch=''): """Justify a string around a character to be lined up.""" if not pl: pl = ' ' if not pr: pr = ' ' if not ch: ch = decpnt sl,sr = st.split(ch,1) return sl.rjust(wl,pl)+ch+sr.ljust(wr,pr) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def chmodref(rfn,fn,filenotfound1=None,filenotfound2=None): """Do chmod with mode from given file.""" try: rstat = stat(rfn,follow_symlinks=False) except FileNotFoundError: return filenotfound1 try: return chmod(fn,rstat.st_mode&4095,follow_symlinks=False) except FileNotFoundError: return filenotfound2 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def chownref(rfn,fn,filenotfound1=None,filenotfound2=None): """Do chown with owner and group from given file.""" try: rstat = stat(rfn,follow_symlinks=False) except FileNotFoundError: return filenotfound1 try: return chown(fn,rstat.st_uid,rstat.st_gid,follow_symlinks=False) except FileNotFoundError: return filenotfound2 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def clip(l,n,h): """Clip high and low values of a number.""" if l is not None: n=max(n,l) if h is not None: n=min(n,h) return n #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def cmds(cmds,**opts): # to be replaced by runpipeline() and its aliases """Run a command (list/tuple of list/tuple of str/bytes/bytearray) discarding its output.""" # for POSIX/BSD/Unix/Linux systems # tested on Ubuntu Linux # each command can be either a list or tuple # but can be different between commands # the command pipeline can be either a list or tuple # one command can be a list/tuple of str/bytes/bytearray if not is_lt(cmds): raise TypeError('argument 1 (cmds) must be a list or tuple (of list/tuple of str/bytes/bytearray).') if all(is_sbb(arg)for arg in cmds): cmds=[cmds] # a list of args becomes a list of one command if not all(is_lt(cmd)for cmd in cmds): raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple (of str/bytes/bytearray).') if not all(all(is_sbb(arg)for arg in cmd)for cmd in cmds): raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple of str or bytes or bytearray.') procs=[] opts=dict(opts) opts['stdout']=PIPE opts['close_fds']=False for cmd in cmds[:-1]: procs.append(Popen(cmd,**opts)) opts['stdin']=procs[-1].stdout # connect next stdin to previous stdout opts.pop('stdout',0) with Popen(cmds[-1],**opts) as lastproc: return lastproc.wait() cmd = cmds #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def cmdsbuf(cmds,**opts): # to be replaced by runpipeline() and its aliases """Run a command pipeline (list/tuple of list/tuple of str/bytes/bytearray) returning its last output as a big bytes buffer of newline separated lines.""" # for POSIX/BSD/Unix/Linux systems # tested on Ubuntu Linux # each command can be either a list or tuple # but can be different between commands # the command pipeline can be either a list or tuple # one command can be a list/tuple of str/bytes/bytearray if not is_lt(cmds): raise TypeError('argument 1 (cmds) must be a list or tuple (of list/tuple of str/bytes/bytearray).') if all(is_sbb(arg)for arg in cmds): cmds=[cmds] # a list of args becomes a list of one command if not all(is_lt(cmd)for cmd in cmds): raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple (of str/bytes/bytearray).') if not all(all(is_sbb(arg)for arg in cmd)for cmd in cmds): raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple of str or bytes or bytearray.') procs=[] opts=dict(opts) opts['stdout']=PIPE opts['close_fds']=False for cmd in cmds[:-1]: procs.append(Popen(cmd,**opts)) opts['stdin']=procs[-1].stdout # connect next stdin to previous stdout with Popen(cmds[-1],**opts) as lastproc: return lastproc.stdout.read() cmdbuf = cmdsbuf #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def cmdslist(cmds,**opts): # to be replaced by runpipeline() and its aliases """Run a command (list/tuple of list/tuple of str/bytes/bytearray) returning its last output as a list/tuple of bytes (lines).""" # for POSIX/BSD/Unix/Linux systems # tested on Ubuntu Linux # each command can be either a list or tuple # but can be different between commands # the command pipeline can be either a list or tuple # one command can be a list/tuple of str/bytes/bytearray if not is_lt(cmds): raise TypeError('argument 1 (cmds) must be a list or tuple (of list/tuple of str/bytes/bytearray).') if all(is_sbb(arg)for arg in cmds): cmds=[cmds] # a list of args becomes a list of one command if not all(is_lt(cmd)for cmd in cmds): raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple (of str/bytes/bytearray).') if not all(all(is_sbb(arg)for arg in cmd)for cmd in cmds): raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple of str or bytes or bytearray.') procs=[] opts=dict(opts) opts['stdout']=PIPE opts['close_fds']=False for cmd in cmds[:-1]: procs.append(Popen(cmd,**opts)) opts['stdin']=procs[-1].stdout # connect next stdin to previous stdout with Popen(cmds[-1],**opts) as lastproc: return lastproc.stdout.read().splitlines() cmdlist = cmdslist #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def cmdspipe(cmds,**opts): # to be replaced by runpipeline() and its aliases """Run a command (list/tuple of list/tuple of str/bytes/bytearray) returning its last output as a file to the read end of a pipe from stdout of the last process.""" # for POSIX/BSD/Unix/Linux systems # tested on Ubuntu Linux # each command can be either a list or tuple # but can be different between commands # the command pipeline can be either a list or tuple # one command can be a list/tuple of str/bytes/bytearray if not is_lt(cmds): raise TypeError('argument 1 (cmds) must be a list or tuple (of list/tuple of str/bytes/bytearray).') if all(is_sbb(arg)for arg in cmds): cmds=[cmds] # a list of args becomes a list of one command if not all(is_lt(cmd)for cmd in cmds): raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple (of str/bytes/bytearray).') if not all(all(is_sbb(arg)for arg in cmd)for cmd in cmds): raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple of str or bytes or bytearray.') procs=[] opts=dict(opts) opts['stdout']=PIPE opts['close_fds']=False for cmd in cmds[:-1]: opts['universal_newlines']=is_str(cmd[0]) procs.append(Popen(cmd,**opts)) opts['stdin']=procs[-1].stdout # connect next stdin to previous stdout return Popen(cmds[-1],**opts).stdout cmdpipe = cmdspipe #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def crdir(dirpath): """Function to create a directory and its needed parent directoris.""" if path.lexists(dirpath): return dirpath,1 # the named directory exists subs = [] while not path.lexists(dirpath): dirpath,sub = path.split(dirpath) subs[:0] = [sub] if not path.isdir(dirpath): return dirpath,3 for sub in subs: dirpath = path.join(dirpath,sub) try: mkdir(dirpath) except PermissionError as err: return dirpath,4 except OSError as err: return dirpath,9 return 0,0 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def crpar(dirpath): """Function to create parent directories for a file object name.""" if path.lexists(dirpath): return dirpath,1 # the named object exists dirpath,sub = path.split(dirpath) if path.lexists(dirpath): return dirpath,2 # parent of named object exists subs = [] while not path.lexists(dirpath): dirpath,sub = path.split(dirpath) subs[:0] = [sub] if not path.isdir(dirpath): return dirpath,3 for sub in subs: dirpath = path.join(dirpath,sub) try: mkdir(dirpath) except PermissionError as err: return dirpath,4 except OSError as err: return dirpath,9 return 0,0 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def dda(*args,**kwargs): """Copy a dictionary (arg 1), deleting items (args 2 ..) and adding items (key word args).""" if args: d = args[0] if not d:d = {} if not is_dict(d):raise TypeError('argument 1 is not a dictionary') if d:d = dict(d) else:d = {} for k in args[1] if len(args)>1 and is_list(args[1]) else args[1:]: if k in d:del d[k] for k,v in kwargs.items():d[k] = v return d #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def deciplex_to_int(digits='0'): """Convert a string in deciplex notation to an int with the given value.""" ### deciplex_table = metric_1024_table try: return int(digits,0) except Exception: pass try: return int('0'+digits,0) except Exception: pass r = 0 n = 0 for c in digits.lower(): if c in metric_1024_table: r += n * metric_1024_table[c] n = 0 elif c.isdigit(): n *= 10 n += int(c) else: raise ValueError(f'invalid deciplex character {c!r}') return r+n assert deciplex_to_int('1m2k3') == ( 1048576 + 2*1024 + 3 ) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def decode16(arg,**kwargs): """Convert a string from base16 to the string it encodes or convert all items in a list.""" if is_list(arg): return [decode16(x,**kwargs)for x in arg] if is_tuple(arg): return tuple(decode16(x,**kwargs)for x in arg) if is_str(arg): return tostr(b16decode(arg,**kwargs)) if is_bytes(arg): return b16decode(arg,**kwargs) if is_bytearray(arg): return bytearray(b16decode(arg,**kwargs)) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def decode32(arg,**kwargs): """Convert a string from base32 to the string it encodes or convert all items in a list.""" if is_str(kwargs.get('map01',None),str): kwargs['map01'] = tobytes(kwargs['map01']) if is_list(arg): return [decode32(x,**kwargs)for x in arg] if is_tuple(arg): return tuple(decode32(x,**kwargs)for x in arg) if is_str(arg): return tostr(b32decode(arg,**kwargs)) if is_bytes(arg): return b32decode(arg,**kwargs) if is_bytearray(arg): return bytearray(b32decode(arg,**kwargs)) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def decode64(arg,**kwargs): """Convert a string from base64 to the string it encodes or convert all items in a list or tuple.""" if is_str(kwargs.get('altchars',None)): kwargs['altchars'] = tobytes(kwargs['altchars']) if is_list(arg,list): return [decode64(x,**kwargs)for x in arg] if is_tuple(arg,tuple): return tuple(decode64(x,**kwargs)for x in arg) if is_str(arg,str): return tostr(b64decode(arg,**kwargs)) if is_bytes(arg,bytes): return b64decode(arg,**kwargs) if is_bytearray(arg,bytearray): return bytearray(b64decode(arg,**kwargs)) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # POSIX only def dslocal(time=None,fmt=None): """Return a date stamp (YYYYmmdd) str of given local time or of now.""" try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ds_fmt return strftime(fmt,localtime(int(time))) dsloc = dslocal ds = dslocal assert dslocal(time=0) == '19691231' assert dslocal(1999999999) == '20330517' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # POSIX only def dsutc(time=None,fmt=None): """Return a date stamp (YYYYmmdd) str of given utc time or of now.""" # print('\n\ndsutc: called\n',time,flush=1) try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ds_fmt return strftime(fmt,gmtime(int(time))) assert dsutc(time=0) == '19700101' assert dsutc(1999999999) == '20330518' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ehr(*args,**kwargs): """Print a horizontal rule to stderr with optional string at end.""" if 'file' not in kwargs: kwargs['file'] = stderr return hr(*args,**kwargs) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def encode16(arg): """Convert a str or bytes to base16 as same type or convert all items in a list/tuple.""" if is_list(arg,list): return [encode16(x)for x in arg] if is_tuple(arg,tuple): return tuple(encode16(x)for x in arg) if is_str(arg,str): return tostr(b16encode(tobytes(arg))) if is_-bytes(arg,bytes): return b16encode(arg) if is_bytearray(arg,bytearray): return tobytearray(b16encode(tobytes(arg))) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def encode32(arg): """Convert a str or bytes to base32 as same type or convert all items in a list/tuple.""" if is_list(arg): return [encode32(x)for x in arg] if is_tuple(arg): return tuple(encode32(x)for x in arg) if is_str(arg): return tostr(b32encode(tobytes(arg))) if is_bytes(arg): return b32encode(arg) if is_bytearray(arg): return tobytearray(b32encode(tobytes(arg))) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def encode64(arg,**kwargs): """Convert a str or bytes to base64 as same type or convert all items in a list/tuple.""" if is_str(kwargs.get('altchars',None)): kwargs['altchars'] = tobytes(kwargs['altchars']) if is_list(arg): return [encode64(x,**kwargs)for x in arg] if is_tuple(arg): return tuple(encode64(x,**kwargs)for x in arg) if is_str(arg): return tostr(b64encode(tobytes(arg),**kwargs)) if is_bytes(arg): return b64encode(arg,**kwargs) if is_bytearrary(arg): return tobytearray(b64encode(tobytes(arg),**kwargs)) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def eprint(*args,**kwargs): """Print to stderr with flush.""" if 'file' not in kwargs: kwargs['file']=stderr if 'flush' not in kwargs: kwargs['flush']=True for x in (stderr,stdout): if x!=kwargs.get('file',None): x.flush() try: return print(*args,**kwargs) except BrokenPipeError: return #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def esc_seq(s): """Convert character or bytes to escape sequences according to type.""" # a sequence of ints also works. if is_str(s): return esc_seq_unicode(s) return esc_seq_ascii(s) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def esc_seq_ascii(a): """Convert ASCII characters or bytes to escape sequences.""" # a sequence of ints also works. if is_str(a): a=[ord(x)for x in a] if is_bb(a): a=[x for x in a] if is_lt(a): o='' for x in a: if x in tty_ctrl: o=o+'\\'+tty_ctrl[x] elif x<127: o=o+repr(chr(x))[1:-1] elif x<256: o=o+'\\x'+hex(x)[-2:] else: o=o+'\\u????' return o return None #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def esc_seq_unicode(u): """Convert Unicode characters or bytes to escape sequences.""" # a sequence of ints also works. if is_str(u,str): u=[ord(x)for x in u] if is_bb(u): u=[x for x in u] if is_lt(u): o='' for x in u: if x in tty_ctrl: o=o+'\\'+tty_ctrl[x] else: o=o+repr(chr(x))[1:-1] return o return None #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def eprint(*args,**opts): """Print to stderr with flush.""" if not args: return None if 'file' not in opts: opts['file']=stderr if 'flush' not in opts: opts['flush']=True f=opts['file'] # flush what is not printed for x in (stderr,stdout): if f!=x: try: x.flush() except BrokenPipeError: return True try: print(*args,**opts) except BrokenPipeError: return True return False emsg = eprint #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- _error_count = 0 def error(msg=None): global _error_count # able to increment """Print message and count it. If no message given and count is non-zero, print count and exit.""" if is_str(msg): eprint(msg) _error_count += 1 if is_bb(msg): # msg can be bytes or bytearray eprint(tostr(msg)) _error_count += 1 else: # anything else check error count if _error_count: try: stderr.flush() except BrokenPipeError: pass exit(f'aborting due to {_error_count} error{"s"[_error_count==1:]}') return return errmsg = error error_check = error #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def extract(s,a,b): """Return part of s after first a and before following b.""" try: x = s.index(a) x += len(a) try: y = s.index(b,x) return s[x:y] except ValueError: return s[x:] except ValueError: try: y = s.index(b) return s[:y] except ValueError: return s assert extract('foo/bar','foo','bar') == '/' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def extractx(s,a,b): """Like extract(s,a,b) but return index and len within s.""" """Return index of part of s after a and before b (or 2-tuple of index and length).""" try: x = s.index(a) x += len(a) try: y = s.index(b,x) return x,y except ValueError: return x,len(s) except ValueError: try: y = s.index(b) return 0,y except ValueError: return 0,len(s) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ff(f,d=3): """Format a float (or int) into a string rounded with a given number of fraction digits.""" try: f = float(f) d = max(int(d),0) except ValueError: return '[ERROR]' s = str(int(f*10.0**d+0.5)) while len(s)<4: s = '0'+s return s if d<1 else f'{s[:-d]}.{s[-d:]}' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def fstype_char(name): """Return a single character describing the file system type of the named object.""" # return type is the same as the type of the given name if is_bytes(name): return tobytes(fstype(tostr(name))) if is_bytearray(name): return tobytearray(fstype(tostr(name))) try: stat = stat(name,follow_symlinks=False) mode = stat.st_mode t = 'unknown' except Exception: t = 'none' if mod_stat.S_ISREG(mode): t = 'f' if mod_stat.S_ISDIR(mode): t = 'd' if mod_stat.S_ISLNK(mode): t = 'l' if mod_stat.S_ISCHR(mode): t = 'c' if mod_stat.S_ISBLK(mode): t = 'b' if mod_stat.S_ISFIFO(mode): t = 'p' if mod_stat.S_ISSOCK(mode): t = 's' if mod_stat.S_ISDOOR(mode): t = 'D' if mod_stat.S_ISPORT(mode): t = 'P' if mod_stat.S_ISWHT(mode): t = 'P' return t fstype = fstype_char filetype = fstype_char filetype_char = fstype_char #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def fstype_short(name): """Return a short lower case string describing the file system type of the named object.""" # return type is the same as the type of the given name if is_bytes(name): return tobytes(fstype(tostr(name))) if is_bytearray(name): return tobytearray(fstype(tostr(name))) try: stat = stat(name,follow_symlinks=False) mode = stat.st_mode t = 'unknown' except Exception: t = 'none' if mod_stat.S_ISREG(mode): t = 'file' if mod_stat.S_ISDIR(mode): t = 'dir' if mod_stat.S_ISLNK(mode): t = 'link' if mod_stat.S_ISCHR(mode): t = 'char' if mod_stat.S_ISBLK(mode): t = 'blk' if mod_stat.S_ISFIFO(mode): t = 'pipe' if mod_stat.S_ISSOCK(mode): t = 'socket' if mod_stat.S_ISDOOR(mode): t = 'door' if mod_stat.S_ISPORT(mode): t = 'port' if mod_stat.S_ISWHT(mode): t = 'whiteout' return t filetype_short = fstype_short #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def fstype_long(name): """Return a long mixed case string describing the file system type of the named object.""" # return type is the same as the type of the given name if is_bytes(name): return tobytes(fstype(tostr(name))) if is_bytearray(name): return tobytearray(fstype(tostr(name))) try: stat = stat(name,follow_symlinks=False) mode = stat.st_mode t = 'unknown' except Exception: t = 'none' if mod_stat.S_ISREG(mode): t = 'Regular File' if mod_stat.S_ISDIR(mode): t = 'Directory' if mod_stat.S_ISLNK(mode): t = 'Symlink' if mod_stat.S_ISCHR(mode): t = 'Character Device' if mod_stat.S_ISBLK(mode): t = 'Block Device' if mod_stat.S_ISFIFO(mode): t = 'Pipe' if mod_stat.S_ISSOCK(mode): t = 'Socket' if mod_stat.S_ISDOOR(mode): t = 'Door' if mod_stat.S_ISPORT(mode): t = 'Port' if mod_stat.S_ISWHT(mode): t = 'Whiteout' return t filetype_long = fstype_long #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def flatten(a): """Flatten any depth of list/tuple of lists/tuples to a single level list/tuple.""" r = [] for x in a: if is_lt(x): r.extend(flatten(x)) else: r.append(a) return type(a)(r) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def fn_exists(name): """File name (sbb) exists test for conditionals, without raising exceptions.""" try: stat(name,follow_symlinks=False) return True except FileNotFoundError: return False except Exception: return None exists = fn_exists #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def fn_status(name,exists=True,missing=False): """File name (sbb) missing test for conditionals, without raising exceptions.""" try: stat(name,follow_symlinks=False) return exists except FileNotFoundError: return missing except Exception: return None fn_stat = fn_status #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ftrgen(apex, mindepth=0, maxdepth=inf, *, key=None, # provide alternate sort key reverse=False, # reverse order of directory list reads ascends=False, # yield directory at ascend from time ): """File tree recursion GENERATOR that yields all paths, in the same order as sorting the whole tree. optionally sorted by sortkey in key= applied to the path. optionally sorted in reverse order. optionally with ascends yielded. as a 3-tuple. """ if is_int(mindepth): if mindepth < 0: mindepth = 0 else: mindepth = 0 if is_int(maxdepth): if maxdepth < mindepth: maxdepth = mindepth if maxdepth > maxsize: maxdepth = maxsize else: maxdepth = maxsize if is_int(mindepth): if mindepth > maxdepth: mindepth = maxdepth depth = 0 # the starting depth tree = [[apex]] # the top always has one name until all done while tree[0]: # loop while the top has one name if tree[-1]: # if the bottom is not empty path = sep.join([x[0] for x in tree]) if depth >= mindepth: if path[:2] == dotsep: path = path[2:] yield 0,depth,path try: readlink(path) except OSError: try: # directory? new = sorted(listdir(path),key=key,reverse=reverse) # path is a directory, do descend, if allowed if depth < maxdepth: # if allowed to go deeper, descend tree.append(new) # use the next list of names depth += 1 # calculate how much deeper continue except OSError: # not a directory, stay at this level and select the next name pass tree[-1].pop(0) # select the next name else: # ascend depth -= 1 # less deep tree[-1:] = [] # shorten tree if ascends: # if ascends wanted path = sep.join([x[0] for x in tree]) # form path for ascend if path[:2] == dotsep: path = path[2:] if depth >= mindepth: yield 1,depth,path tree[-1].pop(0) # next name (may become empty) ftrgen3 = ftrgen #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ftrgen4(apex, mindepth=0, maxdepth=inf, *, key=None, # provide alternate sort key reverse=False, # reverse order of directory list reads ascends=False, # yield directory at ascend from time ): """File tree recursion GENERATOR that yields all paths, in the same order as sorting the whole tree, or reversed. optionally sorted by sortkey in key= applied to the path. yield for each path: path, stat object. ftrgen4 uses ftrgen """ for path,depth in ftrgen(apex, mindepth=mindepth, maxdepth=maxdepth, key=key, reverse=reverse, ascends=ascends ): yield ascend,depth,path,lstat(path) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ftrgen5(apex, mindepth=0, maxdepth=inf, types=None, *, key=None, reverse=False, ascends=False, ): """File tree recursion GENERATOR that yields all paths, in the same order as sorting the whole tree, or reversed. optionally sorted by sortkey in key= applied to the path. yield for each path: path, stat object, type letter (1-str). ftrgen5 uses ftrgen4 """ if types is True or not types: types = 'fdlbcpsPDW?' # default to all types types = frozenset(types) for p,d,s in ftrgen4(apex, mindepth=mindepth, maxdepth=maxdepth, key=key, reverse=reverse, ascends=ascends, ): if S_ISREG(s.st_mode):t='f' elif S_ISDIR(s.st_mode):t='d' elif S_ISLNK(s.st_mode):t='l' elif S_ISBLK(s.st_mode):t='b' elif S_ISCHR(s.st_mode):t='c' elif S_ISFIFO(s.st_mode):t='p' elif S_ISSOCK(s.st_mode):t='s' elif S_ISPORT(s.st_mode):t='P' elif S_ISDOOR(s.st_mode):t='D' elif S_ISWHT(s.st_mode):t='W' else:t='?' if t in types or types is ...: yield a,d,p,s,t #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ftrgen6(apex, mindepth=0, maxdepth=inf, types=None, *, key=None, # provide alternate sort key reverse=False, # reverse order of directory list reads ascends=False, # yield directory at ascend from time ): """File tree recursion GENERATOR that yields all paths, in the same order as sorting the whole tree, or reversed. optionally sorted by sortkey in key= applied to the path. yield for each path: path, stat object, type letter (1-str). ftrgen6 uses ftrgen5 """ for p,d,s,t in ftrgen5(apex, mindepth=mindepth, maxdepth=maxdepth, types=types, key=key, reverse=reverse, ascends=ascends, ): m=filemode(s.st_mode) if t!='f': m=t+m[1:] yield a,d,p,s,t,m #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ftype(path,long=False): """Given a file object path, return a str that gives its type.""" try: o = os.lstat(path) if not o: return '' except Exception: return '' m = o.st_mode i = iter(( 'regular file', 'directory', 'symbolic link', 'block device', 'character device', 'pipe', 'socket', 'Port', 'Door', 'Whiteout', )) if not long: i = iter('fdlbcpsPDW') for f in (S_ISREG,S_ISDIR,S_ISLNK,S_ISBLK,S_ISCHR,S_ISFIFO,S_ISSOCK,S_ISPORT,S_ISDOOR,S_ISWHT): t=next(i) if f(m): return t return '?' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def get_env(n): """Get environment variable value returned in same type as the given name.""" if is_str(n): return environ[n] if is_bytes(n): return tobytes(environ[tostr(n)]) if is_bytearray(n): return bytearray(tobytes(environ[tostr(n)])) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def get_user_dir(path=None,user=None): """Get path or path relative to home directory of this or specified user.""" if user: if is_sbb(user): u = tostr(user) if is_bb(user) else user else: raise TypeError(f'user (arg 2) must be one of str, bytes, bytearray, not {tn(user)}') else: u = '' # this user if path: if is_sbb(path): if user: p = join('~'+u,tostr(path) if is_bb(path) else path) else: p = tostr(path) if is_bb(path) else path else: raise TypeError(f'path (arg 1) must be one of str, bytes, bytearray, not {tn(path)}') else: p = '~'+u p = expanduser(p) if is_bb(path): if is_bytes(path): return tobytes(p) else: return tobytearray(p) else: return p #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def get_terminal_size(fd=1,tty='/dev/tty'): """Return a tuple with terminal screen size columns and lines. function get_terminal_size options fd= which file descriptor of a terminal tty= which pathname of a terminal purpose return a named tuple with terminal screen columns and lines result.columns result.lines """ if fd==1 and tty=='/dev/tty': try: c,l = os.get_terminal_size()[:2] except Exception: c,l = 0,0 else: c,l = 0,0 if c<1 or l<1: try: l,c = unpack('4H',ioctl(fd,TIOCGWINSZ,pack('4H',0,0,0,0)))[:2] except: try: fd = os.open(tty,os.O_WRONLY) l,c = unpack('4H',ioctl(fd,TIOCGWINSZ,pack('4H',0,0,0,0)))[:2] os.close(fd) except: c,l = 0,0 return namedtuple('terminal_size',('columns','lines'))(c,l) # alias get_term_size = get_terminal_size #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def home_dir(user=None): """Return a string of the home directory.""" return get_user_dir(user=user) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # need to refer to this as layering or something like that but also have an insert as well def insert(base,insert='',position=-3): """Insert a string into a larger string at a given position (negative at end).""" if not is_str(insert) or len(insert)<1: # if nothing to insert return base if not is_str(base) or len(base)<1: # if nothing to receive insert return base b = [x for x in base] i = [x for x in insert] if position<0: last = len(base)+position first = last-len(insert) p = first for c in i: b[p] = c p += 1 if first < 0: return False first = position last = first+len(insert) if last >= len(base): return False p = first for c in i: b[p] = c p += 1 return ''.join(b) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def hr(*args,**kwargs): """Return a horizontal rule with optional string at end.""" s = kwargs.pop('size',get_terminal_size().columns) r = s*'-' if args: a = args[0] if is_str(a): l = len(a) r = (s-l)*'-' + a return r #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def hrep(*args,**kwargs): """To STDERR, print a horizontal rule with optional string at end.""" eprint(hr(*args,**kwargs)) eprinthr = hrep #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def hrop(*args,**kwargs): """To STDOUT, print a horizontal rule with optional string at end.""" oprint(hr(*args,**kwargs)) oprinthr = hrop #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def int_to_deciplex(value=0): """Convert an int value to a string in deciplex notation.""" if value%1024: r = [str(value%1024)] else: r = [] if value>=1024 else ['0'] for s in 'kmgtpezyxwv': value //= 1024 if not value: break if value%1024: r[:0] = [str(value%1024)+s] return ''.join(r) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- _intsi_powers={'k':1,'m':2,'g':3,'t':4,'p':5,'e':6,'z':7,'y':8,'x':9,'w':10,'v':11,'r':12,'s':13,'q':14} def intsi(digits='0',base=10,multi=1024,empty=None,invalid=None): """Convert whole numbers with suffixes for SI units (metric).""" if not digits: return empty scale=1 if digits[-1] in _intsi_powers: scale=multi**_intsi_powers[digits[-1]] digits=digits[:-1] if not digits: return empty try: return scale*int(digits,base) except ValueError: return invalid #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def intd(*args,**kwargs): """Like int() but with an additional argument for a default value if int() fails.""" if 'default' in kwargs: default = kwargs.pop('default') try: return int(*args,**kwargs) except Exception: return default if len(args)<3: return int(*args,**kwargs) try: return int(*args[:2],**kwargs) except Exception: return args[2] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def intstr(i,c=None): # c: string of base digits """Convert an int to a str of given base system returning a str iterator.""" if not c: c = digits57 b = len(c) p = 1 if i<0: yield '-' i=-i if i1: p //= b yield c[(i//p)%b] return #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # 1 1 3 3 # 234 8 0 6 2 6 _intx_prefs = '__btq___o_d_____x_______________y___z' _intx_bases = {-8,-1,2,3,4,8,10,16,32,36} _intx_octal = {'0','1','2','3','4','5','6','7'} _intx_zeros = {'nil','nol','nul','null','zero'} def intx(digits='0',bases=None): # None becomes _intx_bases """Convert digits to int, in an extended way, better than int(). function convert digits to int, in an extended way, better than int() but still uses int() purpose implement a function to perform extended int conversion. syntax intx(digits) intx(digits,bases (list,tuple,set,frozenset)) intx(digits,bases= (list,tuple,set,frozenset)) digits all decimal, basic conversion like int(), or ... letters in the following forms may be given in lower or upper case or a mix of them: no prefix: {10} decimal prefix '0b' or 'b': {2} binary prefix '0t' or 't': {3} ternary (trinary) prefix '0q' or 'q': {4} quaternary prefix '0o' or 'o': {8} octal prefix '0d' or 'd': {10} decimal prefix '0x' or 'x': {16} cetal (hexadecimal) prefix '0y' or 'y': {32} duotridecimal prefix '0z' or 'z': {36} hexatrigesimal (hexatridecimal) type int type float converted to int type complex real part converted to int note not all base values below 36 are recognized, for bases not handled by intx(), use the built-in int() function. bases a set of bases allowed for conversion and/or various option flags or any container type supported by set() { -8, -1, 2, 3, 4, 8, 10, 16, 32, 36 }: default bases types list,tuple,set,frozenset are valid. any other type that can be used with if statement is valid and gets defaults feature more values may be included in the set of bases to enable additional features 0: suppress exceptions (return None if so) exceptions are still raised for bad arguments that raise exceptions otherwise -1: allow a prefix of '-' to negate the number so that "-123456" returns a negative value -2: forces the returned number to be negated -8: interpret a leading '0' as octal (as in source) -44 remove ',' characters from the numbers to convert -46 remove '.' characters from the numbers to convert defaults if no digits are given, return False note if bases is not given, the following set is used: { -8, -1, 2, 3, 4, 8, 10, 16, 32, 36 } returns the converted int value unless a suppressed exception or error happens author Phil D. Howard email 11054987560151472272755686915985840251291393453694611309 (provu igi la numeron al duuma) note Please report failures or code improvement to the author. """ # all these become the default set if not bases or bases is True: bases = _intx_bases # at this point all else are exceptions if not is_ltsf(bases): raise TypeError('intx: bases must be a sequence or a set of ints') if not all([is_int(x)for x in bases]): raise TypeError('intx: bases must be a sequence or a set of ints') # make it be a frozenset for fastest lookup bases = frozenset(bases) # handle numeric types like int() does if is_if(digits): return int(digits) # extend numeric types elif is_complex(digits): return int(digits.real) elif is_sbb(digit): pass else: raise TypeError('intx: digits type must be one of: str, bytes, bytearray, int, float, complex') # 0 means to suppress exceptions and return True when they would happen if 0 in bases: try: return intx(digits,bases-{0}) except: return True # special characters can be ignored in digits by adding the # negative of its ASCII code to the set of bases. for x in range(32,64): if -x in bases: digits = digits.replace(chr(x),'') # 'zero' -> 0 , etc if digits.lower() in _intx_zeros: return 0 # a prefix of '-' means a negative value if -1 in bases and digits[:0]=='-': return -intx(digits[:1],bases-{-1}) # invert the sign of the numner if -2 in bases: return -intx(digits,bases-{-2}) # support requested bases for b in (16,8,10,2,3,4,32,36): if b in bases: p = _intx_prefs[b] if digits[:2]=='0'+p: return int(digits[2:],b) elif digits[:1]==p: return int(digits[1:],b) # this special base means ... if -8 in bases: # ... leading '0' is ... if digits[:1]=='0': # ... octal instead of decimal return int(digits[1:],8) # if decimal is requested if 10 in bases: return int(digits,10) return int(digits,0) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- _interfaceips_46 = namedtuple('_interfaceips_46',['ipv4','ipv6']) def interfaceips(name): """Get IP addresses of specified network interface.""" if is_bb(name): name = ''.join(chr(x) for x in name) proc = Popen(['ifconfig',name],stdout=PIPE,bufsize=4096) try: out,err = proc.communicate(timeout=127) except TimeoutExpired: proc.kill() out,err = proc.communicate(timeout=127) lines = out.splitlines() for line in lines: if len(line) < 8: continue tokens = line.strip().split() if len(tokens) < 3: continue if tokens[0]=='inet': ip4 = tokens[1] if tokens[0]=='inet6': ip6 = tokens[1] if ip4 and '/' in ip4: ip4 = ip4.split('/')[0] if ip6 and '/' in ip6: ip6 = ip6.split('/')[0] return _interfaceips_46((ip4,ip6)) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def isare(n,_is='is',_are='are'): # "is" is a reserved word so these keyword options begin with an underscore. """Return '_is=' for 1 else '_are=' (or whatever is specified in keyword options).""" return _is if n==1 else _are #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def isbetween(a,b,c): """Test a <= b <= c (args 1, 2, 3) in the form of a function.""" return a <= b <= c #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def isin(a,b): # operator "in" in the form of a function of two arguments """Return True if arg 1 is in arg 2.""" return a in b #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def isnotin(a,b): # operator "not in" in the form of a function of two arguments """Return True if arg 1 is not in arg 2.""" return a not in b #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ispowerof(n,p=2): """Return True if argument 1 is an exact power of argument 2 (default 2).""" if n: while n!=1: if n%p: return False n//=p return True return False #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def istextlistbytearray(arg): return is_list(arg) and all(is_bytearray(x)for x in arg) def istextlistbytes(arg): return is_list(arg) and all(is_bytes(x)for x in arg) def istextliststr(arg): return is_list(arg) and all(is_str(x)for x in arg) def istextlistbb(arg): return is_list(arg) and all(is_bb(x)for x in arg) def istextlist(arg): return is_list(arg) and all(is_sbb(x)for x in arg) def istexttuplebytearray(arg): return is_tuple(arg) and all(is_bytearray(x)for x in arg) def istexttuplebytes(arg): return is_tuple(arg) and all(is_bytes(x)for x in arg) def istexttuplestr(arg): return is_tuple(arg) and all(is_str(x)for x in arg) def istexttuplebb(arg): return is_tuple(arg) and all(is_bb(x)for x in arg) def istexttuple(arg): return is_tuple(arg) and all(is_sbb(x)for x in arg) def istextseqbytearray(arg): return is_lt(arg) and all(is_bytearray(x)for x in arg) def istextseqbytes(arg): return is_lt(arg) and all(is_bytes(x)for x in arg) def istextseqstr(arg): return is_lt(arg) and all(is_str(x)for x in arg) def istextseqbb(arg): return is_lt(arg) and all(is_bb(x)for x in arg) def istextseq(arg): return is_lt(arg) and all(is_sbb(x)for x in arg) def istextbytearray(arg): return is_bytearray(arg) and (not arg or arg[-1] is bytearray(b'\n')) def istextbytes(arg): return is_bytes(arg) and (not arg or arg[-1] == b'\n') def istextstr(arg): return is_str(arg) and (not arg or arg[-1] == '\n') #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def istype(v,t): """Like isinstance, but more flexible, though less efficient.""" # istype arg2 can be a type or tuple, list, set or frozenset of types. # isinstance arg2 is limited to type or tuple of types, only x = type(v) return x is t or x in t #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def justhex(value,space=None,digits=None): space = 1 if space is None else space digits = space if digits is None else digits return hex(int(value))[2:].rjust(digits,'0').rjust(space) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def limit(a,b,c): """Limit a value (arg 2) to between lower (arg 1) and upper (arg 3) values.""" if b < a: return a if b > c: return c return b bound = limit #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def lineno(level=0): """Return line number of caller N (only arg 1) levels back.""" frame = currentframe().f_back while frame and level>0: level -= 1 frame = frame.f_back return int(frame.f_lineno) if frame else None ln = lineno #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def linenos(): """Return a list of line numbers of callers all the way back.""" nums = [] frame = currentframe().f_back while frame: nums.append(frame.f_lineno) frame = frame.f_back return nums #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def maxlen(seq): """Return the maximum len() of the items in the given sequence.""" return max(len(x)for x in seq) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def minlen(seq): """Return the minimum len() of the items in the given sequence.""" return min(len(x)for x in seq) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def microsecs(): """Return current time value scaled to microseconds.""" return secs()*1000000 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def millisecs(): """Return current time value scaled to milliseconds.""" return secs()*1000 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def nanosecs(): """Return current time value scaled to nanoseconds.""" return secs()*1000000000 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def nodup(seq): """Return a copy of a (maybe shorter) sequence without duplicates.""" if not isinstance(seq,(bytearray,bytes,list,str,tuple)): raise TypeError('argument is not a sequence') a,b = type(seq)(),set() for x in seq: if x not in b: a.append(x) b.add(x) return a rmdup = nodup removedup = nodup #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # for multiple types, test logic is (NOT type1 AND NOT type2 ...) == NOT (IS type1 OR IS type2 ...) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def not_a (x):return not isinstance(x,_adict) def not_ad (x):return not isinstance(x,(_adict,dict)) def not_adict (x):return not isinstance(x,_adict) def not_b (x):return not isinstance(x,bytes) def not_ba (x):return not isinstance(x,bytearray) def not_bb (x):return not isinstance(x,(bytes,bytearray)) def not_bi (x):return not isinstance(x,(bool,int)) def not_bif (x):return not isinstance(x,(bool,int,float)) def not_bifc (x):return not isinstance(x,(bool,int,float,complex)) def not_bifcd (x):return not isinstance(x,(bool,int,float,complex,_dec)) def not_bool (x):return not isinstance(x,bool) def not_bytearray (x):return not isinstance(x,bytearray) def not_bytes (x):return not isinstance(x,bytes) def not_complex (x):return not isinstance(x,complex) def not_decimal (x):return not isinstance(x,_dec) def not_da (x):return not isinstance(x,(dict,_adict)) def not_dict (x):return not isinstance(x,dict) def not_ds (x):return not isinstance(x,(dict,set)) def not_fc (x):return not isinstance(x,(float,complex)) def not_fcd (x):return not isinstance(x,(float,complex,_dec)) def not_fd (x):return not isinstance(x,(float,_dec)) def not_float (x):return not isinstance(x,float) def not_frozenset (x):return not isinstance(x,frozenset) def not_id (x):return not (x).isidentifier() def not_if (x):return not isinstance(x,(int,float)) def not_ifc (x):return not isinstance(x,(int,float,complex)) def not_ifcd (x):return not isinstance(x,(int,float,complex,_dec)) def not_ifd (x):return not isinstance(x,(int,float,_dec)) def not_iio (x):return not isinstance(x,(int,_iob)) def not_int (x):return not isinstance(x,int) def not_iobase (x):return not isinstance(x,_iob) def not_iter (x):return not isinstance(x,_iter) def not_lio (x):return not isinstance(x,(list,_iob)) def not_list (x):return not isinstance(x,list) def not_ls (x):return not isinstance(x,(list,set)) def not_lt (x):return not isinstance(x,(list,tuple)) def not_ltsf (x):return not isinstance(x,(list,tuple,set,frozenset)) def not_ltsfd (x):return not isinstance(x,(list,tuple,set,frozenset,dict)) def not_num (x):return not isinstance(x,_num) def not_number (x):return not isinstance(x,_num) def not_sb (x):return not isinstance(x,(str,bytes)) def not_sbi (x):return not isinstance(x,(str,bytes,int)) def not_sbb (x):return not isinstance(x,(str,bytes,bytearray)) def not_sbbi (x):return not isinstance(x,(str,bytes,bytearray,int)) def not_seq (x):return not isinstance(x,_sequ) def not_set (x):return not isinstance(x,set) def not_sf (x):return not isinstance(x,(set,frozenset)) def not_si (x):return not isinstance(x,(str,int)) def not_str (x):return not isinstance(x,str) def not_tf (x):return not isinstance(x,(tuple,frozenset)) def not_tuple (x):return not isinstance(x,tuple) #-------.-------.-------.-------.-------.-------.-------.-------.-------.------- #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def not1(v,f='s',t=''): # default args returns 's' for plural """Return specified value if first argument is not one. Returns should commonly be str.""" return t if v == 1 else f plural = not1 #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def nzp(value=None,negative=None,zero=None,positive=None,other=None): """Return selected object based on sign of a number, or other object.""" if is_ifd(value): if value<0:return negative if value>0:return positive return zero return other #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def oneofin(many,here): """Return True if one of the items in arg 1 is in arg 2.""" for one in many: if one in here: return True return False #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def oprint(*args,**kwargs): """Print to stdout with flush.""" if 'file' not in kwargs: kwargs['file']=stdout if 'flush' not in kwargs: kwargs['flush']=True for x in (stderr,stdout): if x!=kwargs.get('file',None): x.flush() try: return print(*args,**kwargs) except BrokenPipeError: return #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def parpid(p): """Return parent PID in Linux as same type: int,float,str.""" i = int(p) if i<2: return 1 with open('/proc/'+str(i)+'/status') as o: for x in o: a,b = x.strip().split(None,1) if a=='PPid:': return type(p)(b) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def plural(value=1,one='',not_one='s'): """Syntactic sugar to return 's' for plural cases else '' or allow substituting other strings. function plural purpose syntactic sugar to return 's' for plural cases else '' or allow substituting other strings examples print('I see %d thing%s here'%(n,plural(n))) print('il y a %d chose%s ici'%(n,plural(n))) print('Det er %d element%s her'%(n,plural(n,'','er'))) print('There %s %d thing%s here'%(plural(n,'is','are'),n,plural(n))) """ value = float(value) return one if value==1 else not_one #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def putenv(n,v): """Put an envionment variable even if it is already set and return the previous value.""" if is_bb(n): ns = tostr(n) if is_bb(v): vs = tostr(v) p = environ[ns] if ns in environ: environ[ns] = vs if is_str(n): return p if is_bytes(n): return tobytes(p) if is_bytearray(n): return tobytearray(p) return None environ[ns] = vs return None put_env = putenv #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def pyconfig(*args,**kwargs): """Load variables from a config file in Python format into a dictionary.""" # a config file in Python format is a Python script in a file that sets up a name space # position arguments are path components of the config file to use # key word arguments are initial config name space variables that the config code can use if not args: raise ValueError('pyconfig: no file path in position arguments') exception_return = kwargs.pop('exception_return',None) while is_lt(args[0]): # if arg 1 is a list or tuple then use its items as the arguments args = args[0] if not args: raise ValueError(f'pyconfig: no items in {tn(args)} in argument 1') file_name = path.expanduser(path.join(*args)) if not path.exists(file_name): raise TypeError(f'pyconfig: config file {file_name!r} does not exist') if not path.isfile(file_name): raise TypeError(f'pyconfig: config file {file_name!r} is not a regular file') #### try: code = open(path.expanduser(path.join(*args))).read() # load the config file source if it exists if code: exec(code,kwargs) # exec() always returns None for x in ('__builtins__',): # do not return these kwargs.pop(x,None) else: # an empty file will not be run and the result will be an empty dictionary kwargs = {} except Exception: if exception_return is None: raise kwargs = exception_return return kwargs exec_file = pyconfig #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def rangeforever(start=0,increment=1): """Return a range iterator that will last far longer than the universe will exist.""" return range(start,8**128,increment) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def readline(*args,**kwargs): """Read one line from a named file.""" with open(''.join(args),'r',**kwargs) as this: return this.readline() #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def readlines(*args,**kwargs): """Read all lines from a named file.""" lines = kwargs.pop('lines',-1) with open(''.join(args),'r',**kwargs) as this: return this.readlines(lines=lines) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def real(n): """Return real (part if complex) of a number.""" return n.real if 'real' in dir(n) else n #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def runpipeline(pipeline=None,out=None,mode='xb',*,opts={},encoding='utf-8',skip_error_check=False): """Run a command pipeline (list/tuple of list/tuple of str/bytes/bytearray), not waiting for completion, and returning its output as a pipe file, unless the out= option is used to specify alternative output handling.""" #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # intended for POSIX/BSD/Unix/Linux class platforms # tested on Ubuntu Linux 18.04 with Python 3.6 # each pipeline must be a list/tuple of commands # each command must be a list/tuple of arguments # each argument must be a str/bytes/bytearray #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # opts= caller can modify pipeline Popen() options # skip_error_check= caller can skip error checks #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # check for errors as possible before starting any processes #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if not skip_error_check: # caller can skip error checks #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # here, check for errors in pipeline= (arg 1) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if pipeline is None: raise ValueError('pipeline is None, thus not provided, nothing to run') if pipeline in (False,True,...,list,tuple,bytes,bytearray,str): return pipeline # if pipeline is a list/tuple of... if is_lt(pipeline): if pipeline: # if pipeline is a if all(is_lt(cmd)for cmd in pipeline): if all(cmd for cmd in pipeline): if all(all(is_sbb(arg)for arg in cmd)for cmd in pipeline): pass # the pipeline appears to be layered as specified else: raise TypeError('pipeline (arg 1) must be a (list/tuple of list/tuple) of str/bytes/bytearray') else: raise ValueError('a command in the pipeline is empty') elif all(is_sbb(arg)for arg in pipeline): pipeline = [pipeline] else: raise TypeError('pipeline (arg 1) must be a (list/tuple of) list/tuple (of str/bytes/bytearray)') else: raise ValueError('the pipeline has no commands') elif is_sbb(pipeline): if pipeline: pipeline = [[pipeline]] else: raise ValueError('a single command arg is empty') else: raise TypeError('cmds (arg 1) must be a list/tuple (of list/tuple of str/bytes/bytearray)') #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # check for errors in out= (or arg 2) # the objective is to test for as many errors as is possible before running any processes. #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if is_lt(out): if not out: raise ValueError(f'empty {tn(out)} given in out={out!r}') elif len(out)==1 and out[0] in (str,bytes,bytearray): # if a sequence has only one item then make it be a direct value out=out[0] elif len(out)==2 and out[0] in (list,tuple) and out[1] in (str,bytes,bytearray): pass elif len(out)>2: raise ValueError(f'too many values given in {tn(out)} given in out={out!r}') else: raise ValueError(f'invalid value(s) given in {tn(out)} given in out={out!r}') if out is None: pass elif is_sbb(out): if out: fn = out else: raise ValueError(f'empty {tn(out)} given in out={out!r}') elif is_int(out): if out < 0: raise ValueError(f'invalid int for file descriptor out={out!r}') else: fd = out else: raise TypeError(f'unrecognized type {tn(out)} for out=') #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- else: pass #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # start all the processes #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if 'close_fds' not in opts: opts['close_fds']=True opts['stdout']=PIPE procs=[] for cmd in pipeline[:-1]: proc = Popen(cmd,**opts) procs.append(proc) opts['stdin'] = proc.stdout last = Popen(pipeline[-1],**opts) procs.append(last) pipe = last.stdout #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # return pipe from last process output while processes continue to run #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if out is None: return pipe #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # read all output into one single bytearray buffer #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- buffer = bytearray(b'') for x in range(2**48): count = pipe.readinto(buffer) if not count: break #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # wait for all processes to exit #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- for proc in procs: proc.wait() #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # if output is to be split into lines out will be 2-list # or 2-tuple with the container type and the value type # example: out=[tuple,str] caller wants a tuple of str # example: out=(list,bytes) caller wants a list of bytes #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if is_lt(out): # already tested above for len(out)==2 lines = [] for line in buffer.splitlines(): if out[1] is str: lines.append(str(line,encoding=encoding)) elif out[1] is bytes: lines.append(bytes(line)) else: lines.append(line) return out[0](lines) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # if output is to remain as one sequence (not be split into lines) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if out is str: buffer = str(buffer,encoding=encoding) elif out is bytes: buffer = bytes(buffer) elif out is bytearray: pass return buffer run = runpipeline runcmds = runpipeline runcmdpipeline = runpipeline runcommand = runpipeline runcommands = runpipeline #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def rotate_left(a,n=1): """Rotate a sequence (to lower index) by n positions.""" l=len(a) return (a+a)[l-n:l+l-n] rotatel = rotate_left #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def rotate_right(a,n=1): """Rotate a sequence (to higher index) by n positions.""" return (a+a)[n:n+len(a)] rotater = rotate_right rotate = rotate_right #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def s(i): # note that this function name is lower case for lower case 's' """Return 's' for plural appending if argument is not one. Useful in f-strings.""" return '' if i == 1 else 's' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def S(i): # note that this function name is upper case for upper case 'S' """Return 'S' for plural appending if argument is not one. Useful in f-strings.""" return '' if i == 1 else 'S' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def setenv(n,v): """Set an envionment variable only if it has not been set.""" if is_bb(n): n = tostr(n) if is_bb(v): v = tostr(v) if n not in environ or not environ[n]: environ[n] = v return environ[n] set_env = setenv #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def setenvs(**kwargs): """Set multiple environment variables from keyword arguments (or a **dictionary).""" p = {k:v for k,v in environ.items()} for k,v in kwargs,items(): environ[k] = v return p set_envs = setenvs #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def snake_to_camel(s): """Convert name from snake_form to CamelForm.""" if is_bytes(s): return tobytes(snake_to_camel(tostr(s))) if is_bytearray(s): return tobytearray(snake_to_camel(tostr(s))) return s[:2]+''.join(x.capitalize() or '_' for x in s.split('_'))[2:] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def sleep(offset=0,cycle=0,minimum=0,file=None): """Sleep to an offset time in a time cycle returning time to sleep (seconds in float).""" # arg 1: offset or how long to sleep if no cycle # arg 2: repeating period or cycle if non-zero # arg 3: minimum time to sleep (skip a cycle if needed) # example: sleep 10 minutes: sleep(600) # example: sleep to 10 minutes past the hour: sleep(600,3600) # example: like above but at least 2 minutes: sleep(600,3600,120) # example: sleep to the next noon: sleep(86400,43200) # note: negative values make no sense if cycle > 0: offset = (offset-secs()) % cycle while offset < minimum: offset += cycle elif offset < minimum: offset = minimum if offset > 0: if file: print('time.sleep(',repr(offset),end=' ',file=file,flush=True) time.sleep(offset) if file: print(')',file=file,flush=True) return offset nsleep = sleep wait = sleep zleep = sleep #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def sleep_to(to_time): """Sleep to a given absolute time (float seconds), returning the duration, which will be negative if time is in the past.""" try: to_time = int(to_time) except Exception: try: to_time = float(to_time) except Exception: raise TypeError('to time not valid as float or as int') duration = to_time - secs() if duration > 0: time.sleep(duration) elif duration < 0: raise ValueError('to time is in the past.') return duration sleepto = sleep_to #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def st_size(arg): """Return the size in bytes of the referenced file system object.""" if 'fileno' in dir(arg): return stat(arg.fileno()).st_size try: return stat(arg).st_size except (AttributeError,TypeError): raise TypeError('Type {type(arg).__name__} cannot reference a file system object.') #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def strint(s,c=None): # c: string of base digits """Convert a str in a specified base system to an int.""" if not c: c = digits57 b = len(c) if isinstance(s,type(c)): a = {ord(k):v for k,v in zip(c,range(b))} return sum(a.get(ord(x),0)*b**p for x,p in zip(reversed(s),range(b))) return None #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def strlist(s): """Convert str of characters to a list of ints.""" return [ord(x) for x in s] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def strtuple(s): """Convert str of characters to a tuple of ints.""" return tuple(ord(x) for x in s) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def take(seq,num=1): """Take {num} items out of mutable sequence {seq} and return them.""" out = seq[:num] seq[:num] = [] return out #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tfn(name=None,time=None): """Return a time related temporary file name.""" s = hex(ti(time=time))[2:].rjust(14,'0') if name: s = name+'.'+s return s #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ti(time=None): """Return a time related int.""" try: time = float(time) # time=None simply fails here except Exception: time = secs() return int(time*3906250) # magic number 3906250 gets the significant bits into the result #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tn(*a): """Return space separated names of types of the arguments as one string.""" return ' '.join(type(x).__name__ for x in a) typename = tn typestr = tn #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tobytearray(a): """Convert to bytearray, transparently (no encoding).""" # fails if ord() > 255 if is_str(a): return bytearray(ord(x)for x in a) if is_bytes(a): return bytearray(a) if is_lt(a): if all(is_int(x)for x in a): # all ints is easy return bytearray(a) if all(is_ifd(x)for x in a): # mix of all int, float, decimal return bytearray(int(x)for x in a) if all(is_sbb(x)for x in a): # mix of all str, bytes, bytearray return type(a)([tobytes(x)for x in a]) if is_bytearray(a,bytearray): return a raise TypeError(f'cannot convert {tn(a)} to bytearray') toba = tobytearray to_bytearray = tobytearray #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tobytes(a): """Convert to bytes, transparently (no encoding).""" # fails if ord(x) > 255 if is_str(a): return bytes(ord(x)for x in a) if is_lt(a): if all(is_int(x)for x in a): # all ints is easy return bytes(a) if all(is_ifd(x)for x in a): # mix of all int, float, decimal return bytes(int(x)for x in a) if all(is_sbb(x)for x in a): # mix of all str, bytes, bytearray return type(a)([tobytes(x)for x in a]) if is_bytes(a): return a if is_bytearray(a): return bytes(a) raise TypeError(f'cannot convert {tn(a)} to bytes') tobyte = tobytes tob = tobytes to_bytes = tobytes #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def todec(number,**kwargs): """Try converting a number string to a decimal, returning None if it fails. function todec purpose try converting a string or other number to decimal.Decimal, just returning None if it fails note bytes is not supported by decimal.Decimal so this function converts bytes to str argument 1 (str) digits to be converted with maybe one decimal point 1 (number) number to be converted. returns converted decimal number or None note test should be like this: if return_value is None: """ if is_bb(number): number = tostr(number) if is_str(number): number = number.replace(decsep,'').replace('$','') if len(number.split(decpnt)) not in (1,2): return None if 'error_return' in kwargs: try: return decimal.Decimal(number)+decimal.Decimal(0) except decimal.InvalidOperation: return kwargs['error_return'] else: return decimal.Decimal(number)+decimal.Decimal(0) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tolower(s): """Return object with strings converted to lower case.""" if isinstance(s,(str,bytes,bytearray)): return s.lower() elif isinstance(s,(list,tuple)): return type(s)(tolower(x)for x in s) elif isinstance(s.dict): return {k:tolower(v)for k,v in s.items()} else: return s #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tostr(a): """Convert to str, transparently (no decoding).""" if is_bb(a): return ''.join(chr(x)for x in a) if is_lt(a): if all(is_int(x)for x in a): return tostr(bytes(x for x in a)) if all(is_float(x)for x in a): return tostr(bytes(int(x)for x in a)) if all(is_str(x)for x in a): return ''.join(a) if all(is_sbb(x)for x in a): return type(a)(tostr(x)for x in a) if is_str(a): return a raise TypeError(f'cannot convert {tn(a)} to str') tos = tostr to_str = tostr #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def toupper(s): """Return object with strings converted to upper case.""" if isinstance(s,(str,bytes,bytearray)): return s.upper() elif isinstance(s,(list,tuple)): return type(s)(toupper(x)for x in s) elif isinstance(s.dict): return {k:toupper(v)for k,v in s.items()} else: return s #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tslocal(time=None,fmt=None): """Return a local time stamp string from a given UTC time or for now.""" try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ts_fmt return strftime(fmt,localtime(int(time))) ts = tslocal #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tsmlocal(time=None,sep='T',pnt=None,fmt=None): """Return a local time stamp string with milliseconds from a given UTC time or for now.""" try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ts_fmt if pnt is None: pnt = decpnt return strftime(fmt,localtime(int(time))).replace('-',sep)+pnt+str(int((1+time-int(time))*1000))[1:] tsmloc = tsmlocal tsm = tsmlocal #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tsmutc(time=None,sep='T',pnt=None,fmt=None): """Return a UTC time stamp string with milliseconds from a given UTC time or for now.""" try: time = float(time) except Exception: time = secs() if pnt is None: pnt = decpnt return strftime(fmt,gmtime(int(time))).replace('-',sep)+pnt+str(int((1+time-int(time))*1000))[1:] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tsnlocal(time=None,sep='T',pnt=None,fmt=None): """Return a local time stamp string with nanoseconds from a given UTC time or for now.""" try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ts_fmt if pnt is None: pnt = decpnt return strftime(fmt,localtime(int(time))).replace('-',sep)+pnt+str(int((1+time-int(time))*1000000000))[1:] tsnloc = tsnlocal tsn = tsnlocal #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tsnutc(time=None,sep='T',pnt=None,fmt=None): """Return a UTC time stamp string with nanoseconds from a given UTC time or for now.""" try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ts_fmt if pnt is None: pnt = decpnt return strftime(fmt,gmtime(int(time))).replace('-',sep)+pnt+str(int((1+time-int(time))*1000000000))[1:] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tsulocal(time=None,sep='T',pnt=None,fmt=None): """Return a local time stamp string with microseconds from a given UTC time or for now.""" try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ts_fmt if pnt is None: pnt = decpnt return strftime(fmt,localtime(int(time))).replace('-',sep)+pnt+str(int((1+time-int(time))*1000000))[1:] tsuloc = tsulocal tsu = tsulocal #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tsutc(time=None,fmt=None): """Return a UTC time stamp string from a given UTC time or for now.""" try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ts_fmt return strftime(fmt,gmtime(int(time))) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def tsuutc(time=None,sep='T',pnt=None,fmt=None): """Return a UTC time stamp string with microseconds.""" try: time = float(time) except Exception: time = secs() if fmt is None: fmt = ts_fmt if pnt is None: pnt = decpnt return strftime(fmt,gmtime(int(time))).replace('-',sep)+pnt+str(int((1+time-int(time))*1000000))[1:] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def uniqnum(): """Return a unique number for this process-run.""" return ti()*65536+os.getpid() #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def uniqstr(): """Return a unique number for this process-run as a string.""" return str(uniqnum()) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def unymdhms(*args): """Reverse ymdhms back to floating time.""" while len(args)==1: if is_lt(args[0]): args=args[0] else: raise TypeError("invalid type passed in single argument, expected list or tuple") l=len(args) if l<9: args = list(args)+[2000,1,1,0,0,0,0,-1,-1][l:] return mktime(args) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def validip(addr): """Return values indicating if the given IP address is valid for IPv4 and/or IPv6. function validip purpose return 4 if the given IP address is valid for IPv4 or return 6 if the given IP address is valid for IPv6 or return 0 if the given IP address is not valid for either IPv4 or IPv6. return 10 if the given IP address is ambiguous. argument 1 (str,bytes) that may be an IP address returns 4 or 6 for the address family if it is valid 0 if the address is not valid IPv4 nor valid IPv6 10 if the address is somehow valid for both IPv4 and IPv6 (ambiguous) """ return validipv4(addr)+validipv6(addr) ipversion = validip #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def validipv4(addr): """Return 4 if the given IP address is valid for IPv4 or 0 if not. function validipv4 argument 1 (str,bytes) that may be an IPv4 address purpose return 4 if the given IP address is valid for IPv4 or 0 if not returns 4 if the given IP address is valid for IPv4 or 0 if not """ addr = tostr(addr) if '.' not in addr: r = 0 else: try: inet_pton(AF_INET,addr) r = 4 except (OSError,IOError): r = 0 return r #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def validipv6(addr): """Return 6 if the given IP address is valid for IPv6 or 0 if not. function validipv6 argument 1 (str,bytes) that may be an IPv6 address purpose return 6 if the given IP address is valid for IPv6 or 0 if not returns 6 if the given IP address is valid for IPv6 or 0 if not """ addr = tostr(addr) if ':' not in addr: r = 0 else: try: inet_pton(AF_INET6,addr) r = 6 except (OSError,IOError): r = 0 return r #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def whole_to_int(v): """Return a whole number value (even if given as a string) as an int, else as a float.""" try: v = float(v) except: pass i = int(v) if i == v: return i else: return v #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def xft(a): """Interpret a string with 'true' or 'false', returning the interpreted value or the original.""" if is_sbb(a): a = tostr(a).lower() if a == 'false': return False if a == 'true': return True return a #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def xftn(a): """Interpret a string with 'true' or 'false' or 'none', returning the interpreted value or the original.""" if is_sbb(a): a = tostr(a).lower() if a.lower() == 'false': return False if a.lower() == 'true': return True if a.lower() == 'none': return None return a #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def xfloat(a): if is_sbb(a): try: return float(tostr(a)) except TypeError: pass return a #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def xint(a): if is_sbb(a): try: return int(tostr(a,0)) except TypeError: pass return a #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def xx(i,width,fillchar=' '): """Convert an int to a right justified hexadecimal string.""" if is__ifd(width): x = hex(i)[2:] return x if width<1 else x.rjust(int(width),fillchar) return i,width,fillchar xex = xx #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- _n={'dim','ei','ez','ikke','ingen','inno','innò','n','nao','não','ne','nee','nei','nein','no','non',} _y={'ano','bai','da','ie','iè','ja','jah','jawohl','jes','jo','kylla','oui','si','sí','sì','sim','y','yes',} def yes_or_no(s,yes=True,no=False,*,unknown=None): """Determine if a given string means yes/true or no/false function yes_or_no aliases oui_ou_non,ja_oder_nein,ja_of_nee,ja_eller_nej,ja_eller_nei purpose determine if a given string in various languages means yes or no arguments #1 the string or value to check, str, bytes, or bytearray no= alternate value to return for no, default False yes= alternate value to return for yes, default True unknown= alternate value to return for unknown returns None or the value of no= if string/value indicates no True or the value of yes= if string/value indicates yes None or the value of unknown= if string/value cannot be determined note a TypeError exception is raised if argument #1 is invalid note only a few European languages are represented here """ if isinstance(s,(str,bytes,bytearray)): s = ''.join(x for x in s).lower() if s in _n: return no elif s in _y: return yes try: return yes if int(0,s)&1 else no except Exception: return unknown if isinstance(s,int): return yes if s&1 else no if isinstance(s,bool): return yes if s else no if s and isinstance(s,(list,tuple)): return yes_or_no(s[0],no=no,yes=yes,unknown=unknown) raise TypeError('not str,bytes, bytearray, int, or bool') # Alias this function in a few languages. oui_ou_non = yes_or_no ja_oder_nein = yes_or_no ja_of_nee = yes_or_no ja_eller_nej = yes_or_no ja_eller_nei = yes_or_no jes_au_ne = yes_or_no jo_oder_nee = yes_or_no si_o_no = yes_or_no sim_ou_nau = yes_or_no #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ymdhms_str(frac=6,time=None,seps='--T::.',*,notime=False,nosep=False): """From time in seconds (float) return str like "yyyy-mm-ddThh:mm:ss.fracdigit".""" if not is_ifd(time): time = secs() if not is_str(seps): return '' time = float(time) if frac > 9: frac = 9 if frac < 1: frac = 1 if notime: s=strftime('%Y/%m/%d',localtime(time)) else: if frac and is_ifd(frac): frac=int(frac) s=strftime('%Y/%m/%d/%H/%M/%S/',localtime(time))+str(int((time%1)*10**frac)).rjust(frac,'0') else: s=strftime('%Y/%m/%d/%H/%M/%S',localtime(time)) i=iter(seps+'xxxxxxx') return ''.join(x+next(i).replace('x','')for x in s.split('/')) ymdhms = ymdhms_str #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ymdhmsa_int(time=None): """From time in seconds (float) return 7-list (list of int) of year,month,date,hour,minute,second,attosecond.""" if time is None or time<0: time = secs() return [x for x in localtime(time)[:6]]+[int(time%1*1000000000000000000)] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ymdhmsf_int(time=None): """From time in seconds (float) return 7-list (list of int) of year,month,date,hour,minute,second,femtosecond.""" if time is None or time<0: time = secs() return [x for x in localtime(time)[:6]]+[int(time%1*1000000000000000)] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ymdhmsm_int(time=None): """From time in seconds (float) return 7-list (list of int) of year,month,date,hour,minute,second,millisecond.""" if time is None or time<0: time = secs() return [x for x in localtime(time)[:6]]+[int(time%1*1000)] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ymdhmsn_int(time=None): """From time in seconds (float) return 7-list (list of int) of year,month,date,hour,minute,second,nanosecond.""" if time is None or time<0: time = secs() return [x for x in localtime(time)[:6]]+[int(time%1*1000000000)] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ymdhmsp_int(time=None): """From time in seconds (float) return 7-list (list of int) of year,month,date,hour,minute,second,picosecond.""" if time is None or time<0: time = secs() return [x for x in localtime(time)[:6]]+[int(time%1*1000000000000)] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def ymdhmsu_int(time=None): """From time in seconds (float) return 7-list (list of int) of year,month,date,hour,minute,second,microsecond.""" if time is None or time<0: time = secs() return [x for x in localtime(time)[:6]]+[int(time%1*1000000)] #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # function zfiles_exist # # purpose given a base uncompressed file name, return a list of # it and all compressesed variants that do exist # # this is for the caller to see all file choices of # uncomressed and compressed file names #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def zfiles_exist( name, *, exts=None, ): """Return a list of the given uncompressed file and compressed variants that exist.""" fs = [name] if os.path.exists(name) and os.path.isfile(name) else [] if not exts: exts = 'gz bz2 xz' for ext in exts.split(): n = name+'.'+ext if os.path.exists(n) and os.path.isfile(n): fs.append(n) return fs #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # function zopen_only_one # # purpose open one compressed or uncompressed file for reading if # just one exists #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def zopen_only_one( name=None, mode='r', compresslevel=None, *, buffering=-1, check=-1, closefd=True, compressname=None, encoding=None, errors=None, format=None, newline=None, tempname=False, ): """Open and return an uncompressed file or one of its compressed files if exactly one exists along with the file list.""" exts = kwargs.pop('exts',None) names = zfiles_exist( name, exts=exts, ) if len(names) != 1: raise FileNotFoundError('exactly one file not found out of {name!r} and all compressed forms') if 'mode' in kwargs: if 'r' not in kwargs['mode']: raise PermissionError('this function only supports opening with a read mode') return zopen( names[0], mode, compresslevel, buffering=buffering, check=check, closefd=closefd, compressname=compressname, encoding=encoding, errors=errors, format=format, newline=newline, tempname=tempname, ) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # class zopen # purpose class to open a file with compress as indicated by # the extension(s) of the file name. also, for writing, # there is support for creating a temporary new file and # moving that file to the intended name when closed # function zopen (default temporary files disabled) # function ztopen (default temporary files enabled) # versions this module requires Python3 version 3.6 or later # versions other than 3.6.9 have not been tested # versions up to 4.0 are expected to work #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- """A class to open files with compression based on file name extension(s), including support for temporary written files.""" #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- class zopen(io.IOBase): #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # class zopen (io.IOBase) # # method zopen.__init__ # # purpose open a file with compression based on file name extension(s) and creation # using a temporary name that is moved to the intended name when closed for write # all arguments can be used as keyword arguments # name, mode and compresslevel can be used as the first three positional arguments # # args and kwargs # # name= (str) is the actual name of the file to be opened for reading or writing # (io.IOBase) is open file object to be reopened for reading or writing # mode= (str) is the mode to choose the method of opening the file # compresslevel= (int) compression level (0..9) (default is what the compression library code sets) # tempname= (bool,str) is the choice for using a temporary name for writing # (bool) is the choice to use a temporary name that is renamed at close # (bool) False disables a temporary name # (bool) True enables a temporary name with the actual temporary name chosen by zopen() # (string,bytes,bytearray) enables a temporary name and is the actual temporary name specified # compress= (str) is an alternate name used to select a compression algorithm different than the name implies # (bool) False disables compression even if the file name extension implies compression #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def __init__( self, name=None, mode='r', compresslevel=None, *, buffering=-1, check=-1, closefd=True, compressname=None, encoding=None, errors=None, format=None, newline=None, tempname=False, ): """Open a named file with compression based on file name extension(s).""" #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # lzma_open # # each compression module open method is called differently # so these three functions exist to make them work alike to # the code that chooses which compression algorithm it uses # these functions are never called for non-compressed files #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def lzma_open(file,mode): import lzma if read: f = lzma.open( file, mode, encoding=encoding, errors=errors, newline=newline, ) else: f = lzma.open( file, mode, preset=compresslevel, encoding=encoding, errors=errors, newline=newline, ) return f #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # bz2_open # # each compression module open method is called differently # so these three functions exist to make them work alike to # the code that chooses which compression algorithm it uses # these functions are never called for non-compressed files #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def bz2_open(file,mode): import bz2 if read: f = bz2.open( file, mode, encoding=encoding, errors=errors, newline=newline ) else: f = bz2.open( file, mode, compresslevel=compresslevel, encoding=encoding, errors=errors, newline=newline, ) return f #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # gzip_open # # each compression module open method is called differently # so these three functions exist to make them work alike to # the code that chooses which compression algorithm it uses # these functions are never called for non-compressed files #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def gzip_open(file,mode): import gzip if read: f = gzip.open( file, mode, encoding=encoding, errors=errors, newline=newline, ) else: f = gzip.open( file, mode, compresslevel=compresslevel, encoding=encoding, errors=errors, newline=newline, ) return f #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # map file name extension (str) to compression open function above # these functions are never called for non-compressed files #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- zmap = dict( bz = bz2_open, bz2 = bz2_open, gz = gzip_open, xz = lzma_open, tbz = bz2_open, tgz = gzip_open, txz = lzma_open, ) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # zopen logic #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- cwd = os.getcwd() if not name: raise TypeError('no filename given') elif isinstance(name,(str,io.IOBase,int)): pass elif is_bb(name): # change bytes or bytearray to string so it is simpler to work with name = tostr(name) # not encoding else: m = f'file name (arg 1 or name=) is not a str\n(or bytes or bytearray or int or open file)\nname = {name!r}' raise TypeError(m) if mode is None: mode,read = 'r',True elif is_str(mode): pass elif is_bb(mode): # change bytes or bytearray to string so it is simpler to work with mode = tostr(mode) else: raise TypeError('mode (arg 2 or mode=) is not a str (or bytes or bytearray)\nmode = {mode!r}') if '+' in mode: raise ValueError(f'Update mode is not support by zopen(). Use the builtin open() function for update mode.') if is_str(mode): mode = mode.lower() for ch in mode: if ch not in 'abrtwx': raise ValuError(f'invalid mode, got {mode!r}') append = 'a' in mode binary = 'b' in mode create = 'x' in mode read = 'r' in mode text = 't' in mode write = 'w' in mode if not (write or append or create): # if not any write then always read read = True if read and (write or append or create): raise ValueError(f'invalid mode mixes read and write, got {mode!r}') if binary and text: raise ValueError(f'invalid mode mixes binary and text, got {mode!r}') if compresslevel is None: compresslevel = _zopen_default_compresslevel if is_sbb(compresslevel): if is_bb(compresslevel): compresslevel = tostr(compresslevel) if compresslevel.lower()[:2] == 'no': compresslevel = None else: compresslevel = float(compresslevel) if is_complex(compresslevel): compresslevel = compresslevel.real if is_fcd(compresslevel): compresslevel = int(1/2+compresslevel) if is_int(compresslevel): if compresslevel < 0 or compresslevel > 9: raise ValueError(f'compresslevel ({compresslevel}) is out of range (0..9)') else: raise TypeError(f'compresslevel is not a number, got {compresslevel!r}') if is_str(mode): if not (write or append or create): # read is the default mode,read = 'r',True if read and (write or append or create): raise ValueError(f'invalid mode mixes read and write, got {mode!r}') if binary and text: raise ValueError(f'invalid mode mixes binary and text, got {mode!r}') else: mode,read = 'r',True #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # if stdio requested by file name #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if name == '-': if (write or append or create): return sys.stdout else: return sys.stdin #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # if compressname specified, use it just to decide the compression stack #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- zname = name if is_bb(compressname): zname = _bb_to_str(compressname) elif is_str(compressname): zname = compressname #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # collect (into the list in variable zstack) the file name extensions from back to front while they are for compression #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- zstack = [] while '.' in zname: zname,ext = zname.rsplit('.',1) if ext not in zmap: break zstack.append(zmap[ext]) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # set tempname to bool (name generated here if true) or str (name is given) # if tempname is a str then that will be used as the temporary name add on # else if tempname is true, a str will be generated from the time # a few special words are recognized for tempname values #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if tempname: if is_sbb(tempname): if is_bb(tempname): tempname = tostr(tempname) tn = tempname.lower() if tn in ('false','non','not','no'): tempname = False elif tn in ('true','oui','si','yes'): tempname = True try: tempname = int(tn,0) except ValueError: pass elif is_bifcd(tempname): tempname = bool(tempname) else: raise TypeError(f'invalid type for tempname={tempname!r}') else: tempname = False if tempname and not write: raise ValueError(f'a temporary file name is supported only for write mode for file {name!r}') #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # first, open the requested file directly, without compression # then, compression will be stacked on top of this file object # this open needs to be binary if compression is happening #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- fmode = ('r' if read else 'w') + ('b' if (zorder or binary) else 't') if tempname: # can be True for implicit or a str for explicit self.realname = os.path.join(cwd,name) # full name in case CWD is changed before close does rename self.tempname = self.realname + ('' if is_str(tempname) else ('_' + str(uuid.uuid4()))) openfile = open( self.tempname, fmode, buffering=buffering, encoding=encoding, errors=errors, newline=newline, closefd=closefd, ) else: self.realname = None # not doing tempname self.tempname = None # not doing tempname openfile = open( name, fmode, buffering=buffering, encoding=encoding, errors=errors, newline=newline, closefd=closefd, ) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # open a compression layer for each compression extension at the end of the intended file name # the order of opening is the reverse of the order of the extensions # the first to be opened is at the farthest end #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- allfiles = [openfile] zlen = len(zorder) zcnt = 0 for zcall in zorder: zcnt += 1 if zcnt == zlen: zmode = ('r' if read else 'w') + ('t' if text else 'b') else: zmode = ('r' if read else 'w') + 'b' f = openfile openfile = zcall(openfile,zmode) allfiles[:0] = [openfile] # prepend #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # save important values and return #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- self.allfiles = allfiles return #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # end of zopen.__init__ #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # close also does rename when a temporary file is involved #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def close(self): for x in self.allfiles: result = x.close() if self.realname: # if a real name is present then a temp name was just closed and a rename is needed try: os.rename(self.tempname,self.realname) except OSError: exit(f'failed to rename temporary name {self.tempname!r} to real name {self.realname!r}') return result # in case close ever returns sometthing different than None def __getattr__(self,*args): # reflect attributes of the referred file object as if in this file object return getattr(self.allfiles[0],*args) def __enter__(self): # my lame attempt to support the with statement return self def __exit__(self,exc_type,exc_val,exc_tb): # and exiting the with statement return #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # to enable debug tools, set one of these environment variables # to a string that evaluates as true by the eval() function: # SFCdebug SFCdebugON SFCdebugMODE # SFCDEBUG SFCDEBUGON SFCDEBUGMODE # sfcdebug sfcdebugon sfcdebugmode #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # these functions operate only if debugging is enabled # pr() Do printing just like print(), default to stderr, and flush. # pv() Print variables, given their names, or report if undefined. # sl() Sleep a specified time after flushing stdout or file= file. #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # The pv() function is particularly helpful. Call pv() with one or # more arguments naming variables to be printed. its name and value # will be printed if the variable is defined in the caller's local or # global namespace. If the variable is not defined, then a one line # message is produced explaining that the variable is not defined. # To help with debugging, pv() includes the line number of the source # code it came from on each line of output. If the arument string begins with # a '#' character, then the remainder of the string is a comment, not a # variable. A prefix for each line may be given as a named option on # the call to pv() or given in an argument as a string that begins # with the '=' character. #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # To enable debugging by default, import "sfcdebug" before importing "sfc", or do this: # if 'SFCdebug' not in environ: # environ['SFCdebug'] = '1' #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if not _debugmode: def _do_nothing_(*a,**o): return pr = _do_nothing_ pv = _do_nothing_ sl = _do_nothing_ else: print('SFC debugging enabled',file=stderr,flush=True) # flush multiple files def _debug_flush(*a): stderr.flush() stdout.flush() for f in a: f.flush() return # do printing just like print() but with flush def pr(*args,**kwargs): o = kwargs.get('file',stderr) kwargs['file'] = o _debug_flush() try: print(*args, **kwargs, flush=True) except BrokenPipeError: pass return # print variables, given their names, or report if undefined def pv(*args,**kwargs): o,c = kwargs.get('file',stderr),kwargs.pop('prefix','') kwargs['file'] = o kwargs['flush'] = True _debug_flush() try: if not args: # if nothing to do print(**kwargs) return 0 f = currentframe() if 'f_back' not in dir(f): print(**kwargs) return 0 f = f.f_back l,g = f.f_locals,f.f_globals if 'f_lineno' in dir(f): p='line '+repr(f.f_lineno)+': ' else: p='unknown line: ' _debug_flush() # synchronize outputs for name in args: s = name.split('.') n = s.pop(0) if n[:1] in '!': # use what follows as a var name n = n[1:] elif n[:1] in '+=:': # use what follows as a prefix c = n[1:] continue elif n[:1] == '#': # use what follows as a comment print(p+c+'............. ', n[1:], **kwargs) continue elif not n: # null variable name print(p+c+'............. ', s[0], **kwargs) continue elif n in l: # var is local o,v = p+c+'... local var:',l[n] elif n in g: # var is global o,v = p+c+'.. global var:',g[n] else: # not local or global print(p+c+'............. ', repr(name), 'not assigned in local or global name space', **kwargs) continue while s: m = s.pop(0) if m in dir(v): v,n = getattr(v,m),n+'.'+m else: break if is_str(v) and len(v)<11: x=['['+hex(ord(x)+256)[-2:]+']' for x in v] print(o, repr(n), '=', repr(v), ' '.join(x), **kwargs) else: print(o, repr(n), '=', repr(v), **kwargs) continue except BrokenPipeError: pass return # sleep a specified time after flushing standard files def sl(sec,**kwargs): kwargs.get('file',stdout).flush() _debug_flush() time.sleep(float(sec)) return #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- def _main(args): text = """ This is a module to be imported, not a command. Nothing to see here but code. MOVE ALONG ... """ eprint('',*box(text.splitlines()[1:],padh=4),'',sep='\n') return #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- if __name__ == '__main__': try: result = _main(argv) except BrokenPipeError: result = 141 except KeyboardInterrupt: print(flush=True) result = 98 if result is None or result is True: result = 0 elif result is False: result = 1 exit(result) #-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.-------.------- # EOF