Channel Management

class BaseChannel:
    """
    Represents a living Asterisk channel with shortcut methods for operating on it.
    Acts as a mapping for getting/setting channel variables.
    """
    
    def __init__(manager: BaseManager, id: str):
        """
        Initialize channel object.
        
        Args:
            manager: Manager instance for API communication
            id: Channel identifier (e.g., 'SIP/1001-00000001')
        """
    
    def __eq__(other: BaseChannel) -> bool:
        """Check if channels are equal (same ID and manager)."""
    
    def __str__() -> str:
        """Return channel ID as string."""
    
    def __getitem__(key: str) -> str:
        """Get channel variable value (translates to Getvar action)."""
    
    def __setitem__(key: str, value: str):
        """Set channel variable (translates to Setvar action)."""

class ZapChannel(BaseChannel):
    """
    Specialized channel class for Zapata/DAHDI channels.
    Inherits all BaseChannel functionality with Zapata-specific features.
    """

def AbsoluteTimeout(timeout: int):
    """Set absolute timeout for this channel in seconds."""

def Hangup():
    """Hang up this channel."""

def Monitor(pathname: str = None, format: str = 'wav', mix: bool = False):
    """Start monitoring/recording this channel."""

def StopMonitor():
    """Stop monitoring this channel."""

def ChangeMonitor(pathname: str):
    """Change monitor filename for this channel."""

def get_channel(channel_id: str) -> BaseChannel:
    """
    Get channel object for the given channel ID.
    Returns ZapChannel for Zapata/DAHDI channels, BaseChannel otherwise.
    """

from Asterisk.Manager import Manager

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

# Get channel object
channel = manager.get_channel('SIP/1001-00000001') 

# Hang up the channel
channel.Hangup()

# Set timeout
channel.AbsoluteTimeout(300)  # 5 minutes

# Get channel variable
caller_id = channel['CALLERID(name)']
print(f"Caller ID: {caller_id}")

# Set channel variable
channel['MY_CUSTOM_VAR'] = 'some_value'

# Common channel variables
print(f"Channel state: {channel['CHANNEL(state)']}")
print(f"Call duration: {channel['CDR(duration)']}")
print(f"Source: {channel['CDR(src)']}")
print(f"Destination: {channel['CDR(dst)']}")

# Start monitoring call
channel.Monitor('/var/spool/asterisk/monitor/call_123', 'wav', mix=True)

# Change recording filename
channel.ChangeMonitor('/var/spool/asterisk/monitor/important_call_123')

# Stop monitoring
channel.StopMonitor()

# Get all active channels
channels_info = manager.ShowChannels()

# Process each channel
for channel_data in channels_info.get('channels', []):
    channel_id = channel_data['Channel']
    channel = manager.get_channel(channel_id)
    
    # Check if channel has been active too long
    duration = int(channel.get('CDR(billsec)', 0))
    if duration > 3600:  # 1 hour
        print(f"Long call detected: {channel_id} ({duration}s)")
        
        # Optionally hang up or take other action
        # channel.Hangup()

# For Zapata/DAHDI channels, you get a ZapChannel object
zap_channel = manager.get_channel('Zap/1-1')

# All BaseChannel methods work
zap_channel.Hangup()
zap_channel['CALLERID(name)'] = 'Updated Name'

# ZapChannel automatically returned for channels starting with 'Zap'
assert isinstance(zap_channel, ZapChannel)