fjagepy package

class fjagepy.AgentID(name, topic=False, owner=None)[source]

Bases: object

An identifier for an agent or a topic. This can be used to send, receive messages, and set or get parameters on an agent or topic on the fjåge container. The AgentID is typically owned by a Gateway which is used to send and receive messages.

The AgentID object can be used to set and get parameters on the agent using dot notation. For example, to get the value of a parameter named “param” on an agent represented by an AgentID object aid, you can use value = aid.param. To set the value of the parameter, you can use aid.param = value. If the parameter is indexed, you can use aid[index].param to refer to the indexed parameter. For example, aid[1].param refers to the first indexed parameter “param” of the agent.

Parameters:

  • name (str) – name of the agent

  • topic (bool) – True if this represents a topic. Defaults to False.

  • owner (Optional['Gateway']) – Gateway owner for this AgentID. Defaults to None.

get(index=-1)[source]

Gets the values of all parameters on the agent.

Parameters:

index (int | None) – index for indexed parameters. Defaults to -1 (no index).

Returns:

dictionary of all parameters and their values

Return type:

dict

Raises:

RuntimeError – if this AgentID has no owner (unowned AgentID cannot get parameters)

get_name()[source]

Gets the name of the agent or topic.

Returns:

name of agent or topic

Return type:

str

is_topic()[source]

Returns True if the agent id represents a topic.

Returns:

True if the agent id represents a topic, False if it represents an agent

Return type:

bool

request(msg, timeout=None)[source]

Sends a request to the agent represented by this id and waits for a response.

Parameters:

  • msg (Message) – request to send

  • timeout (int, optional) – timeout in milliseconds. Defaults to owner’s timeout.

Returns:

Response message

Raises:

RuntimeError – if this AgentID has no owner (unowned AgentID cannot send messages)

Return type:

Optional[Message]

send(msg)[source]

Sends a message to the agent represented by this id.

Parameters:

msg (Message) – message to send

Raises:

RuntimeError – if this AgentID has no owner (unowned AgentID cannot send messages)

Return type:

None

class fjagepy.Gateway(hostname='localhost', port=1100, connector=<class 'fjagepy.TCPConnector.TCPConnector'>, reconnect=True, timeout=1000, directory_timeout=6000)[source]

Bases: object

A Gateway is used to connect to a fjåge container and send and receive messages, query for agents and services, and set or get parameters on agents from Python.

The Gateway class provides methods for sending and receiving messages, subscribing to topics, and querying for agents and services. It uses a Connector to handle the underlying communication with the fjåge container.

The Gateway is thread-safe and can be used from multiple threads. It maintains an internal message queue with a configurable maximum size (default 512 messages). A gw.receive() call will first check the internal queue for a matching message before blocking for new messages.

The Gateway can be used as a context manager to ensure proper cleanup of resources.

Parameters:

  • hostname (str)

  • port (int)

  • connector (Type[Connector])

  • reconnect (bool)

  • timeout (int)

  • directory_timeout (int)

BLOCKING = -1
NON_BLOCKING = 0
agent(name)[source]

Creates an AgentID for the given agent name.

Return type:

AgentID

agentForService(service, timeout=None)[source]

Finds an agent that provides the given service.

Parameters:

  • service (str | Enum)

  • timeout (int | None)

Return type:

AgentID | None

agent_for_service(service, timeout=None)[source]

A python style method name for agentForService()

Parameters:

  • service (str | Enum)

  • timeout (int | None)

Return type:

AgentID | None

agent_id()[source]

Gets the AgentID of this gateway.

Returns:

AgentID of this gateway

Return type:

AgentID

agents(timeout=None)[source]

Gets the list of all agents connected to the fjage container

Parameters:

timeout (int | None)

Return type:

list[AgentID]

agentsForService(service, timeout=None)[source]

Retrieves a list of all agents that provide the given service.

Parameters:

  • service (str | Enum)

  • timeout (int | None)

Return type:

list[AgentID]

agents_for_service(service, timeout=None)[source]

A python style method name for agentsForService()

Parameters:

  • service (str | Enum)

  • timeout (int | None)

Return type:

list[AgentID]

close()[source]

Closes the gateway and disconnects from the fjage container.

containsAgent(agentID, timeout=None)[source]

Checks if the given agent is connected to the fjage container

Parameters:

  • agentID (AgentID)

  • timeout (int | None)

Return type:

bool

contains_agent(agentID, timeout=None)[source]

A python style method name for containsAgent()

Parameters:

  • agentID (AgentID)

  • timeout (int | None)

Return type:

bool

flush()[source]

Clears the internal message queue.

getAgentID()[source]

Java style method name for agent_id()

Deprecated since version 2.0.0: Use agent_id() instead.

Return type:

AgentID

isConnected()[source]

Java style method name for is_connected()

Deprecated since version 2.0.0: Use is_connected() instead.

Return type:

bool

is_connected()[source]

Returns True if the gateway is connected to the fjage container.

Returns:

True if connected, False otherwise

Return type:

bool

static match_filter(filter, msg)[source]
Parameters:
Return type:

bool

receive(msg_filter=None, timeout=0, **kwargs)[source]

Receives a message from the fjage container.

Parameters:

  • msg_filter (Callable | Type[Message] | Message | None) – a filter to apply to incoming messages

  • timeout (int) – timeout in milliseconds. Defaults to NON_BLOCKING (0 ms).

Returns:

received message or None if timeout

Return type:

Message or None

request(msg, timeout=None)[source]

Sends a request message and waits for a reply.

Parameters:

  • msg (Message) – request message to send

  • timeout (int | None) – timeout in milliseconds. Defaults to None, which uses the gateway’s default timeout.

Returns:

reply message or None if timeout

Return type:

Message or None

send(msg)[source]

Sends a message to the fjage container.

Parameters:

msg (Message) – message to send

Return type:

None

services(timeout=None)[source]

Gets the list of all services provided by agents connected to the fjage container

Parameters:

timeout (int | None)

Return type:

list[str]

subscribe(topic)[source]

Subscribes to the given topic.

Parameters:

topic (AgentID)

Return type:

bool

topic(topic, topic2=None)[source]

Creates a topic with the given name or based on the given AgentID.

Parameters:

  • topic (str | AgentID)

  • topic2 (str | None)

Return type:

AgentID

unsubscribe(topic)[source]

Unsubscribes from the given topic.

Parameters:

topic (AgentID)

Return type:

bool

class fjagepy.GetFileReq(filename=None, offset=0, length=0, **kwargs)[source]

Bases: Message

Parameters:

  • filename (str | None)

  • offset (int)

  • length (int)

filename: str | None
length: int
offset: int
class fjagepy.GetFileRsp(filename=None, offset=0, contents=None, directory=False, **kwargs)[source]

Bases: Message

Parameters:

  • filename (str | None)

  • offset (int)

  • contents (list[int] | None)

  • directory (bool)

class fjagepy.JSONMessage(json_str=None, owner=None)[source]

Bases: object

A JSONMessage is used to create and parse the on-the-wire JSON based protocol used by fjåge containers. It can be used to send and receive messages, query for agents and services, and set or get parameters on agents. The JSONMessage class provides static factory methods for creating common types of JSON messages and also can parse a JSON string into a JSONMessage object.

Parameters:

json_str (str | None)

static createAgentForService(service)[source]
Parameters:

service (str | Enum)

Return type:

JSONMessage

static createAgents()[source]
Return type:

JSONMessage

static createAgentsForService(service)[source]
Parameters:

service (str | Enum)

Return type:

JSONMessage

static createContainsAgent(agentID)[source]
Parameters:

agentID (AgentID)

Return type:

JSONMessage

static createSend(msg, relay=False)[source]
Parameters:
Return type:

JSONMessage

static createServices()[source]
Return type:

JSONMessage

static createWantsMessagesFor(agentIDs)[source]
Parameters:

agentIDs (List[AgentID])

Return type:

JSONMessage

to_json()[source]
Return type:

str

class fjagepy.Message(in_reply_to_msg=None, perf=Performative.INFORM, **kwargs)[source]

Bases: object

Base class for messages transmitted by one agent to another.

Parameters:
inReplyTo: str | None
msgID: str
perf: Performative
recipient: AgentID | None
sender: AgentID | None
sentAt: int | None
fjagepy.MessageClass(name, parent=<class 'fjagepy.Message.Message'>)[source]

Creates an unqualified message class based on a fully qualified name.

Parameters:

  • name (str) – fully qualified name of the message class

  • parent (type[Message]) – parent class to inherit from. Defaults to Message.

Returns:

A new subclass of Message with the given name.

Return type:

type[Message]

class fjagepy.ParameterReq(index=-1, **kwargs)[source]

Bases: Message

Parameters:

index (int)

get(param)[source]

Request a parameter by name.

Parameters:

param (str) – name of the parameter to request

index: int
param: str | None
perf: Performative
requests: list[dict]
set(param, value)[source]

Set a parameter value.

Parameters:

  • param (str) – name of the parameter to set

  • value – value to set the parameter to

value: Any | None
class fjagepy.ParameterRsp(**kwargs)[source]

Bases: Message

get(param)[source]

Get the value of a parameter by name from the response.

Parameters:

param (str) – name of the parameter to get

Returns:

value of the parameter, or None if not found

index: int
param: str | None
parameters()[source]

Get all parameters in the response as a dictionary.

Return type:

dict[str, Any]

perf: Performative
value: Any | None
values: dict | None
class fjagepy.Performative(*values)[source]

Bases: Enum

AGREE = 'AGREE'
CANCEL = 'CANCEL'
CFP = 'CFP'
CONFIRM = 'CONFIRM'
DISCONFIRM = 'DISCONFIRM'
FAILURE = 'FAILURE'
INFORM = 'INFORM'
NOT_UNDERSTOOD = 'NOT_UNDERSTOOD'
PROPOSE = 'PROPOSE'
QUERY_IF = 'QUERY_IF'
REFUSE = 'REFUSE'
REQUEST = 'REQUEST'
to_json()[source]
class fjagepy.PutFileReq(filename=None, contents=None, offset=0, **kwargs)[source]

Bases: Message

Parameters:

  • filename (str | None)

  • contents (list[int] | None)

  • offset (int)

contents: list[int] | None
filename: str | None
offset: int
class fjagepy.Services[source]

Bases: object

DOCUMENTATION = 'org.arl.fjage.shell.Services.DOCUMENTATION'
SHELL = 'org.arl.fjage.shell.Services.SHELL'
class fjagepy.ShellExecReq(command=None, script=None, scriptArgs=None, ans=False, **kwargs)[source]

Bases: Message

Parameters:

  • command (str | None)

  • script (str | None)

  • scriptArgs (list[str] | None)

  • ans (bool)

ans: bool
command: str | None
script: str | None
scriptArgs: list[str] | None
class fjagepy.TCPConnector(**kwargs)[source]

Bases: Connector

Simple TCP connector using synchronous sockets.

Parameters:

kwargs (Any)

connect()[source]

Establish the connection.

Return type:

None

disconnect()[source]

Close the connection.

Return type:

None

is_connected()[source]

Check if connected.

Return type:

bool

send(msg)[source]

Send a message.

Parameters:

msg (str)

Return type:

None

set_receive_callback(callback)[source]

Set the callback function to handle incoming messages.

Parameters:

callback (Callable[[List[str]], None])

Return type:

None

fjagepy.message(arg)[source]

Decorator to register a Message subclass for JSON inflation.

Can be used as @message or @message('org.example.MyMessage').

Parameters:

arg (type[Message] | str)

Return type:

type[Message] | Callable[[…], type[Message]]