brica1 package

Submodules

brica1.component module

component.py

This module contains the Component, ConstantComponent, PipeComponent, and NullComponent. A Component is a unit of implementation which can exchange any type of data with another Component. `Component`s together with `Unit`s are collectively reffered to as `Unit`s.

class brica1.component.Component

Bases: brica1.unit.Unit

Component is an abstract class for implementation units. Subclasses must override the fire() method to specify its implementation. See the sample implementations, ConstantComponent, NullComponent, and PipeComponent for reference.

clear_result(identifier)

Clear a state value for the given ID.

Args:

identifier (str): a string ID.

Returns:

None.

clear_state(identifier)

Clear a state value for the given ID.

Args:

identifier (str): a string ID.

Returns:

None.

abstract fire()

Perform a calculation.

Users must override fire(self) method to define a new sub-class of Component. fire(self) method implements a function of the form

results, states <- fire(in_ports, states)

which states that this method should be a mutator of these two member variables (results and states), and the result of the computation should solely depend on the values stored in (in_ports and states). One important note here is that it shall mutate results member variable and not out_ports. Values stored in results will automatically be copied to out_ports by the scheduler which calls update_output() method of this Module, so these must be visible and accessible from other Modules. This procedure is especially necessary in concurrent execution of multiple Modules to avoid contention.

Args:

None.

Returns:

None.

get_result(identifier)

Get a result value for the given ID.

Args:

identifier (str): a string ID.

Returns:

any: a result value for the given ID.

get_state(identifier)

Get a state value for the given ID.

Args:

identifier (str): a string ID.

Returns:

any: a state value for the given ID.

input(time)

Obtain inputs from outputs of other Modules.

This method collects the outputs of connected modules and sets the values to the in-ports. It is usually called by the scheduler.

Args:

time (int): the scheduler’s current time.

Returns:

None.

output(time)

Expose results to out_ports

This method exposes the computation results from results to out_ports. It is usually called by the scheduler.

Args:

time (int): the scheduler’s current time.

Returns:

None.

reset()

Reset the component state

This method resets the internal time of the component for reuse.

Args:

None.

Returns:

None.

set_result(identifier, value)

Set a result value for the given ID.

Args:

identifier (str): a string ID. value (any): a result value to set.

Returns:

None.

set_state(identifier, value)

Set a state value for the given ID.

Args:

identifier (str): a string ID. vvalue (any): a state value to set.

Returns:

None.

train()

Perform a training iteration

Returns:

None.

class brica1.component.ComponentSet

Bases: brica1.component.Component

ComponentSet groups components to fire sequentially

add_component(identifier, component, priority)

Add a Component for given ID and priority

Args:

identifier (str): a string ID. component (Component): a component to add for identifier. priority (int): a priority value for scheduling.

Returns:

None.

fire()

Fire sub-components based on priority

Args:

None.

Returns:

None.

get_component(identifier)

Get a Component for given ID.

Args:

identifier (str): a string ID.

Returns:

Component: a Component corresponding to the ID.

class brica1.component.ConstantComponent

Bases: brica1.component.Component

ConstantComponent copies states to out ports.

Use set_state to define the output of this Module.

fire()

Copy state contents to results.

Args:

None.

Returns:

None.

class brica1.component.NullComponent

Bases: brica1.component.Component

NullComponent does nothing.

fire()

Do nothing. Just return.

Args:

None.

Returns:

None.

class brica1.component.PipeComponent

Bases: brica1.component.Component

PipeComponent copies contents of in ports to out ports.

fire()

Perform a calculation.

Users must override fire(self) method to define a new sub-class of Component. fire(self) method implements a function of the form

results, states <- fire(in_ports, states)

which states that this method should be a mutator of these two member variables (results and states), and the result of the computation should solely depend on the values stored in (in_ports and states). One important note here is that it shall mutate results member variable and not out_ports. Values stored in results will automatically be copied to out_ports by the scheduler which calls update_output() method of this Module, so these must be visible and accessible from other Modules. This procedure is especially necessary in concurrent execution of multiple Modules to avoid contention.

Args:

None.

Returns:

None.

set_map(in_identifier, out_identifier)

Map from in-port to out port.

Args:

in_identifier (str): port identifier to map from. out_identifier (str): port identifier to map to.

Returns:

None.

brica1.connection module

connection.py

This module contains the class Connection.

class brica1.connection.Connection(from_port, to_port)

Bases: object

A Connection connects two `Port`s and can be `sync`ed in one direction.

sync()

Sync the value from from_port to to_port.

Args:

None.

Returns:

None.

brica1.module module

module.py

This module contains the class Module and Agent which serve as abstractions for distict areas in the brain or sub-regions of those areas. `Module`s together with `Component`s are collectively reffered to as `Unit`s.

class brica1.module.Agent

Bases: brica1.module.Module

A Agent is a Module which serves as a top-level container for functional `Module`s.

class brica1.module.Module

Bases: brica1.unit.Unit

A Module may not have an implementation and may only exchange numpy.ndarray of the type numpy.short through in/out ports with another module.

add_component(id, component)

Add a Component to this Module.

Args:

id (str): a string ID. component (Component): a component to add for id.

Returns:

None.

add_submodule(id, submodule)

Add a Module to this Module.

Args:

id (str): a string ID. submodule (Module): a module to add for id.

Returns:

None.

get_all_components()

Get all `Component`s of all `Module`s.

Args:

None.

Returns:

array: a array of all `Component`s.

get_all_submodules()

Get all `Module`s recursively.

Args:

None.

Returns:

array: a array of all `Module`s.

get_component(id)

Get a Component for a given id.

Args:

id (str): a string ID.

Returns:

Component: a component for the given id.

get_submodule(id)

Get a Module for a given id.

Args:

id (str): a string ID.

Returns:

Module: a module for the given id.

remove_component(id)

Remove a component from this Module.

Args:

id (str): a string ID.

Returns:

None.

remove_submodule(id)

Remove a module from this Module.

Args:

id (str): a string ID.

Returns:

None.

brica1.port module

port.py

This module contains the class Port.

class brica1.port.Port(value)

Bases: object

A Port has a buffer value and a outward connection to another port. There may only be one outward connection but multiple inward connections.

connect(target)

Create a connection to the target Port.

Args:

target (Port): a Port to connect to.

Returns:

None.

invoke_callbacks()

Invoke all callback functions with buffer value as argument

Args:

None.

Returns:

None.

register_callback(f)

Register a callback function to this Port

Args:

f (Function): a function to register

Returns:

None.

sync()

Sync self with the Connection.

Args:

None.

Returns:

None.

brica1.ros module

brica1.scheduler module

scheduler.py

This module contains the Scheduler class which is a base class for various types of schedulers. The VirtualTimeSyncScheduler is implemneted for now.

class brica1.scheduler.RealTimeSyncScheduler(agent, supervisor=<class 'brica1.supervisor.NullSupervisor'>, interval=1)

Bases: brica1.scheduler.Scheduler

RealTimeSyncScheduler is a Scheduler implementation for real time in a synced manner.

reset()

Reset the Scheduler.

Args:

None.

Returns:

None.

set_interval(interval)

step()

Step by the internal interval.

The methods input(), fire(), and output() are synchronously called for all components.

The time when it started calling input() and output() of the components is stored in self.last_input_time and self.last_output_time, respectively.

self.interval sets the minimum interval between the point in time when input() is called and when output() is called. The actual interval between input() and output() will always be longer than self.interval, although the scheduler tries to make the discrepancy minimum.

When calling fire() of the components takes longer than the set interval, calling output() of the components will be later than the scheduled time self.input_time + self.interval. In this case, self.lagged will be set True.

During the execution of this method, it will also set self.last_spent, which will be the time spent until all components are fired after self.last_input_time is set.

Args:

None.

Returns:

int: the current time of the Scheduler.

class brica1.scheduler.Scheduler(agent, supervisor=<class 'brica1.supervisor.NullSupervisor'>)

Bases: object

This class is an abstract class for creating Scheduler`s. Subclasses must override the `step() method to specify its implementation.

reset()

Reset the Scheduler.

Args:

None.

Returns:

None.

abstract step()

Step over a single iteration

Args:

None.

Returns:

int: the current time of the Scheduler.

update()

Update the Scheduler for given cognitive architecture (agent)

Args:

agent (Agent): a target to update.

Returns:

None.

class brica1.scheduler.VirtualTimeScheduler(agent, supervisor=<class 'brica1.supervisor.NullSupervisor'>)

Bases: brica1.scheduler.Scheduler

VirtualTimeScheduler is a Scheduler implementation for virutal time.

class Event(time, component, action='fire')

Bases: object

Event is a queue object for PriorityQueue in VirtualTimeScheduler.

step(interval=0)

Step with given interval or to the next event

An event is dequeued and output(), input(), and fire() are called for the Component of interest.

Args:

interval (int): interval in milliseconds between each step

Returns:

int: the current time of the Scheduler.

step_for_time(time)

update()

Update the Scheduler for given cognitive architecture (agent)

Args:

agent (Agent): a target to update.

Returns:

None.

class brica1.scheduler.VirtualTimeSyncScheduler(agent, supervisor=<class 'brica1.supervisor.NullSupervisor'>, interval=1)

Bases: brica1.scheduler.Scheduler

VirtualTimeSyncScheduler is a Scheduler implementation for virutal time in a synced manner.

step()

Step by the internal interval.

The methods input(), fire(), and output() are synchronously called and the time is incremented by the given interval for each step.

Args:

None.

Returns:

int: the current time of the Scheduler.

brica1.unit module

unit.py

This module contains the Unit modules which serves as a base class for the Component and Module classes.

class brica1.unit.Unit

Bases: object

Unit is a base class for `Module`s and `Component`s with functionalities for handling ports.

alias_in_port(target, from_id, to_id)

Alias an in-port from a target Unit to this Unit.

Technical note: ALWAYS alias ports OUTSIDE IN

Args:

target (Unit): a Unit to alias from. from_id (str): a port ID to alias from. to_id (str): a port ID to alias to.

Returns:

None.

alias_out_port(target, from_id, to_id)

Alias an out-port from a target Unit to this Unit.

Technical note: ALWAYS alias ports OUTSIDE IN

Args:

target (Unit): a Unit to alias from. from_id (str): a port ID to alias from. to_id (str): a port ID to alias to.

Returns:

None.

connect(target, from_id, to_id)

Connect an out-port of another Unit to an in-port.

Args:

target (Unit): a Unit to connect to. from_id (str): an out-port of the target Unit. to_id(str): an in-port of this Unit.

Returns:

None.

get_in_port(id)

Get values in an in-port from this Unit.

Args:

id (str): a string ID.

Returns:

numpy.ndarray: a value vector for the in-port ID.

get_out_port(id)

Get values in an out-port from this Unit.

Args:

id (str): a string ID.

Returns:

numpy.ndarray: a value vector for the in-port ID.

make_in_port(id, length)

Make an in-port of this Unit.

Args:

id (str): a string ID. length (int): an initial length of the value vector.

Returns:

None.

make_out_port(id, length)

Make an out-port of this Unit.

Args:

id (str): a string ID. length (int): an initial length of the value vector.

Returns:

None.

remove_in_port(id)

Remove an in-port from this Unit.

Args:

id (str): a string ID.

Returns:

None.

remove_out_port(id)

Remove an out-port from this Unit.

Args:

id (str): a string ID.

Returns:

None.

set_in_port(id, port)

Set a port with an actual instance.

Args:

id (str): a string ID. port (Port): a Port instance to set.

Returns:

None.

set_out_port(id, port)

Set a port with an actual instance.

Args:

id (str): a string ID. port (Port): a Port instance to set.

Returns:

None.

brica1.utils module

utils.py

This modules contains utility functions for BriCA.

brica1.utils.alias_in_port(from_tuple, to_tuple)

Alias in-ports of two units

Args:

from_tuple (tuple<Unit, str>): Unit and port id to alias from. to_tuple (tuple<Unit, str>): Unit and port id to alias to.

Returns:

None.

brica1.utils.alias_out_port(from_tuple, to_tuple)

Alias out-ports of two units

Args:

from_tuple (tuple<Unit, str>): Unit and port id to alias from. to_tuple (tuple<Unit, str>): Unit and port id to alias to.

Returns:

None.

brica1.utils.connect(from_tuple, to_tuple)

Connect ports of two units

Args:

from_tuple (tuple<Unit, str>): Unit and port id to connect from. to_tuple (tuple<Unit, str>): Unit and port id to connect to.

Returns:

None.

brica1.utils.current_time()

brica1.utils.current_time_millis()

Module contents

BriCA

Brain-inspired Cognitive Architecture

A scalable, general purpose computing platform for cognitive architectures.