CLI Tools and Utilities

def usage(argv0: str, out_file):
    """Print command-line program usage information."""

def show_actions(action: str = None):
    """
    Show available Manager API actions and their arguments.
    If action is specified, show detailed help for that action.
    """

def execute_action(manager: Manager, argv: list):
    """
    Execute specified Manager API action with given arguments.
    Supports both positional and named arguments (--name=value).
    """

def command_line(argv: list):
    """
    Main command-line interface entry point.
    Handles argument parsing and command dispatch.
    """

class ArgumentsError(BaseException):
    """Raised for bad command-line arguments."""

class AttributeDict(dict):
    """
    Dictionary subclass with special handling for multi-value fields.
    Used for Manager API responses and events.
    """
    
    MULTI_VALUE_FIELD: tuple = ('ChanVariable', 'DestChanVariable')
    """Fields that can have multiple occurrences in Manager responses."""
    
    def __setitem__(key: str, value: str):
        """Set item with special handling for multi-value fields."""

class EventCollection:
    """
    Collection for managing Asterisk event subscriptions and dispatching.
    Provides event filtering, subscription management, and callback handling.
    """
    
    def __init__(initial: dict = None):
        """Initialize with optional initial event handlers."""
    
    def subscribe(event_type: str, handler: callable):
        """Subscribe handler function to specific event type."""
    
    def unsubscribe(event_type: str, handler: callable):
        """Unsubscribe handler from event type."""
    
    def clear():
        """Remove all event subscriptions."""
    
    def fire(name: str, *args, **kwargs):
        """Fire event with given name and arguments to all subscribed handlers."""
    
    def copy() -> 'EventCollection':
        """Return a copy of this event collection."""
    
    def __iadd__(collection: 'EventCollection') -> 'EventCollection':
        """Add events from another collection to this one."""
    
    def __isub__(collection: 'EventCollection') -> 'EventCollection':
        """Remove events from another collection from this one."""

class Unspecified:
    """
    Represents an unspecified value that cannot be None.
    Used where None may be a valid argument value.
    """

Unspecified: Unspecified
"""Singleton instance of Unspecified class."""

class SubscriptionError(BaseException):
    """
    Raised when attempt to register the same (event, handler) tuple twice.
    """

def dump_packet(packet: dict, file = sys.stdout):
    """
    Dump packet data for debugging Manager API communication.
    
    Args:
        packet: Manager API packet/response dictionary
        file: Output file object (default stdout)
    """

def dump_human(data: dict, file = sys.stdout, _indent: int = 0):
    """
    Human-readable dump of structured data with indentation.
    
    Args:
        data: Data structure to dump
        file: Output file object (default stdout)  
        _indent: Internal indentation level
    """

class AsteriskLogger:
    """
    Enhanced logger class with additional logging levels for Asterisk debugging.
    """
    
    def state(msg: str, *args, **kwargs):
        """Log message with 'STATE' severity level."""
    
    def packet(msg: str, *args, **kwargs):
        """Log message with 'PACKET' severity level."""
    
    def io(msg: str, *args, **kwargs):
        """Log message with 'IO' severity level."""

class InstanceLogger:
    """
    Provides instance-specific logging functionality.
    """
    
    def getLogger() -> AsteriskLogger:
        """Get logger instance for the object."""

logging.STATE: int
"""Custom logging level (INFO - 1) for state changes."""

logging.PACKET: int  
"""Custom logging level (DEBUG - 1) for packet tracing."""

logging.IO: int
"""Custom logging level (PACKET - 1) for I/O operations."""

# asterisk-dump script
def main(argv: list):
    """Event dumping utility with auto-reconnect functionality."""

# asterisk-recover script  
def main(argv: list):
    """Interactive channel recovery utility for troubleshooting."""

# Show available actions
py-asterisk actions

# Execute specific action
py-asterisk action ShowChannels

# Execute action with arguments
py-asterisk action Originate SIP/1001 default 1002 1

# Execute action with named arguments
py-asterisk action Originate --channel=SIP/1001 --context=default --exten=1002

# Execute console command
py-asterisk command "core show channels"

# Show help for specific action
py-asterisk help Originate

# Dump all Manager API events to console
asterisk-dump --host=127.0.0.1 --port=5038 --username=admin --secret=password

# Dump events with auto-reconnect
asterisk-dump --host=127.0.0.1 --port=5038 --username=admin --secret=password --reconnect

# Interactive channel recovery interface
asterisk-recover --host=127.0.0.1 --port=5038 --username=admin --secret=password

from Asterisk.Manager import Manager
from Asterisk.Util import EventCollection

manager = Manager(('127.0.0.1', 5038), 'admin', 'secret', listen_events=True)

# Event handler function
def handle_hangup(event):
    print(f"Channel {event['Channel']} hung up")
    print(f"Cause: {event.get('Cause', 'Unknown')}")

def handle_newchannel(event):
    print(f"New channel: {event['Channel']}")
    print(f"State: {event['ChannelState']}")

# Subscribe to events
manager.events.subscribe('Hangup', handle_hangup)
manager.events.subscribe('Newchannel', handle_newchannel)

# Events will be dispatched automatically while manager is connected

from Asterisk.Util import AttributeDict, Unspecified

# AttributeDict handles multi-value fields specially
response = AttributeDict()
response['ChanVariable'] = 'var1=value1'  # Creates nested dict
response['ChanVariable'] = 'var2=value2'  # Adds to existing dict

# Unspecified for optional parameters
def my_function(required_param, optional_param=Unspecified):
    if optional_param is not Unspecified:
        # Parameter was provided
        process_optional(optional_param)

from Asterisk.Util import dump_packet, dump_human

# Debug Manager API packets  
response = manager.ShowChannels()
dump_packet(response)

# Human-readable data dump
dump_human({
    'channels': 5,
    'queues': {'support': {'calls': 2}, 'sales': {'calls': 0}}
})

import logging
from Asterisk.Logging import AsteriskLogger

# The enhanced logger is automatically installed system-wide
logger = logging.getLogger('my_app')

# Use custom logging levels
logger.state("Manager connected and authenticated")
logger.packet("Sent action: ShowChannels")  
logger.io("Reading response from socket")

# Standard levels still work
logger.info("Application started")
logger.error("Connection failed")