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[ソース]

ベースクラス: 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(id)[ソース]

Clear a state value for the given ID.

Args:
id (str): a string ID.
Returns:
None.
clear_state(id)[ソース]

Clear a state value for the given ID.

Args:
id (str): a string ID.
Returns:
None.
fire()[ソース]

Perform a calculation.

Users must override fire(self) method to define a new sub-class of Module. 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(id)[ソース]

Get a result value for the given ID.

Args:
id (str): a string ID.
Returns:
any: a result value for the given ID.
get_state(id)[ソース]

Get a state value for the given ID.

Args:
id (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 (float): 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 (float): 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(id, value)[ソース]

Set a result value for the given ID.

Args:
id (str): a string ID. v (any): a result value to set.
Returns:
None.
set_state(id, value)[ソース]

Set a state value for the given ID.

Args:
id (str): a string ID. v (any): a state value to set.
Returns:
None.
class brica1.component.ComponentSet[ソース]

ベースクラス: brica1.component.Component

ComponentSet groups components to fire sequentially

add_component(id, component, priority)[ソース]

Add a Component for given ID and priority

Args:
id (str): a string ID. component (Component): a component to add for id. priority (int): a priority value for scheduling.
Returns:
None.
fire()[ソース]

Fire sub-components based on priority

Args:
None.
Returns:
None.
class brica1.component.ConstantComponent[ソース]

ベースクラス: 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.PipeComponent[ソース]

ベースクラス: brica1.component.Component

PipeComponent copies contents of in ports to out ports.

fire()[ソース]
set_map(in_id, out_id)[ソース]

Map from in-port to out port.

Args:
in_id (str): port id to map from. out_id (str): port id to map to.
Returns:
None.
class brica1.component.NullComponent[ソース]

ベースクラス: brica1.component.Component

NullComponent does nothing.

fire()[ソース]

Do nothing. Just return.

Args:
None.
Returns:
None.

brica1.connection module

connection.py

This module contains the class Connection.

class brica1.connection.Connection(from_port, to_port)[ソース]

ベースクラス: 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.Module[ソース]

ベースクラス: 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.
class brica1.module.Agent(scheduler)[ソース]

ベースクラス: brica1.module.Module

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

add_submodule(id, submodule)[ソース]

Add a Module and update the Scheduler.

Args:
id (str): a string ID. submodule (Module): a module to add for id.
Returns:
None.
step()[ソース]

Step the Scheduler.

Args:
None.
Returns:
float: the current time of the scheduler
update_scheduler()[ソース]

Udpate the Scheduler with this Agent.

Args:
None.
Returns:
None.

brica1.port module

port.py

This module contains the class Port.

class brica1.port.Port(value)[ソース]

ベースクラス: 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

ros.py

This module containes classes for ROS integration.

class brica1.ros.ROSAdapter(name='BriCA1 Node')[ソース]

ベースクラス: brica1.unit.Unit

ROSAdapter is a BriCA Unit which is intented to provide a bridge over the ROS Publisher/Subscriber and BriCA Agent.

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.
setup_publisher(topic, id, length)[ソース]

Setup a ROS subscriber

Args:
topic (str): a topic to subscribe to. id (str): a string ID. length (int): an initial length of the value vector.
Returns:
None.
setup_subscriber(topic, msg_type, id, length, converter)[ソース]

Setup a ROS subscriber

Args:
topic (str): a topic to subscribe to. msg_type (msg): incoming message type. id (str): a string ID. length (int): an initial length of the value vector.
Returns:
None.

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.Scheduler[ソース]

ベースクラス: 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.
step()[ソース]

Step over a single iteration

Args:
None.
Returns:
float: the current time of the Scheduler.
update(ca)[ソース]

Update the Scheduler for given cognitive architecture (ca)

Args:
ca (Agent): a target to update.
Returns:
None.
class brica1.scheduler.VirtualTimeSyncScheduler(interval=1.0)[ソース]

ベースクラス: 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:
float: the current time of the Scheduler.
class brica1.scheduler.VirtualTimeScheduler[ソース]

ベースクラス: brica1.scheduler.Scheduler

VirtualTimeScheduler is a Scheduler implementation for virutal time.

class Event(time, component)[ソース]

ベースクラス: object

Event is a queue object for PriorityQueue in VirtualTimeScheduler.

VirtualTimeScheduler.step()[ソース]

Step by the internal interval.

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

Args:
None.
Returns:
float: the current time of the Scheduler.
VirtualTimeScheduler.update(ca)[ソース]

Update the Scheduler for given cognitive architecture (ca)

Args:
ca (Agent): a target to update.
Returns:
None.
class brica1.scheduler.RealTimeSyncScheduler(interval=1.0)[ソース]

ベースクラス: brica1.scheduler.Scheduler

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

reset()[ソース]
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:
float: 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[ソース]

ベースクラス: 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.current_time()[ソース]
brica1.utils.current_time_millis()[ソース]
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.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.

Module contents

BriCA

Brain-inspired Cognitive Architecture

A scalable, general purpose computing platform for cognitive architectures.