utils.py

importbase64
importdecimal
importjsonas_json
importsys
importre
fromfunctoolsimportreduce

from_plotly_utils.optional_importsimportget_module
from_plotly_utils.basevalidatorsimport (
ImageUriValidator,
copy_to_readonly_numpy_array,
is_homogeneous_array,
)


int8min=-128
int8max=127
int16min=-32768
int16max=32767
int32min=-2147483648
int32max=2147483647

uint8max=255
uint16max=65535
uint32max=4294967295

plotlyjsShortTypes= {
"int8": "i1",
"uint8": "u1",
"int16": "i2",
"uint16": "u2",
"int32": "i4",
"uint32": "u4",
"float32": "f4",
"float64": "f8",
}


defto_typed_array_spec(v):
"""
 Convert numpy array to plotly.js typed array spec
 If not possible return the original value
 """
v=copy_to_readonly_numpy_array(v)

# Skip b64 encoding if numpy is not installed,
# or if v is not a numpy array, or if v is empty
np=get_module("numpy", should_load=False)
ifnotnpornotisinstance(v, np.ndarray) orv.size==0:
returnv

dtype=str(v.dtype)

# convert default Big Ints until we could support them in plotly.js
ifdtype=="int64":
max=v.max()
min=v.min()
ifmax<=int8maxandmin>=int8min:
v=v.astype("int8")
elifmax<=int16maxandmin>=int16min:
v=v.astype("int16")
elifmax<=int32maxandmin>=int32min:
v=v.astype("int32")
else:
returnv

elifdtype=="uint64":
max=v.max()
min=v.min()
ifmax<=uint8maxandmin>=0:
v=v.astype("uint8")
elifmax<=uint16maxandmin>=0:
v=v.astype("uint16")
elifmax<=uint32maxandmin>=0:
v=v.astype("uint32")
else:
returnv

dtype=str(v.dtype)

ifdtypeinplotlyjsShortTypes:
arrObj= {
"dtype": plotlyjsShortTypes[dtype],
"bdata": base64.b64encode(v).decode("ascii"),
 }

ifv.ndim>1:
arrObj["shape"] =str(v.shape)[1:-1]

returnarrObj

returnv


defis_skipped_key(key):
"""
 Return whether the key is skipped for conversion to the typed array spec
 """
skipped_keys= ["geojson", "layer", "layers", "range"]
returnany(skipped_key==keyforskipped_keyinskipped_keys)


defconvert_to_base64(obj):
ifisinstance(obj, dict):
forkey, valueinobj.items():
ifis_skipped_key(key):
continue
elifis_homogeneous_array(value):
obj[key] =to_typed_array_spec(value)
else:
convert_to_base64(value)
elifisinstance(obj, list) orisinstance(obj, tuple):
forvalueinobj:
convert_to_base64(value)


defcumsum(x):
"""
 Custom cumsum to avoid a numpy import.
 """

def_reducer(a, x):
iflen(a) ==0:
return [x]
returna+ [a[-1] +x]

ret=reduce(_reducer, x, [])
returnret


classPlotlyJSONEncoder(_json.JSONEncoder):
"""
 Meant to be passed as the `cls` kwarg to json.dumps(obj, cls=..)

 See PlotlyJSONEncoder.default for more implementation information.

 Additionally, this encoder overrides nan functionality so that 'Inf',
 'NaN' and '-Inf' encode to 'null'. Which is stricter JSON than the Python
 version.

 """

defcoerce_to_strict(self, const):
"""
 This is used to ultimately *encode* into strict JSON, see `encode`

 """
# before python 2.7, 'true', 'false', 'null', were include here.
ifconstin ("Infinity", "-Infinity", "NaN"):
returnNone
else:
returnconst

defencode(self, o):
"""
 Load and then dump the result using parse_constant kwarg

 Note that setting invalid separators will cause a failure at this step.

 """
# this will raise errors in a normal-expected way
encoded_o=super(PlotlyJSONEncoder, self).encode(o)
# Brute force guessing whether NaN or Infinity values are in the string
# We catch false positive cases (e.g. strings such as titles, labels etc.)
# but this is ok since the intention is to skip the decoding / reencoding
# step when it's completely safe

ifnot ("NaN"inencoded_oor"Infinity"inencoded_o):
returnencoded_o

# now:
# 1. `loads` to switch Infinity, -Infinity, NaN to None
# 2. `dumps` again so you get 'null' instead of extended JSON
try:
new_o=_json.loads(encoded_o, parse_constant=self.coerce_to_strict)
exceptValueError:

# invalid separators will fail here. raise a helpful exception
raiseValueError(
"Encoding into strict JSON failed. Did you set the separators "
"valid JSON separators?"
 )
else:
return_json.dumps(
new_o,
sort_keys=self.sort_keys,
indent=self.indent,
separators=(self.item_separator, self.key_separator),
 )

defdefault(self, obj):
"""
 Accept an object (of unknown type) and try to encode with priority:
 1. builtin: user-defined objects
 2. sage: sage math cloud
 3. pandas: dataframes/series
 4. numpy: ndarrays
 5. datetime: time/datetime objects

 Each method throws a NotEncoded exception if it fails.

 The default method will only get hit if the object is not a type that
 is naturally encoded by json:

 Normal objects:
 dict object
 list, tuple array
 str, unicode string
 int, long, float number
 True true
 False false
 None null

 Extended objects:
 float('nan') 'NaN'
 float('infinity') 'Infinity'
 float('-infinity') '-Infinity'

 Therefore, we only anticipate either unknown iterables or values here.

 """
# TODO: The ordering if these methods is *very* important. Is this OK?
encoding_methods= (
self.encode_as_plotly,
self.encode_as_sage,
self.encode_as_numpy,
self.encode_as_pandas,
self.encode_as_datetime,
self.encode_as_date,
self.encode_as_list, # because some values have `tolist` do last.
self.encode_as_decimal,
self.encode_as_pil,
 )
forencoding_methodinencoding_methods:
try:
returnencoding_method(obj)
exceptNotEncodable:
pass
return_json.JSONEncoder.default(self, obj)

@staticmethod
defencode_as_plotly(obj):
"""Attempt to use a builtin `to_plotly_json` method."""
try:
returnobj.to_plotly_json()
exceptAttributeError:
raiseNotEncodable

@staticmethod
defencode_as_list(obj):
"""Attempt to use `tolist` method to convert to normal Python list."""
ifhasattr(obj, "tolist"):
returnobj.tolist()
else:
raiseNotEncodable

@staticmethod
defencode_as_sage(obj):
"""Attempt to convert sage.all.RR to floats and sage.all.ZZ to ints"""
sage_all=get_module("sage.all")
ifnotsage_all:
raiseNotEncodable

ifobjinsage_all.RR:
returnfloat(obj)
elifobjinsage_all.ZZ:
returnint(obj)
else:
raiseNotEncodable

@staticmethod
defencode_as_pandas(obj):
"""Attempt to convert pandas.NaT / pandas.NA"""
pandas=get_module("pandas", should_load=False)
ifnotpandas:
raiseNotEncodable

ifobjispandas.NaT:
returnNone

# pandas.NA was introduced in pandas 1.0
ifhasattr(pandas, "NA") andobjispandas.NA:
returnNone

raiseNotEncodable

@staticmethod
defencode_as_numpy(obj):
"""Attempt to convert numpy.ma.core.masked"""
numpy=get_module("numpy", should_load=False)
ifnotnumpy:
raiseNotEncodable

ifobjisnumpy.ma.core.masked:
returnfloat("nan")
elifisinstance(obj, numpy.ndarray) andobj.dtype.kind=="M":
try:
returnnumpy.datetime_as_string(obj).tolist()
exceptTypeError:
pass

raiseNotEncodable

@staticmethod
defencode_as_datetime(obj):
"""Convert datetime objects to iso-format strings"""
try:
returnobj.isoformat()
exceptAttributeError:
raiseNotEncodable

@staticmethod
defencode_as_date(obj):
"""Attempt to convert to utc-iso time string using date methods."""
try:
time_string=obj.isoformat()
exceptAttributeError:
raiseNotEncodable
else:
returniso_to_plotly_time_string(time_string)

@staticmethod
defencode_as_decimal(obj):
"""Attempt to encode decimal by converting it to float"""
ifisinstance(obj, decimal.Decimal):
returnfloat(obj)
else:
raiseNotEncodable

@staticmethod
defencode_as_pil(obj):
"""Attempt to convert PIL.Image.Image to base64 data uri"""
image=get_module("PIL.Image")
ifimageisnotNoneandisinstance(obj, image.Image):
returnImageUriValidator.pil_image_to_uri(obj)
else:
raiseNotEncodable


classNotEncodable(Exception):
pass


defiso_to_plotly_time_string(iso_string):
"""Remove timezone info and replace 'T' delimeter with ' ' (ws)."""
# make sure we don't send timezone info to plotly
if (iso_string.split("-")[:3] =="00:00") or (iso_string.split("+")[0] =="00:00"):
raiseException(
"Plotly won't accept timestrings with timezone info.\n"
"All timestrings are assumed to be in UTC."
 )

iso_string=iso_string.replace("-00:00", "").replace("+00:00", "")

ifiso_string.endswith("T00:00:00"):
returniso_string.replace("T00:00:00", "")
else:
returniso_string.replace("T", " ")


deftemplate_doc(**names):
def_decorator(func):
ifnotsys.version_info[:2] == (3, 2):
iffunc.__doc__isnotNone:
func.__doc__=func.__doc__.format(**names)
returnfunc

return_decorator


def_natural_sort_strings(vals, reverse=False):
defkey(v):
v_parts=re.split(r"(\d+)", v)
foriinrange(len(v_parts)):
try:
v_parts[i] =int(v_parts[i])
exceptValueError:
# not an int
pass
returntuple(v_parts)

returnsorted(vals, key=key, reverse=reverse)


def_get_int_type():
np=get_module("numpy", should_load=False)
ifnp:
int_type= (int, np.integer)
else:
int_type= (int,)
returnint_type


defsplit_multichar(ss, chars):
"""
 Split all the strings in ss at any of the characters in chars.
 Example:

 >>> ss = ["a.string[0].with_separators"]
 >>> chars = list(".[]_")
 >>> split_multichar(ss, chars)
 ['a', 'string', '0', '', 'with', 'separators']

 :param (list) ss: A list of strings.
 :param (list) chars: Is a list of chars (note: not a string).
 """
iflen(chars) ==0:
returnss
c=chars.pop()
ss=reduce(lambdax, y: x+y, map(lambdax: x.split(c), ss))
returnsplit_multichar(ss, chars)


defsplit_string_positions(ss):
"""
 Given a list of strings split using split_multichar, return a list of
 integers representing the indices of the first character of every string in
 the original string.
 Example:

 >>> ss = ["a.string[0].with_separators"]
 >>> chars = list(".[]_")
 >>> ss_split = split_multichar(ss, chars)
 >>> ss_split
 ['a', 'string', '0', '', 'with', 'separators']
 >>> split_string_positions(ss_split)
 [0, 2, 9, 11, 12, 17]

 :param (list) ss: A list of strings.
 """
returnlist(
map(
lambdat: t[0] +t[1],
zip(range(len(ss)), cumsum([0] +list(map(len, ss[:-1])))),
 )
 )


defdisplay_string_positions(p, i=None, offset=0, length=1, char="^", trim=True):
"""
 Return a string that is whitespace except at p[i] which is replaced with char.
 If i is None then all the indices of the string in p are replaced with char.

 Example:

 >>> ss = ["a.string[0].with_separators"]
 >>> chars = list(".[]_")
 >>> ss_split = split_multichar(ss, chars)
 >>> ss_split
 ['a', 'string', '0', '', 'with', 'separators']
 >>> ss_pos = split_string_positions(ss_split)
 >>> ss[0]
 'a.string[0].with_separators'
 >>> display_string_positions(ss_pos,4)
 ' ^'
 >>> display_string_positions(ss_pos,4,offset=1,length=3,char="~",trim=False)
 ' ~~~ '
 >>> display_string_positions(ss_pos)
 '^ ^ ^ ^^ ^'
 :param (list) p: A list of integers.
 :param (integer|None) i: Optional index of p to display.
 :param (integer) offset: Allows adding a number of spaces to the replacement.
 :param (integer) length: Allows adding a replacement that is the char
 repeated length times.
 :param (str) char: allows customizing the replacement character.
 :param (boolean) trim: trims the remaining whitespace if True.
 """
s= [" "for_inrange(max(p) +1+offset+length)]
maxaddr=0
ifiisNone:
forp_inp:
forlinrange(length):
maxaddr=p_+offset+l
s[maxaddr] =char
else:
forlinrange(length):
maxaddr=p[i] +offset+l
s[maxaddr] =char
ret="".join(s)
iftrim:
ret=ret[: maxaddr+1]
returnret


defchomp_empty_strings(strings, c, reverse=False):
"""
 Given a list of strings, some of which are the empty string "", replace the
 empty strings with c and combine them with the closest non-empty string on
 the left or "" if it is the first string.
 Examples:
 for c="_"
 ['hey', '', 'why', '', '', 'whoa', '', ''] -> ['hey_', 'why__', 'whoa__']
 ['', 'hi', '', "I'm", 'bob', '', ''] -> ['_', 'hi_', "I'm", 'bob__']
 ['hi', "i'm", 'a', 'good', 'string'] -> ['hi', "i'm", 'a', 'good', 'string']
 Some special cases are:
 [] -> []
 [''] -> ['']
 ['', ''] -> ['_']
 ['', '', '', ''] -> ['___']
 If reverse is true, empty strings are combined with closest non-empty string
 on the right or "" if it is the last string.
 """

def_rev(l):
return [s[::-1] forsinl][::-1]

ifreverse:
return_rev(chomp_empty_strings(_rev(strings), c))
ifnotlen(strings):
returnstrings
ifsum(map(len, strings)) ==0:
return [c* (len(strings) -1)]

class_Chomper:
def__init__(self, c):
self.c=c

def__call__(self, x, y):
# x is list up to now
# y is next item in list
# x should be [""] initially, and then empty strings filtered out at the
# end
iflen(y) ==0:
returnx[:-1] + [x[-1] +self.c]
else:
returnx+ [y]

returnlist(filter(len, reduce(_Chomper(c), strings, [""])))


# taken from
# https://en.wikibooks.org/wiki/Algorithm_Implementation/Strings/Levenshtein_distance#Python
deflevenshtein(s1, s2):
iflen(s1) <len(s2):
returnlevenshtein(s2, s1) # len(s1) >= len(s2)
iflen(s2) ==0:
returnlen(s1)
previous_row=range(len(s2) +1)
fori, c1inenumerate(s1):
current_row= [i+1]
forj, c2inenumerate(s2):
# j+1 instead of j since previous_row and current_row are one character longer
# than s2
insertions=previous_row[j+1] +1
deletions=current_row[j] +1
substitutions=previous_row[j] + (c1!=c2)
current_row.append(min(insertions, deletions, substitutions))
previous_row=current_row
returnprevious_row[-1]


deffind_closest_string(string, strings):
def_key(s):
# sort by levenshtein distance and lexographically to maintain a stable
# sort for different keys with the same levenshtein distance
return (levenshtein(s, string), s)

returnsorted(strings, key=_key)[0]