fdi.utils package
Submodules
fdi.utils.checkjson module
fdi.utils.common module
- fdi.utils.common.attrstr(p, v, missingval='', ftime=False, state=True, width=1, **kwds)[源代码]
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)[源代码]
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)[源代码]
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)[源代码]
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)[源代码]
Generates a set of strings for param.toString().
- Param
Parameeter or xDstaset.
- Extra
Whether to include less often used attributes suc as
`fits_keyword`
.
- fdi.utils.common.findShape(data, element_seq=<class 'str'>)[源代码]
Shape of list/dict of list/dict.
- Element_seq
treat elements of these sequence types as scalars.
- fdi.utils.common.getObjectbyId(idn, lgbv)[源代码]
lgb is from deserializing caller’s globals().values() locals().values() and built-ins
- fdi.utils.common.getUidGid(username)[源代码]
returns the UID and GID of the named user.
return: -1 if not available
- fdi.utils.common.grouper(iterable, n, fillvalue=None)[源代码]
Collect data into fixed-length chunks or blocks
- fdi.utils.common.guess_value(input_string, parameter=False, last=<class 'str'>)[源代码]
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) |
- fdi.utils.common.ld2tk(v)[源代码]
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)[源代码]
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)[源代码]
Makes a presentation string at a detail level.
‘tablefmt’ is needed to be passed in recursive calls under some conditions it is used.
- fdi.utils.common.pathjoin(*p)[源代码]
join path segments with given separater (default ‘/’). Useful when ‘' is needed.
- fdi.utils.common.wls(st, width=15, fill=None, unprintable='#')[源代码]
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.
- st
input string. If not a string,
`str(st)`
is used.- width
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. - unprintable
substitute unprintable characters with is. default is ‘#’.
fdi.utils.fetch module
- fdi.utils.fetch.fetch(paths, nested, re='', sep='/', exe=['is'], not_quoted=True)[源代码]
Use paths to access values of internal elements of a nested python object.
- Paths
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.
- Nested
a live nested data structure.
- Re
datapath representation for nested. Can be applied to reproduce the result.
- Exe
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.
- Not_quoted
the method-args string is not encoded with quote.
fdi.utils.fits module
fdi.utils.fits_kw module
- fdi.utils.fits_kw.getFitsKw(name, ndigits=2, extra=None)[源代码]
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:
Look up in the Param_Names table (inverse FITS_KEYWORDS table, if fails,
try the extra dictionary if provided. if fails,
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.
- name
the name of e.g. a parameter.
- ndigits
how many digits (right to left) to take maximum if name ends with digits. default 2. Raises ValueError if more than 7.
- extra
tuple of (fits_kw, parameter_name) tuples to provide more look-up dictionary that override values from the default one.
- returns
FITS keyword.
fdi.utils.getconfig module
- 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': 'http://10.0.10.114:9885/fdi/v0.12/e2e', 'e2e127': 'http://127.0.0.1:9885/fdi/v0.12/e2e', 'e2e5k': 'http://127.0.0.1:5000/fdi/v0.12/e2e'}, 'mqtt': {'host': 'x.x.x.x', 'passwd': 'bar', 'port': 31876, 'username': 'foo'}, 'node': {'host': '127.0.0.1', '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': '127.0.0.1', 'self_password': 'bar', 'self_port': 9883, 'self_username': 'foo', 'server_poolpath': '/tmp/data', 'serveruser': 'mh', 'timeout': 10, 'userclasses': ''}, force=False)[源代码]
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 infdi.pns.config
. Th config file directory can be modified by the environment variableCONF_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)[源代码]
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)
基类:
tuple
Create new instance of LeapSecond(utc, dTAI_UTC)
- property dTAI_UTC
Alias for field number 1
- property utc
Alias for field number 0
- fdi.utils.leapseconds.dTAI_UTC_from_tai(tai_time)[源代码]
UTC time = tai_time - dTAI_UTC_from_tai(tai_time).
- fdi.utils.leapseconds.dTAI_UTC_from_utc(utc_time)[源代码]
TAI time = utc_time + dTAI_UTC_from_utc(utc_time).
- fdi.utils.leapseconds.gps_to_tai(gps_time)[源代码]
Convert GPS time given as datetime object to TAI time.
- fdi.utils.leapseconds.gps_to_utc(gps_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)[源代码]
Extract leap seconds from tzfiles.
- fdi.utils.leapseconds.tai_to_gps(tai_time)[源代码]
Convert TAI time given as datetime object to GPS time.
- fdi.utils.leapseconds.tai_to_utc(tai_time)[源代码]
Convert TAI time given as datetime object to UTC time.
fdi.utils.loadfiles module
- fdi.utils.loadfiles.loadcsv(filepath, delimiter=',', header=0, return_dict=False, pad=None)[源代码]
Loads the contents of a CSV file into a list of tuples.
- Header
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.moduleloader module
https://stackoverflow.com/a/43573798/13472124
- class fdi.utils.moduleloader.MyLoader(filename)[源代码]
基类:
importlib.abc.Loader
- class fdi.utils.moduleloader.SelectiveMetaFinder[源代码]
基类:
importlib.abc.MetaPathFinder
Raise ExcludedModule exception if loading modules whose names have any of cls.exclude.
- exclude = []
fdi.utils.monokaimod module
- class fdi.utils.monokaimod.MonokaiMod[源代码]
基类:
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)[源代码]
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’: ‘0.0.0.0’, ‘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)[源代码]
基类:
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)[源代码]
- keepalive = 60
- logger = None
- subclient = None
- wait_for(client, msgType, period=0.1)[源代码]
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.tofits module
fdi.utils.totalsize module
https://code.activestate.com/recipes/577504/
- fdi.utils.totalsize.total_size(o, handlers={}, verbose=False)[源代码]
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.ydump module
- class fdi.utils.ydump.MyRepresenter(default_style=None, default_flow_style=None, dumper=None)[源代码]
基类:
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)[源代码]
基类:
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