Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
set_var = {1, 2, '3'}
recursive_var = {'a': {'b': 123}, 'c': True}
comp_var = 1+2j
seri_var = pandas.Series([1,2,3,3,3,3])
''')
wait_for_idle(kc)
execute(kc=kc, code='''
%use Julia
%get num_var num_arr_var logic_var logic_arr_var char_var char_arr_var set_var list_var dict_var recursive_var comp_var seri_var
%dict -r
%put num_var num_arr_var logic_var logic_arr_var char_var char_arr_var set_var list_var dict_var recursive_var comp_var seri_var
%use sos
seri_var = list(seri_var)
''')
wait_for_idle(kc)
execute(kc=kc, code='''
%dict num_var num_arr_var logic_var logic_arr_var char_var char_arr_var set_var list_var dict_var recursive_var comp_var seri_var
''')
res = get_result(iopub)
#self.assertEqual(res['null_var'], None)
self.assertEqual(res['num_var'], 123)
self.assertEqual(list(res['num_arr_var']), [1,2,3])
self.assertEqual(res['logic_var'], True)
self.assertEqual(res['logic_arr_var'], [True, False, True])
self.assertEqual(res['char_var'], '1"23')
self.assertEqual(res['char_arr_var'], ['1', '2', '3'])
self.assertEqual(res['set_var'], {1, 2, '3'})
self.assertEqual(res['list_var'], [1,2,'3'])
self.assertEqual(res['dict_var'], {'a': 1, 'b': 2, 'c': '3'})
#self.assertEqual(res['mat_var'].shape, (2,2))
self.assertEqual(res['recursive_var'], {'a': {'b': 123}, 'c': True})
self.assertEqual(res['comp_var'], 1+2j)
def testMagicSessioninfo(self):
with sos_kernel() as kc:
# preview variable
execute(kc=kc, code='''
%use R
%use Python3
%use SoS
%sessioninfo
''')
wait_for_idle(kc)
b = 1
run:
touch a.txt
[B_1]
depends: 'a.txt'
[B_2]
print(b)
'''
with sos_kernel() as kc:
iopub = kc.iopub_channel
execute(kc=kc, code=script)
wait_for_idle(kc)
execute(kc=kc, code="b")
res = get_result(iopub)
self.assertEqual(res, 1)
def testGetPythonDataFrameFromOctave(self):
# Python -> Matlab/Octave
with sos_kernel() as kc:
iopub = kc.iopub_channel
# create a data frame
execute(kc=kc, code='''
import pandas as pd
import numpy as np
import scipy.io as sio
arr = np.random.randn(1000)
arr[::10] = np.nan
df = pd.DataFrame({'column_{0}'.format(i): arr for i in range(10)})
''')
clear_channels(iopub)
execute(kc=kc, code="%use Octave")
wait_for_idle(kc)
#_, stderr = assemble_output(iopub)
#self.assertEqual(stderr, '')
execute(kc=kc, code="%get df")
wait_for_idle(kc)
execute(kc=kc, code="display(size(df))")
stdout, _ = assemble_output(iopub)
def testMagicSet(self):
# test preview of remote file
with sos_kernel() as kc:
iopub = kc.iopub_channel
# preview variable
execute(kc=kc, code='''
%set
%set -v2
%set
%set -v1
''')
stdout, stderr = assemble_output(iopub)
self.assertEqual(stderr, '', 'Got {}'.format(stderr))
self.assertTrue('set' in stdout, 'Got {}'.format(stdout))
def testDirectExchangeWithPython3(self):
'''Test getting data from another instance of Python3 (bypassing sos)'''
with sos_kernel() as kc:
iopub = kc.iopub_channel
# create a data frame
execute(kc=kc, code='''
%use Python3
import numpy
d3var = numpy.array([1, 2, 3])
''')
_, stderr = assemble_output(iopub)
self.assertEqual(stderr, '', "GOT ERROR {}".format(stderr))
#
execute(kc=kc, code='''
%use P3 -l Python3
%get d3var --from Python3
''')
wait_for_idle(kc)
execute(kc=kc, code="print(d3var.shape[0])")
stdout, _ = assemble_output(iopub)
self.assertEqual(stdout.strip(), '3', 'Expect {}'.format(stdout))
execute(kc=kc, code="%use sos")
execute(kc=kc, code='''
import pandas as pd
import numpy as np
arr = np.random.randn(1000)
arr[::10] = np.nan
df = pd.DataFrame({'column_{0}'.format(i): arr for i in range(10)})
''')
clear_channels(iopub)
execute(kc=kc, code="%use JavaScript")
wait_for_idle(kc)
execute(kc=kc, code="%get df")
wait_for_idle(kc)
execute(kc=kc, code="Object.keys(df).length")
res = get_display_data(iopub)
self.assertEqual(res, '1000')
execute(kc=kc, code="%use sos")
wait_for_idle(kc)
def testMagicSandbox(self):
with sos_kernel() as kc:
# preview variable
execute(kc=kc, code='''
%sandbox
with open('test_blah.txt', 'w') as tb:
tb.write('a')
''')
wait_for_idle(kc)
self.assertFalse(os.path.exists('test_blah.txt'))
def testAutoSharedVars(self):
with sos_kernel() as kc:
iopub = kc.iopub_channel
execute(kc=kc, code="sos_null = None")
wait_for_idle(kc)
execute(kc=kc, code="sos_num = 123")
wait_for_idle(kc)
execute(kc=kc, code="%use R")
wait_for_idle(kc)
execute(kc=kc, code="sos_num")
res = get_display_data(iopub)
self.assertEqual(res, '[1] 123')
execute(kc=kc, code="sos_num = sos_num + 10")
wait_for_idle(kc)
execute(kc=kc, code="%use sos")
wait_for_idle(kc)
execute(kc=kc, code="sos_num")
res = get_display_data(iopub)
self.assertEqual(res, '133')
def testMagicRender(self):
with sos_kernel() as kc:
# preview variable
execute(kc=kc, code='''
%render
"""
# header
* item1
* item2
"""
''')
wait_for_idle(kc)