Reference Manual

import pycxsom as cx

t = cx.typing.Map2D(500, cx.typing.Array(3))

print('type is "{}"'.format(t))
print('  Is it a Type ?   {}'.format(isinstance(t, cx.typing.Type)))
print('  Is it a Scalar ? {}'.format(isinstance(t, cx.typing.Scalar)))
print('  Is it a Pos1D ?  {}'.format(isinstance(t, cx.typing.Pos1D)))
print('  Is it a Pos2D ?  {}'.format(isinstance(t, cx.typing.Pos2D)))
print('  Is it a Array ?  {}'.format(isinstance(t, cx.typing.Array)))
print('  Is it a Map   ?  {}'.format(isinstance(t, cx.typing.Map)))
print('  Is it a Map1D ?  {}'.format(isinstance(t, cx.typing.Map1D)))
print('  Is it a Map2D ?  {}'.format(isinstance(t, cx.typing.Map2D)))

import pycxsom as cx

type_descriptions = ['',                     # [EXCEPT] <- '' : Parse error : make('') : Empty type description
                     'Skalar',               # [EXCEPT] <- 'Skalar' : Parse error : parse_check_next('Skalar', ...) : 'Scalar' expected.
                     'Scalar',               # [  OK  ] <- 'Scalar'
                     'Pos1D',                # [  OK  ] <- 'Pos1D'
                     'Pos2D',                # [  OK  ] <- 'Pos2D'
                     'Pos3D',                # [EXCEPT] <- 'Pos3D' : Parse error : parse_DIM('3D') : This is not 1D or 2D.
                     'Array=10',             # [  OK  ] <- 'Array=10'
                     'Mop1D<Scalar>=100',    # [EXCEPT] <- 'Mop1D<Scalar>=100' : Parse error : parse_check_next('Mop1D<Scalar>=100', ...) : 'Map' expected.
                     'Map1D<Scalar>=100',    # [  OK  ] <- 'Map1D<Scalar>=100'
                     'Map1D<Pos1D>=100',     # [  OK  ] <- 'Map1D<Pos1D>=100'
                     'Map1D<Pos2D>=100',     # [  OK  ] <- 'Map1D<Pos2D>=100'
                     'Map1D<Array=10>=100',  # [  OK  ] <- 'Map1D<Array=10>=100'
                     'Map2D<Scalar>=100',    # [  OK  ] <- 'Map2D<Scalar>=100'
                     'Map2D<Pos1D>=100',     # [  OK  ] <- 'Map2D<Pos1D>=100'
                     'Map2D<Pos2D>=100',     # [  OK  ] <- 'Map2D<Pos2D>=100'
                     'Map2D<Array=10>=100',  # [  OK  ] <- 'Map2D<Array=10>=100'
                     'Map3D<Array=10>=100']  # [EXCEPT] <- 'Map3D<Array=10>=100' : Parse error : parse_DIM('3D<Array=10>=100') : This is not 1D or 2D.

for descr in type_descriptions:
    msg = "[  OK  ]"
    error = ''
    try:
        t = cx.typing.make(descr)
        if descr != str(t):
            msg = "[FAILED]"
    except cx.error.Any as err:
            msg = "[EXCEPT]"
            error = str(err)
    display = "{} <- '{}'".format(msg, descr)
    if msg == "[FAILED]":
        display += " instead of " + str(t)
    elif msg == "[EXCEPT]":
        display += " : " + str(error)
    print(display)
            
    

import os
import sys
import pycxsom as cx
import numpy as np

# The variable is stored in a .var file. The path is built up from the
# root directory, wher all variables of a simulation are stored, a
# timeline (i.e. the first level of subdirectory), and a name, which
# can contain subdirectories as well.

if len(sys.argv) < 5:
    print()
    print('Usage:')
    print('  {} <root_dir> <timeline> <varname> COMMAND'.format(sys.argv[0]))
    print('where COMMAND is one of the following:')
    print('  realize <type> <cache_size> <file_size> // Realizes/creates the variable.')
    print('  range                                   // Shows the time range')
    print('  type                                    // Shows the variable type')
    print('  dump                                    // Shows the file content')
    print('  *@ <at>                                 // Reads a value')
    print('  +@ <at> <value>                         // Writes a value')
    print('  ++ <value>                              // Writes a at next timestep')
    print()
    print('Nota: The position <at> can be negative.')
    print()
    sys.exit(0)

root_dir = sys.argv[1]
timeline = sys.argv[2]
varname  = sys.argv[3]
command  = sys.argv[4]
args     = 5
nb_cmd_args = len(sys.argv) - 5

var_path = cx.variable.path_from(root_dir, timeline, varname)

print('File : {}'.format(var_path))


if command == 'realize':
    if nb_cmd_args != 3:
        print()
        print('Command usage: realize <type> <cache_size> <file_size>')
        sys.exit(1)
    var_type   = cx.typing.make(sys.argv[args + 0])
    cache_size = int(sys.argv[args + 1])
    file_size  = int(sys.argv[args + 2])
    # This creates/checks the file. After this call,
    # the object v is associated to the file.
    v = cx.variable.Realize(var_path, var_type, cache_size, file_size)
    print('done')
elif command == 'dump':
    print()
    for t, value in cx.variable.data_range_full(var_path) :
        print('{} @{}'.format(value, t))
elif command == 'type':
    with cx.variable.Realize(var_path) as v:
        print(v.datatype)
else:
    if command not in ['range', '*@', '+@', '++']:
        print()
        print('Command "{}" is not implemented'.format(command))
        sys.exit(1)
            
    # We open the variable and work with it. It should exist,
    # since we do not provide any information.
    with cx.variable.Realize(var_path) as v:
        try:
            if command == 'range':
                if nb_cmd_args != 0:
                    print()
                    print('Command usage: range')
                    sys.exit(0)
                r = v.time_range()
                if r:
                    print('Time range is [{}, {}]'.format(r[0], r[1]))
                else:
                    print('The file holds no data.')
            if command == '*@':
                if nb_cmd_args != 1:
                    print()
                    print('Command usage: *@ <at>')
                    sys.exit(0)
                at = int(sys.argv[args + 0])
                print('got {}'.format(v[at]))
            if command == '+@':
                if nb_cmd_args != 2:
                    print()
                    print('Command usage: *@ <at> <value>')
                    sys.exit(0)
                at    = int  (sys.argv[args + 0])
                value = float(sys.argv[args + 1])
                v[at] = np.full(v.datatype.shape(), value, dtype=float)
                print('done')
            if command == '++':
                if nb_cmd_args != 1:
                    print()
                    print('Command usage: ++ <value>')
                    sys.exit(0)
                value = float(sys.argv[args + 0])
                v += np.full(v.datatype.shape(), value, dtype=float)
                print('done')
        except cx.error.Busy:
            print('Nothing can be read, slot {} is busy'.format(at))
        except cx.error.Ready:
            print('Nothing can be written, slot {} is already ready'.format(at))
        except cx.error.Forgotten:
            print('Nothing can be accessed, slot {} is forgotten'.format(at))
        except cx.error.Any as err:
            print(err)
        except:
            print("Unexpected error:", sys.exc_info()[1])

import os
import numpy as np
import pycxsom as cx

# We remove an eventual 'dummy_dir' directory.
os.system("rm -rf dummy_dir")
os.system("mkdir dummy_dir")

def varpath(name):
    return cx.variable.path_from('dummy_dir', 'dummy', name)

# Let us write different kinds of variables.

with cx.variable.Realize(varpath('A'), cx.typing.make('Scalar'), 1, 1) as x :
    x += 12.345
    
with cx.variable.Realize(varpath('B'), cx.typing.make('Pos1D'), 1, 1) as x :
    x += 12.345
    
with cx.variable.Realize(varpath('C'), cx.typing.make('Pos2D'), 1, 1) as x :
    value = np.zeros(x.datatype.shape(), dtype=float)
    value[0] = 12.345
    value[1] = 67.89
    x += value
    
with cx.variable.Realize(varpath('D'), cx.typing.make('Array=10'), 1, 1) as x :
    value = np.arange(10, dtype=float).reshape(x.datatype.shape())
    x += value
    
with cx.variable.Realize(varpath('E'), cx.typing.make('Map1D<Scalar>=10'), 1, 1) as x :
    value = np.arange(10, dtype=float).reshape(x.datatype.shape())
    x += value
    
with cx.variable.Realize(varpath('F'), cx.typing.make('Map1D<Pos1D>=10'), 1, 1) as x :
    value = np.arange(10, dtype=float).reshape(x.datatype.shape())
    x += value
    
with cx.variable.Realize(varpath('G'), cx.typing.make('Map1D<Pos2D>=5'), 1, 1) as x :
    value = np.arange(10, dtype=float).reshape(x.datatype.shape())
    x += value
    
with cx.variable.Realize(varpath('H'), cx.typing.make('Map1D<Array=3>=5'), 1, 1) as x :
    value = np.arange(15, dtype=float).reshape(x.datatype.shape())
    x += value
    
with cx.variable.Realize(varpath('I'), cx.typing.make('Map2D<Scalar>=5'), 1, 1) as x :
    value = np.arange(25, dtype=float).reshape(x.datatype.shape())
    x += value
    
with cx.variable.Realize(varpath('J'), cx.typing.make('Map2D<Pos1D>=5'), 1, 1) as x :
    value = np.arange(25, dtype=float).reshape(x.datatype.shape())
    x += value
    
with cx.variable.Realize(varpath('K'), cx.typing.make('Map2D<Pos2D>=5'), 1, 1) as x :
    value = np.arange(50, dtype=float).reshape(x.datatype.shape())
    x += value
    
with cx.variable.Realize(varpath('L'), cx.typing.make('Map2D<Array=3>=5'), 1, 1) as x :
    value = np.arange(75, dtype=float).reshape(x.datatype.shape())
    x += value

# Now, let us display the values we have stored.

for varname in ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L']:
    with cx.variable.Realize(varpath(varname)) as x :
        print('### {} : {}'.format(varname, x.datatype))
        print(x[0])
        print()

import os
import sys
import time
import pycxsom as cx

if (len(sys.argv) == 1) or (sys.argv[1] not in ['generate', 'wait', 'listen']):
    print()
    print('Usage: {} <mode>'.format(sys.argv[0]))
    print()
    print('There are 3 modes for running the example.')
    print('   generate -> This cleans all, and writes periodically the variable.')
    print('               Start with this, and run other modes in another terminal.')
    print('   wait     -> This waits for next variable change, prints it, and exits.')
    print('   listen   -> This loops and notifies any changes.')
    print()
    exit()

mode    = sys.argv[1]
varpath = os.path.join('dummy_dir', 'timeline', 'ints.var')
                          
if mode == 'generate':
    os.system("rm -rf dummy_dir")
    os.system("mkdir -p dummy_dir/timeline")
    with cx.variable.Realize(varpath, cx.typing.make('Scalar'), 10, 10) as x :
        i = 0
        while True:
            time.sleep(2.0)
            x += i
            i += 1
                          
if mode == 'wait':
    with cx.variable.Realize(varpath) as x :
        x.sync_init() # we record the current time, from which we wait for a modification.
        print('waiting for new value...')
        x.wait_next(sleep_duration=.01) # We scan every 10ms to check for a value change.
        print('... got {} !'.format(x[-1]))

class Listener:
    def __init__(self):
        pass
    def on_new_value(self, v):
        print('v = {}'.format(v[-1]))
        
if mode == 'listen':
    l = Listener()
    with cx.variable.Realize(varpath) as x :
        x.sync_init() # we record the current time, from which we wait the first.
        print('Listening...')
        x.listen(l.on_new_value, sleep_duration=.01) # We scan every 10ms to check for a value change.
    

import pycxsom as cx
import numpy as np
import tkinter as tk
import matplotlib as plt
import sys

# This viewer draws a sine curve. It inherits from cx.tkviewer. At,
# meaning that it is invoked when we need to draw something at a given
# timestep. This timestep is providing to the on_draw_at callback,
# that you have to override. Afterwards, when building is done, an
# history slider can be defined as the one providing our viewer with
# time step information.

class MyView(cx.tkviewer.At):
    def __init__(self, master, title, color, figsize=(5, 4), dpi=100):
        super().__init__(master, title, figsize, dpi)
        self.X = np.linspace(0, 5*np.pi, 200)
        self.color = color
        
    def on_draw_at(self, at):
        Y = np.sin(self.X + .1*at)
        self.fig.clear()
        ax = self.fig.gca()
        ax.set_ylim((-1.1, 1.1))
        ax.plot(self.X, Y, color=self.color)
    
root = tk.Tk()
root.protocol('WM_DELETE_WINDOW', lambda : sys.exit(0))

hs   = cx.tkviewer.HistorySlider(root, 'Main History', 0, 100, 50)
hbox = tk.Frame(root)
    
v1 = MyView(hbox, 'v1', '#ff0000')
v2 = MyView(hbox, 'v2', '#0000ff')
    
hs.widget().pack(side=tk.TOP, fill=tk.BOTH)
hbox.pack(side=tk.TOP, fill=tk.BOTH)
v1.widget().pack(side=tk.LEFT, fill=tk.BOTH)
v2.widget().pack(side=tk.LEFT, fill=tk.BOTH)

v1.set_history_slider(hs)
v2.set_history_slider(hs)

tk.mainloop()

import pycxsom as cx
import numpy as np
import tkinter as tk
import matplotlib as plt
import sys

# This viewer draws a sine curve, whoe phase changes each time the
# user pressed the "Refresh" button. We only have to override the
# on_draw method in order to redraw when refresh is pressed.

class MyView(cx.tkviewer.Refresh):
    def __init__(self, master, title, figsize=(5, 4), dpi=100):
        super().__init__(master, title, figsize, dpi)
        self.X = np.linspace(0, 5*np.pi, 200)
        self.phase = 0
        self.draw() # This forces a first drawing
        
    def on_draw(self):
        Y = np.sin(self.X + self.phase)
        self.fig.clear()
        ax = self.fig.gca()
        ax.set_ylim((-1.1, 1.1))
        ax.plot(self.X, Y)
        self.phase += .1
        if self.phase > 2*np.pi:
            self.phase -= 2*np.pi
    
root = tk.Tk()
root.protocol('WM_DELETE_WINDOW', lambda : sys.exit(0))

v = MyView(root, 'Translating sine')
v.widget().pack(side=tk.LEFT, fill=tk.BOTH)

tk.mainloop()

import sys
import pycxsom as cx
import numpy as np
import tkinter as tk
import matplotlib as plt


if len(sys.argv) < 2:
    print()
    print('Usage:')
    print('  {} <varpath1> <varpath2> ... '.format(sys.argv[0]))
    print()
    sys.exit(0)

# Let us check that all the variables have a 1D shape.
varpaths = sys.argv[1:]
error = False
for var_path in varpaths :
    with cx.variable.Realize(var_path) as v:
        if isinstance(v.datatype, cx.typing.Map1D) and v.datatype.content.shape()[0] == 1 :
            pass
        else:
            print('{} contains type {} which is not a 1D map of scalars.'.format(var_path, v.datatype))
            error = True
if error:
    sys.exit(1)
    
    
# Let us launch a graphical interface

# This is the main frame
root = tk.Tk()
root.protocol('WM_DELETE_WINDOW', lambda : sys.exit(0))

# This is a time slider, associated to the first variable varpaths[0]
slider = cx.tkviewer.HistoryFromVariableSlider(root,
                                               'from {}'.format(varpaths[0]),
                                               varpaths[0])
slider.widget().pack(fill=tk.BOTH, side=tk.TOP)

# Now, let us add a viewer. The idea is to inherit from a viewer
# class, and override your drawing function. Here, we want the viewer
# to handle the variables to be displayed.
class MyViewer(cx.tkviewer.At):
    def __init__(self, master, title, varpaths, figsize=(5, 4), dpi=100):
        super().__init__(master, title, figsize, dpi)
        self.varpaths = varpaths

    # This is the inherited (and overrided) method for drawing
    def on_draw_at(self, at):
        self.fig.clear()    # self.fig is the matplotlib figure
        ax = self.fig.gca()
        nb_curves = 0
        for varpath in self.varpaths:
            _, timeline, name = cx.variable.names_from(varpath)
            with cx.variable.Realize(varpath) as v:
                try:
                    Y = v[at]
                    X = np.arange(len(Y))
                    ax.plot(X, Y, label='({}){}'.format(timeline, name))
                    nb_curves += 1
                except cx.error.Busy:
                    pass
                except cx.error.Forgotten:
                    pass
        if nb_curves > 0:
            ax.legend()

# We add an instance of our viewer in the GUI.
viewer = MyViewer(root, 'My viewer', varpaths)
viewer.widget().pack(fill=tk.BOTH, side=tk.TOP)
viewer.set_history_slider(slider) # This viewer is controlled by our slider.

# Then we start the GUI.
tk.mainloop()

mylogin@mymachine:~$ cxsom-ping <hostname> <port>

import sys
import pycxsom as cx


if len(sys.argv) < 3:
    print()
    print('Usage:')
    print('  {} <hostname> <port> ... '.format(sys.argv[0]))
    print()
    sys.exit(0)

hostname = sys.argv[1]
port     = int(sys.argv[2])

error = cx.client.ping(hostname, port)
if error :
    print('Error : {}'.format(error))
else:
    print('Done')

    

import matplotlib.pyplot as plt
import numpy as np

import pycxsom as cx

side = 100

r = np.linspace(0, 1, side, endpoint = True)
X = np.repeat(r.reshape(1,side), side, axis=0)
Y = X.T

# This is the data you could get from Map2d<Scalar>=100
# or a Map2d<Pos1D>=100
W_2d_Pos1d = .5*(X+Y)

# This is the data you could get from Map2d<Pos2D>=100
W_2d_Pos2d = np.dstack((Y,X))

# Let us plot the weigh grids.

fig = plt.figure(figsize=(5,5))
plt.title('Grid of scalars')
cx.plot.grid_of_1D(fig.gca(), W_2d_Pos1d)

fig = plt.figure(figsize=(5,5))
plt.title('Grid of pairs in [0,1]')
cx.plot.grid_of_2D(fig.gca(), W_2d_Pos2d)

plt.show()