fdi.utils package


fdi.utils.checkjson module

fdi.utils.checkjson.checkjson(obj, dbg=0, **kwds)[source]

seriaizes the given object and deserialize. check equality.


fdi.utils.common module

exception fdi.utils.common.UserOrGroupNotFoundError[source]

Bases: BaseException

fdi.utils.common.attrstr(p, v, missingval='', ftime=False, state=True, width=1, **kwds)[source]

generic string representation of an attribute of a parameter or dataset.

p: parameter object. v: name of parameter attribute. ‘_valid’, ‘_type’, ‘_default’, ‘_value’ (for Parameter) or ‘_data’ (dataset) missingval: string used when the parameter does not have the attribute. ftime: True means that attribute value will be FineTime if _type is ‘finetime’. state: The state validity of the parameter is returned in place of value, if the state is not in Ommitted_Valid_Rule_Names – ‘valid’, ‘range’, ‘’ or ‘default’.

fdi.utils.common.attrstr1(p, v, missingval='', ftime=False, state=True, width=1, **kwds)[source]

generic string representation of an attribute of a parameter or dataset.

p: parameter object. v: name of parameter attribute. ‘_valid’, ‘_type’, ‘_default’, ‘_value’ (for Parameter) or ‘_data’ (dataset) missingval: string used when the parameter does not have the attribute. ftime: True means that attribute value will be FineTime if _type is ‘finetime’. state: The state validity of the parameter is returned in place of value, if the state is not in Ommitted_Valid_Rule_Names – ‘valid’, ‘range’, ‘’ or ‘default’.

fdi.utils.common.binhexstring(val, typ_, width=0, v=None, p=None, level=0, **kwds)[source]

returns val in binary, hex, or string according to typ_.

val; list of validity descriptor entries. typ_: parameter type in DataTypes.

fdi.utils.common.bstr(x, length=0, tostr=True, quote="'", level=0, tablefmt='rst', tablefmt1='simple', tablefmt2='rst', width=0, heavy=True, yaml=False, **kwds)[source]

returns the best string representation. if the object is a string, return single-quoted; if has toString(), use it; else returns str(). Length limited by lls(lls)

fdi.utils.common.exprstrs(param, v='_value', extra=False, **kwds)[source]

Generates a set of strings for param.toString().


Parameeter or xDstaset.


Whether to include less often used attributes suc as `fits_keyword`.

fdi.utils.common.findShape(data, element_seq=<class 'str'>)[source]

Shape of list/dict of list/dict.


treat elements of these sequence types as scalars.


full class name with module name.


fdi.utils.common.getObjectbyId(idn, lgbv)[source]

lgb is from deserializing caller’s globals().values() locals().values() and built-ins


returns the UID and GID of the named user.

return: -1 if not available

fdi.utils.common.grouper(iterable, n, fillvalue=None)[source]

Collect data into fixed-length chunks or blocks

fdi.utils.common.guess_value(input_string, parameter=False, last=<class 'str'>)[source]

Returns guessed value from a string.

input | output |
`'None'` | None |
integer | int() |
float | float() |
`'True'`, `'False`` | True, False |
string starting with `'0x'` | hex() |
else | run `last`(input_string) |

convert lists to tuples in nested data structures


convert lists, to tuples and dicts to frozensets in nested data structures array.array is converted to (typecode, itemsize, size, ld2tk(0th element))

fdi.utils.common.lls(s, length=80)[source]

length-limited string.

Returns the str if len <= length or length <=3. Returns ‘begin…end’ if not.

fdi.utils.common.mstr(obj, level=0, width=1, excpt=None, indent=4, depth=0, tablefmt='rst', tablefmt1='simple', tablefmt2='rst', **kwds)[source]

Makes a presentation string at a detail level.

‘tablefmt’ is needed to be passed in recursive calls under some conditions it is used.


join path segments with given separater (default ‘/’). Useful when ‘' is needed.


convert tuples to lists in nested data structures


trace back

fdi.utils.common.wls(st, width=15, fill=None, unprintable='#')[source]

generates a string comtaining width-limited strings separated with ‘ ‘.

Identifies Line-breaks with str.splitlines https://docs.python.org/3.6/library/stdtypes.html#str.splitlines Removes trailing line-breaks.


input string. If not a string, `str(st)` is used.


if > 0 returns the str with ‘

‘ inserted every width chars. Or else return the input st. Default is 15. A CJK characters occupies 2 in widths.

substitute unprintable characters with is. default is ‘#’.

fdi.utils.fetch module

fdi.utils.fetch.fetch(paths, nested, re='', sep='/', exe=['is'], not_quoted=True)[source]

Use paths to access values of internal elements of a nested python object.


1). If given as a string, the string will be splitted with sep into a list of strings, then go on to 2); 2). If given as a list of strings, its 0th member is to match one of the first level of nested attributes, keys, or method names. If the list was made in 1) the 0th member will be converted to an integer if possible.

  • if the 0th member is a string and can be parsed by deserialize_args(), the result to used as te named method and its arguments.

  • if that fails, it will be taken as a string and check if there is a match in keys (members);

  • else search in attributes.


a live nested data structure.


datapath representation for nested. Can be applied to reproduce the result.

  1. A list of patterns which if found in the name of a method/function the matching method/function is allowed to run. 2) If one of the pattern is ‘*’, all methods/functions are allowed to run. 3) If a pattern starts with a ‘-’ then the matching method/function to the pattern (‘-’ removed) is not allowed to run (overriding previous rules.


the method-args string is not encoded with quote.

fdi.utils.fits module

fdi.utils.fits.add_header(meta, header)[source]
fdi.utils.fits.fits_dataset(hdul, ima_list)[source]

fdi.utils.fits_kw module

fdi.utils.fits_kw.getFitsKw(name, ndigits=2, extra=None)[source]

Returns the FITS keyword for a name.

If name ends with a digit, split name to a digital part consists if all digits on the right, and a “non-digital part” on the left. Take ndigits continuoud digits, counting from right, to form the “numeric-part”. The “non-digital part”, or the name if not endibg with digits, get the pre-translation according to:

  1. Look up in the Param_Names table (inverse FITS_KEYWORDS table, if fails,

  2. try the extra dictionary if provided. if fails,

  3. take key value

If name ends with a digit, append the “numeric-part” to the first 8 - min(ndigits,number of digits in name) (maximum) of characters from pre-transition, uppercased, to form the resukt;

else take a maximum of 8 characters from pre-transition. uppercased, to form the resukt.


the name of e.g. a parameter.


how many digits (right to left) to take maximum if name ends with digits. default 2. Raises ValueError if more than 7.


tuple of (fits_kw, parameter_name) tuples to provide more look-up dictionary that override values from the default one.


FITS keyword.

fdi.utils.getconfig module

class fdi.utils.getconfig.Instance[source]

Bases: object

get(name=None, conf='pns')[source]
fdi.utils.getconfig.getConfig(name=None, conf='pns', builtin={'api_base': '/fdi', 'api_version': 'v0.12', 'base_poolpath': '/tmp', 'baseurl': '/fdi/v0.12', 'defaultpool': 'default', 'logginglevel': 20, 'lookup': {'e2e10': '', 'e2e127': '', 'e2e5k': ''}, 'mqtt': {'host': 'x.x.x.x', 'passwd': 'bar', 'port': 31876, 'username': 'foo'}, 'node': {'host': '', 'password': 'bar', 'port': 9883, 'ro_password': 'only5%', 'ro_username': 'poolro', 'rw_password': 'k/p=0', 'username': 'foo'}, 'paths': {'inputdir': '/home/mh/pns/input', 'inputfiles': ['pns.cat', 'pns.pn'], 'outputdir': '/home/mh/pns/output', 'outputfiles': ['xycc.dat', 'atc.cc'], 'pnshome': '/home/mh/pns'}, 'ptsuser': 'mh', 'scheme': 'http', 'scripts': {'clean': ['/home/mh/pns/cleanPTS', ''], 'config': ['/home/mh/pns/configPTS', ''], 'init': ['/home/mh/pns/initPTS', ''], 'run': ['/home/mh/pns/runPTS', '']}, 'self_host': '', 'self_password': 'bar', 'self_port': 9883, 'self_username': 'foo', 'server_poolpath': '/tmp/data', 'serveruser': 'mh', 'timeout': 10, 'userclasses': ''}, force=False)[source]

Imports a dict named [conf]config.

The contents of the config are defined in the .config/[conf]local.py file. The contenss are used to update defaults in fdi.pns.config. Th config file directory can be modified by the environment variable CONF_DIR, which, if not given or pointing to an existing directory, is the process owner’s ~/.config directory.

name: if given the poolurl in poolurl_of is returned, else construct a pooluul ending with `/{name}` from the contents in dict <conf>config. Default `None`. conf: configuration ID. default ‘pns’, so the file is ‘pnslocal.py’.

fdi.utils.getconfig.make_pool(pool, conf='pns', wipe=False)[source]

Return a ProductStorage with given pool name or poolURL.

;name: PoolURL, or pool name (has no “://”), in which case a pool URL is made based on the result of getConfig(name=pool, conf=conf). Default is ‘’. :conf: passed to getconfig to determine which configuration. Default `pns`. :wipe: whether to delete everything in the pool first.

fdi.utils.images module

fdi.utils.leapseconds module

Get TAI-UTC difference in seconds for a given time using tzdata.

i.e., find the number of seconds that must be added to UTC to compute TAI for any timestamp at or after the given time[1].

>>> from datetime import datetime
>>> import leapseconds
>>> leapseconds.dTAI_UTC_from_utc(datetime(2005, 1, 1, tzinfo = timezone.utc))
datetime.timedelta(0, 32)
>>> leapseconds.utc_to_tai(datetime(2015, 7, 1, tzinfo = timezone.utc))
datetime.datetime(2015, 7, 1, 0, 0, 36, tzinfo=datetime.timezone.utc)
>>> leapseconds.tai_to_utc(datetime(2015, 7, 1, 0, 0, 36, tzinfo = timezone.utc))
datetime.datetime(2015, 7, 1, 0, 0, tzinfo=datetime.timezone.utc)
>>> leapseconds.tai_to_utc(datetime(2015, 7, 1, 0, 0, 35, tzinfo = timezone.utc)) # leap second
datetime.datetime(2015, 7, 1, 0, 0, tzinfo=datetime.timezone.utc)
>>> leapseconds.tai_to_utc(datetime(2015, 7, 1, 0, 0, 34, tzinfo = timezone.utc))
datetime.datetime(2015, 6, 30, 23, 59, 59, tzinfo=datetime.timezone.utc)

Python 2.6+, Python 3, Jython, Pypy support.

[1]: https://github.com/eggert/tz/blob/master/leap-seconds.list [2]: https://github.com/eggert/tz/blob/master/tzfile.h [3]: https://github.com/eggert/tz/blob/master/zic.c [4]: http://datacenter.iers.org/eop/-/somos/5Rgv/latest/16

class fdi.utils.leapseconds.LeapSecond(utc, dTAI_UTC)

Bases: tuple

Create new instance of LeapSecond(utc, dTAI_UTC)

property dTAI_UTC

Alias for field number 1

property utc

Alias for field number 0


UTC time = tai_time - dTAI_UTC_from_tai(tai_time).


TAI time = utc_time + dTAI_UTC_from_utc(utc_time).


Convert GPS time given as datetime object to TAI time.


Convert GPS time given as datetime object to UTC time.

fdi.utils.leapseconds.leapseconds(tzfiles=['/usr/share/zoneinfo/right/UTC', '/usr/lib/zoneinfo/right/UTC'], use_fallback=True)[source]

Extract leap seconds from tzfiles.


Convert TAI time given as datetime object to GPS time.


Convert TAI time given as datetime object to UTC time.


Convert UTC time given as datetime object to GPS time.


Convert UTC time given as datetime object to TAI time.

fdi.utils.loadfiles module

fdi.utils.loadfiles.loadMedia(filename, content_type='image/png')[source]
fdi.utils.loadfiles.loadcsv(filepath, delimiter=',', header=0, return_dict=False, pad=None)[source]

Loads the contents of a CSV file into a list of tuples.


the first header linea are taken as column headers if `header > 0`. If no column header given (default 0), `colN` where N = 1, 2, 3… are returned.

the second header linea are also recorded (usually units) if header ` > 1. :return_dict: if ```True`` returns `dict[colhd]=(col, unit). Default is ```False`.


Default is a list of (colhd, column, unit) tuplees.

fdi.utils.masked module

fdi.utils.masked.masked(value, mask)[source]

Returns the masked part of the value, high bit number of mask, how wide the mask is..

e.g. value=0b00101100 mask=0b00011100, the result is 0b011, 5, 3

fdi.utils.moduleloader module


class fdi.utils.moduleloader.MyLoader(filename)[source]

Bases: importlib.abc.Loader


Return a module to initialize and into which to load.

This method should raise ImportError if anything prevents it from creating a new module. It may return None to indicate that the spec should create the new module.

class fdi.utils.moduleloader.MyMetaFinder[source]

Bases: importlib.abc.MetaPathFinder

find_spec(fullname, path, target=None)[source]
class fdi.utils.moduleloader.SelectiveMetaFinder[source]

Bases: importlib.abc.MetaPathFinder

Raise ExcludedModule exception if loading modules whose names have any of cls.exclude.

exception ExcludedModule[source]

Bases: Exception

exclude = []
find_spec(fullname, path, target=None)[source]

Inserts the finder into the import machinery


Inserts the finder into the import machinery


fdi.utils.monokaimod module

class fdi.utils.monokaimod.MonokaiMod[source]

Bases: pygments.styles.monokai.MonokaiStyle

styles = {Token: '#f8f8f2', Token.Comment: 'bold italic #af2', Token.Comment.Hashbang: '', Token.Comment.Multiline: '', Token.Comment.Preproc: '', Token.Comment.PreprocFile: '', Token.Comment.Single: '', Token.Comment.Special: '', Token.Error: '#960050 bg:#1e0010', Token.Escape: '', Token.Generic: '', Token.Generic.Deleted: '#f92672', Token.Generic.Emph: 'italic', Token.Generic.Error: '', Token.Generic.Heading: '', Token.Generic.Inserted: '#a6e22e', Token.Generic.Output: 'bg:#022 ansibrightcyan', Token.Generic.Prompt: 'bold #f92672', Token.Generic.Strong: 'bold', Token.Generic.Subheading: '#75715e', Token.Generic.Traceback: '', Token.Keyword: '#66d9ef', Token.Keyword.Constant: '', Token.Keyword.Declaration: '', Token.Keyword.Namespace: '#f92672', Token.Keyword.Pseudo: '', Token.Keyword.Reserved: '', Token.Keyword.Type: '', Token.Literal: '#ae81ff', Token.Literal.Date: '#e6db74', Token.Literal.Number: '#ae81ff', Token.Literal.Number.Bin: '', Token.Literal.Number.Float: '', Token.Literal.Number.Hex: '', Token.Literal.Number.Integer: '', Token.Literal.Number.Integer.Long: '', Token.Literal.Number.Oct: '', Token.Literal.String: '#e6db74', Token.Literal.String.Affix: '', Token.Literal.String.Backtick: '', Token.Literal.String.Char: '', Token.Literal.String.Delimiter: '', Token.Literal.String.Doc: 'bold italic #af2', Token.Literal.String.Double: '', Token.Literal.String.Escape: '#ae81ff', Token.Literal.String.Heredoc: '', Token.Literal.String.Interpol: '', Token.Literal.String.Other: '', Token.Literal.String.Regex: '', Token.Literal.String.Single: '', Token.Literal.String.Symbol: '', Token.Name: '#f8f8f2', Token.Name.Attribute: '#a6e22e', Token.Name.Builtin: '', Token.Name.Builtin.Pseudo: '', Token.Name.Class: '#a6e22e', Token.Name.Constant: '#66d9ef', Token.Name.Decorator: '#a6e22e', Token.Name.Entity: '', Token.Name.Exception: '#a6e22e', Token.Name.Function: '#a6e22e', Token.Name.Function.Magic: '', Token.Name.Label: '', Token.Name.Namespace: '', Token.Name.Other: '#a6e22e', Token.Name.Property: '', Token.Name.Tag: '#f92672', Token.Name.Variable: '', Token.Name.Variable.Class: '', Token.Name.Variable.Global: '', Token.Name.Variable.Instance: '', Token.Name.Variable.Magic: '', Token.Operator: '#f92672', Token.Operator.Word: '', Token.Other: '', Token.Punctuation: '#f8f8f2', Token.Text: 'bg:#022 ansibrightcyan', Token.Text.Whitespace: ''}

Style definitions for individual token types.

fdi.utils.options module

fdi.utils.options.opt(ops, argv=None)[source]

Example: ops = [ {‘long’:’help’, ‘char’:’h’, ‘default’: false, ‘description’:’print help’}, {‘long’:’verbose’, ‘char’:’v’, ‘default’: false, ‘description’:’print info’}, {‘long’:’username=’, ‘char’:’u’, ‘default’: ‘foo’, ‘description’:’non-empty user name/ID’}, {‘long’:’password=’, ‘char’:’p’, ‘default’: ‘bar’, ‘description’:’password’}, {‘long’:’host=’, ‘char’:’i’, ‘default’: ‘’, ‘description’:’host IP/name’}, {‘long’:’port=’, ‘char’:’o’, ‘default’: 5000, ‘description’:’port number’} ]

Optionly use ‘OPTSTART’ in command line to start options an have ppreceeding ones ignored. Useful when application is invoked with other options, e.g. by pytest

argv: if provided will be used in plsce of sys.argv.

fdi.utils.queueworks module

class fdi.utils.queueworks.queuework2(topics, host=None, port=None, username=None, passwd=None, client_id=None, callback=None, qos=1, userdata=None, clean_session=None)[source]

Bases: object

init_args(topics, host=None, port=None, username=None, passwd=None, client_id=None, on_msg_callback=None, qos=1, clean_session=None, userdata=None)[source]
keepalive = 60
logger = None
on_connect(client, userdata, flags, rc)[source]
on_disconnect(client, userdata, rc)[source]
on_message(client, userdata, msg)[source]
on_publish(client, userdata, mid)[source]
on_subscribe(client, userdata, mid, granted_qos)[source]
process_mid(mid, where)[source]

if mid is already in mwa, check it; if not put it in

send(topics, text, conn=True)[source]
subclient = None
wait_for(client, msgType, period=0.1)[source]

Waiting for subscribe and publish to clear.

http://www.steves-internet-guide.com/subscribing-topics-mqtt-client/ return: 0 if successful.

fdi.utils.run_proc module

fdi.utils.run_proc.run_proc(cmd, as_user, pwdir, timeout)[source]

Execute a shell command and return status.

fdi.utils.tofits module

fdi.utils.tofits.add_header(meta, header)[source]
fdi.utils.tofits.fits_dataset(hdul, dataset_list, name_list=None)[source]
fdi.utils.tofits.toFits(data, file='', **kwds)[source]

convert dataset to FITS.


a list of Dataset or a BaseProduct.


‘’ for returning fits stream. string for file name. default ‘’.

fdi.utils.totalsize module



sum size of object & members.


Recursively iterate to sum size of object & members.

fdi.utils.totalsize.total_size(o, handlers={}, verbose=False)[source]

Returns the approximate memory footprint an object and all of its contents.

Automatically finds the contents of the following builtin containers and their subclasses: tuple, list, deque, dict, set and frozenset. To search other containers, add handlers to iterate over their contents:

handlers = {SomeContainerClass: iter,

OtherContainerClass: OtherContainerClass.get_elements}

fdi.utils.tree module

fdi.utils.tree.logger = <Logger fdi.utils.tree (DEBUG)>

Aaron Hall https://stackoverflow.com/a/59109706

fdi.utils.tree.tree(data_object, level=0, style='line', prefix='', seen=None)[source]

A recursive generator, given an object object will yield a visual tree structure line by line with each line prefixed by the same characters

fdi.utils.ydump module

class fdi.utils.ydump.MyRepresenter(default_style=None, default_flow_style=None, dumper=None)[source]

Bases: ruamel.yaml.representer.RoundTripRepresenter

yaml_representers = {<class 'NoneType'>: <function RoundTripRepresenter.represent_none>, <class 'str'>: <function SafeRepresenter.represent_str>, <class 'bytes'>: <function SafeRepresenter.represent_binary>, <class 'bool'>: <function SafeRepresenter.represent_bool>, <class 'int'>: <function SafeRepresenter.represent_int>, <class 'float'>: <function SafeRepresenter.represent_float>, <class 'list'>: <function SafeRepresenter.represent_list>, <class 'tuple'>: <function SafeRepresenter.represent_list>, <class 'dict'>: <function SafeRepresenter.represent_dict>, <class 'set'>: <function SafeRepresenter.represent_set>, <class 'ruamel.yaml.compat.ordereddict'>: <function SafeRepresenter.represent_ordereddict>, <class 'collections.OrderedDict'>: <function RoundTripRepresenter.represent_dict>, <class 'datetime.date'>: <function SafeRepresenter.represent_date>, <class 'datetime.datetime'>: <function SafeRepresenter.represent_datetime>, None: <function SafeRepresenter.represent_undefined>, <class 'ruamel.yaml.scalarstring.LiteralScalarString'>: <function RoundTripRepresenter.represent_literal_scalarstring>, <class 'ruamel.yaml.scalarstring.FoldedScalarString'>: <function RoundTripRepresenter.represent_folded_scalarstring>, <class 'ruamel.yaml.scalarstring.SingleQuotedScalarString'>: <function RoundTripRepresenter.represent_single_quoted_scalarstring>, <class 'ruamel.yaml.scalarstring.DoubleQuotedScalarString'>: <function RoundTripRepresenter.represent_double_quoted_scalarstring>, <class 'ruamel.yaml.scalarstring.PlainScalarString'>: <function RoundTripRepresenter.represent_plain_scalarstring>, <class 'ruamel.yaml.scalarint.ScalarInt'>: <function RoundTripRepresenter.represent_scalar_int>, <class 'ruamel.yaml.scalarint.BinaryInt'>: <function RoundTripRepresenter.represent_binary_int>, <class 'ruamel.yaml.scalarint.OctalInt'>: <function RoundTripRepresenter.represent_octal_int>, <class 'ruamel.yaml.scalarint.HexInt'>: <function RoundTripRepresenter.represent_hex_int>, <class 'ruamel.yaml.scalarint.HexCapsInt'>: <function RoundTripRepresenter.represent_hex_caps_int>, <class 'ruamel.yaml.scalarfloat.ScalarFloat'>: <function RoundTripRepresenter.represent_scalar_float>, <class 'ruamel.yaml.scalarbool.ScalarBoolean'>: <function RoundTripRepresenter.represent_scalar_bool>, <class 'ruamel.yaml.comments.CommentedSeq'>: <function RoundTripRepresenter.represent_list>, <class 'ruamel.yaml.comments.CommentedMap'>: <function RoundTripRepresenter.represent_dict>, <class 'ruamel.yaml.comments.CommentedOrderedMap'>: <function SafeRepresenter.represent_ordereddict>, <class 'ruamel.yaml.comments.CommentedSet'>: <function RoundTripRepresenter.represent_set>, <class 'ruamel.yaml.comments.TaggedScalar'>: <function RoundTripRepresenter.represent_tagged_scalar>, <class 'ruamel.yaml.timestamp.TimeStamp'>: <function RoundTripRepresenter.represent_datetime>}
class fdi.utils.ydump.MyYAML(_kw=<object object>, typ=None, pure=False, output=None, plug_ins=None)[source]

Bases: ruamel.yaml.main.YAML

_kw: not used, forces keyword arguments in 2.7 (in 3 you can do (*, safe_load=..) typ: ‘rt’/None -> RoundTripLoader/RoundTripDumper, (default)

‘safe’ -> SafeLoader/SafeDumper, ‘unsafe’ -> normal/unsafe Loader/Dumper ‘base’ -> baseloader

pure: if True only use Python modules input/output: needed to work as context manager plug_ins: a list of plug-in files

dump(data, stream=None, default_flow_style=False, width=80, allow_unicode=True, **kw)[source]
fdi.utils.ydump.ydump(od, stream=None, register=None, **kwds)[source]

YAML dump that outputs OrderedDict like dict.

fdi.utils.ydump.yinit(mapping=4, sequence=4, offset=2, register=None)[source]

Initializes YAML.