Protocols#

Coordination protocols for vertical and horizontal agent coordination.

Base Classes#

class heron.protocols.base.Protocol#

Base class for coordination protocols.

abstractmethod execute(agents: list, **kwargs) → dict#: Execute the protocol among agents.

class heron.protocols.base.CommunicationProtocol#

Protocol with message-based communication support.

async execute_distributed(agents: list, broker: MessageBroker) → dict#: Execute protocol with message passing.

Vertical Protocols#

Protocols for top-down coordination between hierarchy levels.

SetpointProtocol#

class heron.protocols.vertical.SetpointProtocol#

Direct setpoint control from coordinator to subordinates.

execute(coordinator, setpoints: dict) → dict#

Distribute setpoints to subordinate agents.

Parameters:

coordinator – Coordinator agent
setpoints – Dict mapping subordinate IDs to setpoint values

Usage:

from heron.protocols.vertical import SetpointProtocol

protocol = SetpointProtocol()

setpoints = {
    "device_1": {"p": 1.0, "q": 0.5},
    "device_2": {"p": 0.8, "q": 0.3},
}
result = protocol.execute(coordinator, setpoints)

PriceSignalProtocol#

class heron.protocols.vertical.PriceSignalProtocol(initial_price: float = 50.0)#

Market-based coordination via price signals.

Parameters:: initial_price – Starting price value

price: float#: Current price signal.

set_price(price: float)#: Update the price signal.

execute(coordinator, price: float = None) → dict#: Broadcast price to all subordinates.

Usage:

from heron.protocols.vertical import PriceSignalProtocol

protocol = PriceSignalProtocol(initial_price=50.0)

# System operator sets price based on conditions
protocol.set_price(65.0)  # High demand period
result = protocol.execute(coordinator)

Horizontal Protocols#

Protocols for peer-to-peer coordination between same-level agents.

P2PTradingProtocol#

class heron.protocols.horizontal.P2PTradingProtocol#

Peer-to-peer energy trading protocol.

execute(agents: list) → dict#

Execute trading round among agents.

Returns:: Dict with trades and settlement

Usage:

from heron.protocols.horizontal import P2PTradingProtocol

protocol = P2PTradingProtocol()
result = protocol.execute([agent1, agent2, agent3])
# result = {"trades": [...], "total_volume": 1.5}

ConsensusProtocol#

class heron.protocols.horizontal.ConsensusProtocol(max_iterations: int = 10, tolerance: float = 0.01)#

Distributed consensus algorithm for agreement.

Parameters:

max_iterations – Maximum consensus rounds
tolerance – Convergence tolerance

execute(agents: list) → dict#

Run consensus until convergence.

Returns:: Dict with consensus value and convergence info

Usage:

from heron.protocols.horizontal import ConsensusProtocol

protocol = ConsensusProtocol(max_iterations=20, tolerance=0.001)
result = protocol.execute(agents)
# result = {"consensus_value": 42.5, "iterations": 8, "converged": True}

Protocol Selection Guide#

Protocol	Direction	Use Case
SetpointProtocol	Vertical	Direct control of subordinate devices
PriceSignalProtocol	Vertical	Market-based incentive coordination
P2PTradingProtocol	Horizontal	Energy/resource trading between peers
ConsensusProtocol	Horizontal	Agreement on shared values/decisions