[FreePBX] Python으로 AMI 프로그래밍 테스트 2 (이벤트 수신 및 처리 - 비동기 방식)

 

 

안녕하세요.

 

 

 지난 글 "Python으로 FreePBX AMI 프로그래밍 테스트 1 (연결, Ping, 종료흐름)"에서는 Python의 socket 모듈을 사용하여 FreePBX AMI에 접속하고, 간단한 명령(Login, Ping, Logoff)을 실행하는 동기적인 기본 흐름을 살펴보았습니다.

 

 하지만 전화 시스템의 상태 변화(예: 전화 수신, 통화 종료, 채널 상태 변경 등)를 실시간으로 감지하고 이에 맞게 처리하려면 조금 다른 접근 방식이 필요합니다.

 바로 AMI가 지속적으로 보내주는 이벤트(Event)를 수신하고 처리하는 것입니다.

 

 이번 글에서는 Python의 asyncio라이브러리를 활용하여 FreePBX AMI로부터 이벤트를 비동기적으로 수신하고, 특정 이벤트를 감지하여 처리하는 방법을 알아보려고 합니다.

 

---

 

 

1. 이전 방식의 한계와 비동기 프로그래밍의 필요성

 

 이전 글의 동기 방식 코드는 read_response() 함수가 서버로부터 응답이 올 때까지 프로그램의 실행을 멈추고 기다립니다(Blocking).

 

 만약 우리가 전화기의 상태 변화 같은 실시간 이벤트를 계속 듣고 싶다면, 동기 방식으로는 하나의 작업이 끝날 때까지 다른 작업을 할 수 없다는 문제가 있습니다.

 

 예를 들어, 로그인 후 계속해서 "새로운 전화가 왔는지?", "통화가 연결되었는지?" 등의 이벤트를 기다려야 하는데, 동기 방식의 read_response()는 하나의 응답만 처리하고 끝나버립니다.

 

 asyncio는 I/O 바운드(I/O-bound) 작업 (네트워크 통신, 파일 읽기/쓰기 등 응답을 기다리는 시간이 많은 작업)을 효율적으로 처리하기 위한 Python 라이브러리입니다.

 

 async와 await 키워드를 사용해서 코드가 특정 작업의 완료를 기다리는 동안에도 다른 작업을 수행할 수 있게 해 줍니다(Non-blocking).

 

 이를 통해 AMI로부터 오는 이벤트를 지속적으로 감지하면서도 프로그램이 멈추지 않도록 할 수 있습니다.

 

 

2. 준비사항

 

  - PC에 Python 3.7 이상 설치 권장 (asyncio는 Python 3.4 이상 버전에서 기본으로 제공）

  - FreePBX 서버 정보 (IP 주소, AMI 포트 - 보통 5038, AMI 사용자 이름, 비밀번호)

 

 

3. AMI 이벤트 수신 위한 Python 소스코드

 

 아래 코드는 asyncio를 사용하여 FreePBX AMI에 연결하고, 로그인한 후 지속적으로 이벤트를 수신합니다.

 테스트로 "200번 내선이 300번 내선으로 전화를 걸어 벨이 울리는 상황(DialState 이벤트)"을 감지하여 콘솔에 로그를 출력하는 간단한 로직으로 만들어 봤습니다.

 

 AMI_HOST, AMI_USERNAME, AMI_SECRET 등은 FreePBX에서 생성한 계정에 맞게 수정해야 합니다.

 

import socket 
import asyncio

# --- FreePBX AMI 설정 ---
AMI_HOST = '192.168.219.110'  # FreePBX 서버 IP (실제 환경에 맞게 수정)
AMI_PORT = 5038              # AMI 포트 (기본값)
AMI_USERNAME = 'myuser'      # AMI 사용자 이름 (실제 환경에 맞게 수정)
AMI_SECRET = 'myuser'        # AMI 비밀번호 (실제 환경에 맞게 수정)
# --- 설정 끝 ---

async def receive_events():
    """
    AMI에 연결하여 로그인하고, 지속적으로 이벤트를 수신하여 처리합니다.
    """
    reader, writer = None, None # finally 블록에서 사용하기 위해 초기화
    try:
        # 비동기적으로 AMI 서버에 연결
        reader, writer = await asyncio.open_connection(AMI_HOST, AMI_PORT)
        print(f"Connected to AMI: {AMI_HOST}:{AMI_PORT}")

        # 로그인 액션 전송
        
        login_action = f"Action: Login\r\nUsername: {AMI_USERNAME}\r\nSecret: {AMI_SECRET}\r\n\r\n"
        writer.write(login_action.encode('utf-8'))
        await writer.drain() # 버퍼의 모든 데이터가 전송될 때까지 대기

        # 로그인 응답 수신 및 확인
        login_response = await reader.readuntil(b'\r\n\r\n') # 메시지 구분자까지 읽음
        login_response_str = login_response.decode('utf-8', errors='ignore').strip()
        print(f"Login Response:\n{login_response_str}")

        if 'Response: Success' not in login_response_str:
            print("AMI Login failed.")
            return

        print("AMI Login successful. Listening for events...")

        # 이벤트 수신 루프
        while True:
            # AMI 메시지는 '\r\n\r\n'으로 끝나므로, 이를 기준으로 메시지를 읽습니다.
            event_data = await reader.readuntil(b'\r\n\r\n')
            if not event_data: # 연결이 끊긴 경우
                print("Connection closed by AMI.")
                break
            
            event_str = event_data.decode('utf-8', errors='ignore').strip()
            if event_str: # 빈 메시지가 아닐 경우에만 처리
                print(f"--- Received Event ---\n{event_str}\n----------------------")

                # 특정 이벤트 감지: 200번이 300번으로 전화 걸 때 (벨 울림 상태)
                if 'Event: DialState' in event_str and \
                   'Channel: PJSIP/200-' in event_str and \
                   'DestChannel: PJSIP/300-' in event_str and \
                   'DialStatus: RINGING' in event_str:
                    print("\n>>> [LOG] 200번이 300번으로 전화를 겁니다 (벨 울리는 중).\n")
                    # 여기에 원하는 액션 추가 (예: 데이터베이스 기록, 알림 전송 등)

    except ConnectionRefusedError:
        print(f"Connection refused to {AMI_HOST}:{AMI_PORT}. Check AMI server and firewall.")
    except asyncio.TimeoutError:
        print("Timeout occurred while trying to connect or read from AMI.")
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        if writer:
            print("Closing AMI connection...")
            # 로그아웃 액션 (선택 사항, 연결이 끊기면 자동으로 로그아웃됨)
            # logoff_action = "Action: Logoff\r\n\r\n"
            # writer.write(logoff_action.encode('utf-8'))
            # await writer.drain()
            writer.close()
            await writer.wait_closed() # 소켓이 완전히 닫힐 때까지 대기
        print("AMI Connection closed.")

async def main():
    await receive_events()

if __name__ == "__main__":
    asyncio.run(main())

 

 

4. 소스코드 흐름 분석

 

(1)  receive_events() 비동기 함수

 asyncio.open_connection(AMI_HOST, AMI_PORT) 함수를 이용해서 FreePBX AMI 서버에 비동기적으로 TCP 연결을 시도합니다. 성공하면 데이터를 읽기 위한 reader와 쓰기를 위한 writer 객체를 반환합니다.

 await 키워드는 이 연결 작업이 완료될 때까지 기다리되, 다른 asyncio 작업을 허용합니다.

 

(2) 로그인

  writer.write(login_action.encode('utf-8')): 명령을 바이트로 인코딩하여 서버에 전송합니다.

  await writer.drain(): 전송 버퍼의 내용이 모두 소켓으로 전달될 때까지 대기합니다.

  await reader.readuntil(b'\r\n\r\n'): 서버로부터 응답을 읽습니다. AMI 메시지는 보통 빈 줄 (`\r\n\r\n`)로 끝나므로, 이 구분자가 나올 때까지 데이터를 읽습니다.

 

(3) 이벤트 수신 루프 (while True)

  await reader.readuntil(b'\r\n\r\n'): AMI 서버로부터 다음 메시지(이벤트 또는 응답)가 올 때까지 비동기적으로 대기하며 읽습니다.

 event_data.decode('utf-8', errors='ignore').strip(): 수신된 바이트 데이터를 문자열로 디코딩하고, 앞뒤 공백을 제거합니다.

 

  (4) 특정 이벤트 감지

 코드에서는 Event: DialState(다이얼 상태 변경 이벤트)이면서, Channel이 `PJSIP/200-`으로 시작하고, DestChannel이 PJSIP/300-으로 시작하며, DialStatus가 RINGING인 경우를 감지합니다.

 

이는 "200번 내선에서 300번 내선으로 전화를 걸어 상대방에게 벨이 울리고 있는 상황"을 나타냅니다.

 

 (5) main() 비동기 함수 및 실행

   asyncio.run(main()): 스크립트가 직접 실행될 때 asyncio 이벤트 루프를 시작하고 main() 함수 코루틴을 실행합니다.

 

 

5. 실행 및 테스트 결과

 

 PC에서 MicroSIP(300) 프로그램을 실행했고, 스마트폰에서는 MizuDroid(200) 앱을 실행해서 테스트했습니다.

 

 (1) 위 코드를 실행하면 AMI에 연결하고 로그인한 후 "AMI Login successful. Listening for events..." 메시지가 나옵니다. 그리고 받은 이벤트도 출력합니다.

 

 

 

 (2) 이제 FreePBX에 등록된 200번 스마트폰 앱을 이용하여 300번 PC 전화기로 전화를 겁니다.

 

 (3) 300번 전화기에서 벨이 울이며, 콘솔에 이벤트가 출력되면서, 지정한 로그 메시지도 함께 나타나는 것을 확인할 수 있습니다.

 

 

 

이것으로 AMI를 이용한 이벤트(Event)를 수신하고 처리하는 방법과 동작을 확인 했습니다.

 

 

6. 추가적으로 보완해 볼 수 있는 내용

 

  (1) 현재는 문자열 포함 여부로 단순하게 확인하지만, 실제로는 각 이벤트의 필드를 정확히 파싱 하여 (예: 줄 단위로 나누고, ':'를 기준으로 키-값 분리) 정규 표현식이나 간단한 파서를 만들어 보완해 볼 수 있을 것입니다.

 

  (2) DialState 외에도 Newchannel, Hangup, QueueMemberStatus 등 다양한 AMI 이벤트를 활용하여 전화 시스템의 상태를 모니터링하고 원하는 동작을 수행할 수 있을 것입니다.

 

  (3) 네트워크 문제 등으로 AMI 연결이 끊겼을 때 자동으로 재연결을 시도하는 로직을 추가하면 더욱 안정적인 시스템을 만들 수 있을 것입니다.

 

  (4) asyncio를 사용해서 테스트했지만, aiopyami (asyncio 기반) 또는 PAMI (동기 방식) 같은 Python AMI 라이브러리를 사용해 구현해 볼 수도 있을 것입니다.

 

 

참고) 위의 소스는 완벽한 소스코드가 아니기 때문에 보완해야 할 부분이 많이 있습니다. 참고 정도로 보셨으면 합니다.

 

 

감사합니다.

 

 

 

<참고사이트>

1. [FreePBX] Python으로 FreePBX AMI 프로그래밍 테스트 1 (연결, Ping, 종료 흐름)

https://remnant24c1.tistory.com/538

2. ettoreleandrotognoli/python-ami

https://github.com/ettoreleandrotognoli/python-ami

3. AMI Libraries and Frameworks

https://docs.asterisk.org/Configuration/Interfaces/Asterisk-Manager-Interface-AMI/AMI-Libraries-and-Frameworks/