### Start Recording Audio Stream to WAV File Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Begin recording the audio from the stream to a specified WAV file. Ensure the audio stream object is initialized. ```python def start_recording(self, filename): """Start recording the call to a WAV file""" if self.audio_stream: self.audio_stream.start_recording(filename) ``` -------------------------------- ### Handle Media Stream Start Notification Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Callback for when the media stream starts. Logs details about the stream, including codec, sample rate, and RTP addresses. ```python def _NH_MediaStreamDidStart(self, notification): stream = notification.sender print(f"Audio stream started") print(f" Codec: {stream.codec}") print(f" Sample rate: {stream.sample_rate}") print(f" Local RTP: {stream.local_rtp_address}:{stream.local_rtp_port}") print(f" Remote RTP: {stream.remote_rtp_address}:{stream.remote_rtp_port}") ``` -------------------------------- ### Implement Custom SIP Storage Backend Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Implement the ISIPSimpleStorage interface to create a custom backend for storing SIP-related data. This example shows initialization with a connection string and factory for XCAP storage. ```python from zope.interface import implementer from sipsimple.storage import ISIPSimpleStorage @implementer(ISIPSimpleStorage) class CustomStorage: def __init__(self, connection_string): # Initialize your custom backend (e.g., database) self.configuration_backend = DatabaseBackend(connection_string) self.xcap_storage_factory = lambda account: DatabaseXCAPStorage(connection_string, account) ``` -------------------------------- ### Working with FileSelector Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Demonstrates creating and parsing FileSelector objects for managing file metadata in MSRP file transfers. Includes examples for creating from a file path, parsing from an SDP attribute, and generating an SDP representation. ```python from sipsimple.streams.msrp.filetransfer import FileSelector # Create from file path selector = FileSelector.for_file('/path/to/document.pdf') print(f"Name: {selector.name}") print(f"Size: {selector.size}") print(f"Type: {selector.type}") # Auto-detected MIME type # Parse from SDP attribute sdp_attr = 'name:"photo.jpg" type:image/jpeg size:102400 hash:sha-1:72:24:5F:E8:...' selector = FileSelector.parse(sdp_attr) # Get SDP representation sdp_repr = selector.sdp_repr # For including in SDP offer/answer ``` -------------------------------- ### Handle Audio Stream Recording Start Notification Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Callback for when the AudioStream begins recording. Logs the filename. ```python def _NH_AudioStreamDidStartRecording(self, notification): print(f"Recording started: {notification.data.filename}") ``` -------------------------------- ### Configure AMR-WB Codec on Linux Source: https://github.com/agprojects/python3-sipsimple/blob/master/docs/Dependencies.txt Configure and install the wide-band AMR codec on Linux systems. The installation prefix defaults to '/usr/local/'. ```bash ./configure --prefix=/usr/local/ make sudo make install ``` -------------------------------- ### Initiate a Video Call Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Start a video call by creating a Session and adding both AudioStream and VideoStream. Observers are added to handle session and video stream notifications. The call is connected using SIP URI and routes. ```python from sipsimple.streams.rtp.video import VideoStream from sipsimple.video import VideoDevice from sipsimple.session import Session from application.notification import NotificationCenter, IObserver from zope.interface import implementer @implementer(IObserver) class VideoCallHandler: def __init__(self, account): self.account = account self.session = None self.video_stream = None self.notification_center = NotificationCenter() def start_video_call(self, target_uri, routes): """Start a video call""" from sipsimple.streams.rtp.audio import AudioStream self.session = Session(self.account) audio_stream = AudioStream() video_stream = VideoStream() self.notification_center.add_observer(self, sender=self.session) self.notification_center.add_observer(self, sender=video_stream) target = SIPURI.parse(f'sip:{target_uri}') self.session.connect(ToHeader(target), routes, [audio_stream, video_stream]) ``` -------------------------------- ### Configure AMR-WB Codec on Mac Source: https://github.com/agprojects/python3-sipsimple/blob/master/docs/Dependencies.txt Configure and install the wide-band AMR codec on macOS. Ensure to set the SIPSIMPLE_AMR_WB_PATH environment variable after installation. ```bash ./configure --prefix=/usr/local/opt/vo-amrwbenc make sudo make install ``` -------------------------------- ### Install Python Dependencies on Debian/Ubuntu Source: https://github.com/agprojects/python3-sipsimple/blob/master/docs/Dependencies.txt Use this command to install Python 3 dependencies from AG Projects repositories on Debian or Ubuntu based systems. This command reads dependency information from the debian/control file. ```bash sudo mk-build-deps --install debian/control ``` -------------------------------- ### Fetch PJSIP Dependencies Source: https://github.com/agprojects/python3-sipsimple/blob/master/docs/Dependencies.txt Use this script to fetch the PJSIP base revision. Specify '2.11' as an argument to get version 2.11, otherwise it defaults to 2.10. ```bash ./get_dependencies.sh ``` ```bash ./get_dependencies.sh 2.11 ``` -------------------------------- ### Play WAV Audio File into Call Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Inject audio from a WAV file into the active call. Requires the SIPApplication to be configured with a voice audio mixer. The player is added to the stream's bridge and then started. ```python def play_audio_file(self, filename): """Play a WAV file into the call""" from sipsimple.application import SIPApplication player = WavePlayer( SIPApplication.voice_audio_mixer, filename, loop_count=1, # Play once initial_delay=0 ) self.audio_stream.bridge.add(player) player.play() ``` -------------------------------- ### Access and Modify SIP Configuration Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Manage SIP Simple application settings by accessing configuration data. Use get_names to retrieve lists of configured items, get to fetch specific values, and set to modify them, followed by save() to persist changes. ```python config_manager = ConfigurationManager() # Get raw configuration data account_names = config_manager.get_names(['Accounts']) print(f"Configured accounts: {account_names}") # Get specific setting value = config_manager.get('SIPSimpleSettings.audio.sample_rate') print(f"Sample rate: {value}") # Set specific setting config_manager.set('SIPSimpleSettings.audio.sample_rate', 48000) config_manager.save() ``` -------------------------------- ### Handle New Incoming SIP Session Notification Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Processes notifications for new incoming SIP sessions, printing caller information and proposed streams. Includes a commented-out example for auto-answering. ```python def _NH_SIPSessionNewIncoming(self, notification): session = notification.sender print(f"Incoming call from {session.remote_identity}") print(f"Proposed streams: {notification.data.streams}") # Auto-answer example: # self.answer_call(session) ``` -------------------------------- ### Handle Video Stream Notifications Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Process notifications related to the video stream lifecycle and events. This includes handling stream start, remote format changes, keyframe events, and camera changes. ```python def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, None) if handler: handler(notification) def _NH_MediaStreamDidStart(self, notification): stream = notification.sender if isinstance(stream, VideoStream): self.video_stream = stream print("Video stream started") print(f" Codec: {stream.codec}") print(f" Remote format: {stream.producer.width}x{stream.producer.height}") def _NH_VideoStreamRemoteFormatDidChange(self, notification): width = notification.data.width height = notification.data.height print(f"Remote video format changed: {width}x{height}") def _NH_VideoStreamReceivedKeyFrame(self, notification): print("Received video keyframe") def _NH_VideoStreamMissedKeyFrame(self, notification): # SDK automatically requests keyframe print("Missed video keyframe, requesting new one") def _NH_VideoDeviceDidChangeCamera(self, notification): old_camera = notification.data.old_camera new_camera = notification.data.new_camera print(f"Camera changed from {old_camera} to {new_camera}") ``` -------------------------------- ### Initialize and Run SIPApplication Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Use SIPApplication to manage the SDK lifecycle. Initialize with file-based storage and subscribe to lifecycle notifications. ```python from sipsimple.application import SIPApplication from sipsimple.storage import FileStorage from application.notification import NotificationCenter, IObserver from zope.interface import implementer @implementer(IObserver) class MyApp: def __init__(self): self.application = SIPApplication() self.notification_center = NotificationCenter() # Subscribe to application notifications self.notification_center.add_observer(self, name='SIPApplicationDidStart') self.notification_center.add_observer(self, name='SIPApplicationDidEnd') self.notification_center.add_observer(self, name='SIPApplicationFailedToStartTLS') def start(self): # Initialize with file-based storage for configuration storage = FileStorage('~/.sipclient') self.application.start(storage) def stop(self): self.application.stop() def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, None) if handler: handler(notification) def _NH_SIPApplicationDidStart(self, notification): print("SIP Application started successfully") print(f"Voice Audio Mixer: {self.application.voice_audio_mixer}") print(f"Alert Audio Mixer: {self.application.alert_audio_mixer}") def _NH_SIPApplicationDidEnd(self, notification): print("SIP Application stopped") # Usage app = MyApp() app.start() # ... application runs ... app.stop() ``` -------------------------------- ### Initialize and Manage ChatStream Source: https://context7.com/agprojects/python3-sipsimple/llms.txt This snippet demonstrates initializing a ChatStream, setting up a session, and subscribing to notifications. It requires importing necessary classes from sipsimple. ```python from sipsimple.streams.msrp.chat import ChatStream, ChatIdentity, CPIMPayload from sipsimple.session import Session from sipsimple.core import SIPURI, ToHeader from application.notification import NotificationCenter, IObserver from zope.interface import implementer @implementer(IObserver) class ChatHandler: def __init__(self, account): self.account = account self.session = None self.chat_stream = None self.notification_center = NotificationCenter() def start_chat_session(self, target_uri, routes): """Start a new chat session""" self.session = Session(self.account) self.chat_stream = ChatStream() # Subscribe to notifications self.notification_center.add_observer(self, sender=self.session) self.notification_center.add_observer(self, sender=self.chat_stream) target = SIPURI.parse(f'sip:{target_uri}') self.session.connect(ToHeader(target), routes, [self.chat_stream]) ``` -------------------------------- ### Configure Video Settings in Python Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Use this snippet to configure video resolution, framerate, and H.264 profile using SIPSimpleSettings. Ensure SIPSimpleSettings and VideoResolution are imported. ```python from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.configuration.datatypes import VideoResolution settings = SIPSimpleSettings() settings.video.resolution = VideoResolution('1280x720') settings.video.framerate = 30 settings.video.h264.profile = 'baseline' settings.save() ``` -------------------------------- ### Manage SIP Accounts with AccountManager Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Demonstrates using AccountManager to load, manage, and find SIP accounts. It shows how to set a default account and find an account by its contact URI. ```python # Using AccountManager to manage multiple accounts manager = AccountManager() manager.load() # Load accounts from configuration # Set default account for outgoing calls accounts = manager.get_accounts() if accounts: manager.default_account = accounts[0] # Find account by contact URI account = manager.find_account(contact_uri) ``` -------------------------------- ### Configure and Enable a SIP Account Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Sets up a new SIP account with authentication, SIP, RTP, NAT traversal, and presence configurations. The account is then enabled and saved. Subscribe to account notifications for status updates. ```python from sipsimple.account import Account, AccountManager from sipsimple.configuration.datatypes import SIPAddress, SIPProxyAddress from application.notification import NotificationCenter, IObserver from zope.interface import implementer @implementer(IObserver) class AccountHandler: def __init__(self): self.notification_center = NotificationCenter() def setup_account(self): # Create a new SIP account account = Account('alice@example.com') # Configure authentication account.auth.username = 'alice' account.auth.password = 'secret123' # Configure SIP settings account.sip.outbound_proxy = SIPProxyAddress('proxy.example.com', 5060, 'udp') account.sip.register = True account.sip.register_interval = 600 # Configure RTP/audio settings account.rtp.audio_codec_list = ['opus', 'G722', 'PCMU', 'PCMA'] account.rtp.encryption.enabled = True account.rtp.encryption.key_negotiation = 'opportunistic' # Configure NAT traversal account.nat_traversal.use_ice = True account.nat_traversal.stun_server_list = ['stun.example.com:3478'] # Configure presence account.presence.enabled = True # Enable and save the account account.enabled = True account.save() # Subscribe to account notifications self.notification_center.add_observer(self, sender=account) return account def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, None) if handler: handler(notification) def _NH_SIPAccountDidActivate(self, notification): account = notification.sender print(f"Account {account.id} activated") def _NH_SIPAccountRegistrationDidSucceed(self, notification): account = notification.sender print(f"Account {account.id} registered successfully") print(f"Contact: {notification.data.contact_header}") print(f"Expires: {notification.data.expires}") def _NH_SIPAccountRegistrationDidFail(self, notification): account = notification.sender print(f"Account {account.id} registration failed: {notification.data.error}") def _NH_SIPAccountGotMessageSummary(self, notification): # Voicemail message waiting indication summary = notification.data.message_summary print(f"Voicemail: {summary.summaries.get('voice-message', {})}") ``` -------------------------------- ### Initialize and Manage SIP Call Session Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Sets up a new SIP session, subscribes to notifications, and prepares media streams (audio, optional video). Requires an account object and target URI. ```python from sipsimple.session import Session from sipsimple.core import SIPURI, ToHeader from sipsimple.streams.rtp.audio import AudioStream from sipsimple.streams.rtp.video import VideoStream from sipsimple.lookup import DNSLookup from application.notification import NotificationCenter, IObserver from zope.interface import implementer @implementer(IObserver) class CallManager: def __init__(self, account): self.account = account self.session = None self.notification_center = NotificationCenter() def make_call(self, target_uri, with_video=False): """Initiate an outgoing call""" # Create the session self.session = Session(self.account) # Subscribe to session notifications self.notification_center.add_observer(self, sender=self.session) # Create media streams streams = [AudioStream()] if with_video: streams.append(VideoStream()) # Parse target URI target = SIPURI.parse(f'sip:{target_uri}') # Perform DNS lookup for routes lookup = DNSLookup() self.notification_center.add_observer(self, sender=lookup) lookup.lookup_sip_proxy(target, ['udp', 'tcp', 'tls']) ``` -------------------------------- ### Initialize File Storage Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Configure file-based storage for SIP application configuration and XCAP documents. This is the recommended backend for production environments. ```python from sipsimple.storage import FileStorage # File-based storage (recommended for production) file_storage = FileStorage('/path/to/config/directory') # Creates: # /path/to/config/directory/config - Configuration file # /path/to/config/directory/xcap/ - XCAP document cache # /path/to/config/directory/otr.key - OTR private key # /path/to/config/directory/otr.trusted - OTR trusted peers # Start application with file storage from sipsimple.application import SIPApplication app = SIPApplication() app.start(file_storage) ``` -------------------------------- ### Configure Global Audio Settings Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Set global audio configuration parameters for SIPSimple, including input/output devices, sample rate, and echo cancellation settings. The settings are saved after modification. ```python from sipsimple.configuration.settings import SIPSimpleSettings settings = SIPSimpleSettings() settings.audio.input_device = 'system_default' settings.audio.output_device = 'system_default' settings.audio.sample_rate = 32000 settings.audio.echo_canceller.enabled = True settings.audio.echo_canceller.tail_length = 2 settings.save() ``` -------------------------------- ### Configure SIPSimple Global Settings Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Access and modify global SDK settings using the SIPSimpleSettings singleton. Configure audio, video, RTP, SIP transport, and TLS parameters. ```python from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.configuration.datatypes import ( AudioCodecList, VideoCodecList, PortRange, SIPTransportList, Path, VideoResolution ) # Get settings singleton settings = SIPSimpleSettings() # Audio configuration settings.audio.input_device = 'system_default' # or specific device name settings.audio.output_device = 'system_default' settings.audio.alert_device = 'system_default' # for ringtones settings.audio.sample_rate = 32000 # 8000, 16000, 32000, 44100, 48000 settings.audio.muted = False settings.audio.silent = False # Suppress alert sounds settings.audio.echo_canceller.enabled = True settings.audio.echo_canceller.tail_length = 2 # Video configuration settings.video.device = 'system_default' settings.video.resolution = VideoResolution('1280x720') settings.video.framerate = 25 settings.video.max_bitrate = None # No limit settings.video.muted = False settings.video.h264.profile = 'baseline' settings.video.h264.level = '3.1' # RTP configuration settings.rtp.port_range = PortRange(50000, 50500) settings.rtp.timeout = 30 # seconds settings.rtp.audio_codec_list = AudioCodecList(['opus', 'G722', 'PCMU', 'PCMA']) settings.rtp.video_codec_list = VideoCodecList(['H264', 'VP8', 'VP9']) # SIP transport configuration settings.sip.udp_port = 0 # 0 = auto-assign settings.sip.tcp_port = 0 settings.sip.tls_port = 0 settings.sip.transport_list = SIPTransportList(['tls', 'tcp', 'udp']) settings.sip.invite_timeout = 90 # seconds # TLS configuration settings.tls.ca_list = Path('/path/to/ca-certificates.crt') settings.tls.certificate = Path('/path/to/client.pem') settings.tls.verify_server = True # File transfer settings settings.file_transfer.directory = Path('~/Downloads') ``` -------------------------------- ### Initialize Memory Storage Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Configure memory-based storage for SIP application configuration. This is useful for testing as configuration is not persisted. ```python from sipsimple.storage import MemoryStorage from sipsimple.application import SIPApplication # Memory-based storage (useful for testing) memory_storage = MemoryStorage() app = SIPApplication() app.start(memory_storage) # Configuration is not persisted ``` -------------------------------- ### Copy pjsip Files for Apple Silicon Build Source: https://github.com/agprojects/python3-sipsimple/blob/master/deps/patches/apple-silicon.txt These commands copy necessary files and directories from a pjsip source to the current project directory. This is required to adapt the build for Apple Silicon by excluding Intel SSE2 instructions and ensuring compatibility with newer FFMPEG patches. ```bash cp -r ../pjsip/third_party/zsrtp third_party/ cp -r ../pjsip/third_party/build/zsrtp third_party/build/ cp ../pjsip/aconfigure aconfigure cp ../pjsip/aconfigure.ac aconfigure.ac cp ../pjsip/base_rev base_rev cp ../pjsip/pjsip/src/pjsip/sip_transport.c pjsip/src/pjsip/ cp ../pjsip/pjsip/src/pjsip/sip_transport_tls.c pjsip/src/pjsip/ cp ../pjsip/pjlib/src/pj/os_core_unix.c pjlib/src/pj/os_core_unix.c cp ../pjsip/pjsip/include/pjsip/sip_msg.h pjsip/include/pjsip/ cp ../pjsip/pjsip-simple/include/pjsip-simple/evsub.h pjsip/include/pjsip-simple/ cp ../pjsip/pjsip/src/pjsip/sip_msg.c pjsip/src/pjsip/sip_msg.c cp ../pjsip/pjsip/src/pjsip-simple/evsub.c pjsip/src/pjsip-simple/evsub.c cp ../pjsip/pjsip/src/pjsip-simple/evsub_msg.c pjsip/src/pjsip-simple/evsub_msg.c cp ../pjsip/pjsip/src/pjsip-ua/sip_inv.c pjsip/src/pjsip-ua/sip_inv.c cp ../pjsip/pjnath/include/pjnath/ice_strans.h pjnath/include/pjnath/ cp ../pjsip/pjnath/src/pjnath/ice_strans.c pjnath/src/pjnath/ice_strans.c cp ../pjsip/third_party/build/os-darwinos.mak third_party/build/os-darwinos.mak cp ../pjsip/third_party/build/os-linux.mak third_party/build/os-linux.mak cp ../pjsip/third_party/build/os-win32.mak third_party/build/os-win32.mak cp ../pjsip/pjlib/src/pj/ssl_sock_ossl.c pjlib/src/pj/ssl_sock_ossl.c cp ~/config_site.h pjlib/include/pj/config_site.h cp ~/config_auto.h pjmedia/include/pjmedia/config_auto.h cp ../pjsip/pjmedia/build/Makefile pjmedia/build/Makefile cp ../pjsip/pjmedia/include/pjmedia/audiodev.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/event.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/format.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/mixer_port.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/rtcp.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/signatures.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/sound_port.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/transport_ice.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/transport_zrtp.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia/vid_stream.h pjmedia/include/pjmedia/ cp ../pjsip/pjmedia/include/pjmedia-audiodev/audiodev_imp.h pjmedia/include/pjmedia-audiodev/ cp ../pjsip/pjmedia/include/pjmedia.h pjmedia/include/pjmedia.h cp ../pjsip/pjmedia/include/pjmedia-videodev/fb_dev.h pjmedia/include/pjmedia-videodev/ cp ../pjsip/pjmedia/include/pjmedia_videodev.h pjmedia/include/pjmedia_videodev.h cp ../pjsip/pjmedia/src/pjmedia-audiodev/audiodev.c pjmedia/src/pjmedia-audiodev/audiodev.c cp ../pjsip/pjmedia/src/pjmedia-audiodev/alsa_dev.c pjmedia/src/pjmedia-audiodev/alsa_dev.c cp ../pjsip/pjmedia/src/pjmedia-audiodev/coreaudio_dev.m pjmedia/src/pjmedia-audiodev/coreaudio_dev.m cp ../pjsip/pjmedia/src/pjmedia-audiodev/wmme_dev.c pjmedia/src/pjmedia-audiodev/wmme_dev.c cp ../pjsip/pjmedia/src/pjmedia/audiodev.c pjmedia/src/pjmedia/audiodev.c cp ../pjsip/pjmedia/src/pjmedia/converter.c pjmedia/src/pjmedia/converter.c cp ../pjsip/pjmedia/src/pjmedia/echo_common.c pjmedia/src/pjmedia/echo_common.c cp ../pjsip/pjmedia/src/pjmedia/echo_webrtc_aec.c pjmedia/src/pjmedia/echo_webrtc_aec.c cp ../pjsip/pjmedia/src/pjmedia/endpoint.c pjmedia/src/pjmedia/endpoint.c cp ../pjsip/pjmedia/src/pjmedia/format.c pjmedia/src/pjmedia/format.c cp ../pjsip/pjmedia/src/pjmedia/mixer_port.c pjmedia/src/pjmedia/mixer_port.c cp ../pjsip/pjmedia/src/pjmedia/rtcp.c pjmedia/src/pjmedia/rtcp.c cp ../pjsip/pjmedia/src/pjmedia/sound_port.c pjmedia/src/pjmedia/sound_port.c cp ../pjsip/pjmedia/src/pjmedia/transport_ice.c pjmedia/src/pjmedia/transport_ice.c cp ../pjsip/pjmedia/src/pjmedia/transport_zrtp.c pjmedia/src/pjmedia/transport_zrtp.c cp ../pjsip/pjmedia/src/pjmedia/vid_stream.c pjmedia/src/pjmedia/vid_stream.c cp ../pjsip/pjmedia/src/pjmedia/vid_tee.c pjmedia/src/pjmedia/vid_tee.c cp ../pjsip/pjmedia/src/pjmedia-codec/opus.c pjmedia/src/pjmedia-codec/opus.c cp ../pjsip/pjmedia/src/pjmedia-codec/vpx.c pjmedia/src/pjmedia-codec/vpx.c cp ../pjsip/pjmedia/src/pjmedia-videodev/avf_dev.m pjmedia/src/pjmedia-videodev/avf_dev.m cp ../pjsip/pjmedia/src/pjmedia-videodev/dshow_dev.c pjmedia/src/pjmedia-videodev/dshow_dev.c cp ../pjsip/pjmedia/src/pjmedia-videodev/dshow_filter.cpp pjmedia/src/pjmedia-videodev/dshow_filter.cpp cp ../pjsip/pjmedia/src/pjmedia-videodev/fb_dev.c pjmedia/src/pjmedia-videodev/fb_dev.c ``` -------------------------------- ### Configure SIP Settings Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Set various SIP-related configurations including logging levels, user agent, and instance ID. Call settings.save() to persist changes. ```python settings.logs.trace_sip = True settings.logs.trace_msrp = True settings.logs.trace_pjsip = False settings.logs.pjsip_level = 5 settings.user_agent = 'MyApp/1.0' settings.instance_id = 'urn:uuid:12345678-1234-1234-1234-123456789abc' settings.save() ``` -------------------------------- ### Media Configuration Update Source: https://github.com/agprojects/python3-sipsimple/blob/master/deps/patches/apple-silicon.txt This command copies a configuration file for pjmedia. Ensure the path to `config_auto.h` is correct for your environment. ```bash cp ~/config_auto.h pjmedia/include/pjmedia/config_auto.h ``` -------------------------------- ### Create and Serialize PIDF Document Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Manually construct a PIDF document with service status, person details, and activities. Serialize the document to XML format. ```python from sipsimple.payloads.pidf import PIDFDocument, Person, Service, Status, Basic, Note from sipsimple.payloads.rpid import Activities, Activity # Build a PIDF document pidf = PIDFDocument(entity='sip:alice@example.com') # Add a service (tuple) service = Service( service_id='pres-12345', status=Status(basic=Basic('open')), contact='sip:alice@example.com' ) pidf.add(service) # Add person element with activities person = Person(person_id='alice') person.activities = Activities() person.activities.add(Activity('busy')) pidf.add(person) # Add a note service.notes.add(Note('In a meeting', lang='en')) # Serialize to XML xml_output = pidf.toxml(pretty_print=True) print(xml_output.decode()) ``` -------------------------------- ### Initialize and Manage AudioStream in SIP Session Source: https://context7.com/agprojects/python3-sipsimple/llms.txt This class finds and manages an AudioStream within a SIP session. It sets up observers for various audio stream events. Ensure the session object is available and contains an AudioStream. ```python from sipsimple.streams.rtp.audio import AudioStream from sipsimple.audio import WavePlayer, WaveRecorder from application.notification import NotificationCenter, IObserver from zope.interface import implementer @implementer(IObserver) class AudioHandler: def __init__(self, session): self.session = session self.audio_stream = None self.notification_center = NotificationCenter() # Find audio stream in session for stream in session.streams: if isinstance(stream, AudioStream): self.audio_stream = stream break if self.audio_stream: self.notification_center.add_observer(self, sender=self.audio_stream) ``` -------------------------------- ### Handle Incoming Calls with Specific Logic Source: https://context7.com/agprojects/python3-sipsimple/llms.txt An observer that specifically handles incoming SIP sessions. It accepts calls with audio streams and rejects video-only calls. ```python # Example: Handling incoming sessions @implementer(IObserver) class IncomingCallHandler: def __init__(self): notification_center = NotificationCenter() notification_center.add_observer(self, name='SIPSessionNewIncoming') def handle_notification(self, notification): if notification.name == 'SIPSessionNewIncoming': session = notification.sender streams = notification.data.streams # Accept audio calls, reject video-only calls audio_streams = [s for s in streams if s.type == 'audio'] if audio_streams: session.accept(audio_streams) else: session.reject(code=488, reason='Not Acceptable Here') ``` -------------------------------- ### Create SIP Route Manually Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Manually create a SIP Route object without performing DNS lookups. This is useful for defining static routes. ```python route = Route( address='proxy.example.com', port=5060, transport='udp' ) ``` -------------------------------- ### Echo Cancellation Configuration Source: https://github.com/agprojects/python3-sipsimple/blob/master/deps/patches/apple-silicon.txt This command copies a configuration file for echo cancellation. Ensure the path to `config_site.h` is correct for your environment. ```bash cp ~/config_site.h pjlib/include/pj/config_site.h ``` -------------------------------- ### Manage Video Stream Producer Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Control the video stream's producer for display and transmission. Use get_video_producer to retrieve the remote video producer, pause_video to stop transmission by setting the producer to None, and resume_video to re-enable it. ```python def get_video_producer(self): """Get the remote video producer for display""" if self.video_stream and self.video_stream.producer: return self.video_stream.producer return None def pause_video(self): """Pause video transmission""" if self.video_stream: self.video_stream.device.producer = None def resume_video(self): """Resume video transmission""" if self.video_stream: from sipsimple.application import SIPApplication self.video_stream.device.producer = SIPApplication.video_device.producer ``` -------------------------------- ### Set SIP Presence State Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Set the presence state for the SIP account. Supported states include 'available', 'away', 'busy', and 'offline'. An optional note can be provided. ```python from sipsimple.account.publication import PresenceState def set_presence_state(self, state, note=None): """ Set presence state for the account. States: 'available', 'away', 'busy', 'offline' """ # Create presence state (account handles PIDF document creation) presence_state = PresenceState(state, note) self.account.presence_state = presence_state ``` -------------------------------- ### Connect SIP Session After DNS Lookup Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Connects the SIP session using the resolved routes. This method is typically called after a DNS lookup completes. ```python def _connect_session(self, routes): """Connect after DNS lookup completes""" target = SIPURI.parse(f'sip:{self.target_uri}') self.session.connect(ToHeader(target), routes, streams) ``` -------------------------------- ### Control Call State (Hold/Resume) Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Provides methods to put an active call on hold and resume it. Checks the session state before attempting these actions. ```python def hold_call(self): """Put call on hold""" if self.session and self.session.state == 'connected': self.session.hold() def resume_call(self): """Resume call from hold""" if self.session and self.session.on_hold: self.session.unhold() ``` -------------------------------- ### Enable OTR Encryption Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Enables Out-of-Band (OTR) encryption for the chat stream. This requires that the chat stream and its encryption module are initialized. ```python def enable_otr_encryption(self): """Enable OTR encryption for the chat""" if self.chat_stream and self.chat_stream.encryption: self.chat_stream.encryption.start() ``` -------------------------------- ### Access SIP Settings Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Retrieve current SIP settings from anywhere in the application. Access specific settings like audio sample rate and supported codecs. ```python current_settings = SIPSimpleSettings() print(f"Sample rate: {current_settings.audio.sample_rate}") print(f"Codecs: {list(current_settings.rtp.audio_codec_list)}") ``` -------------------------------- ### Handle SIP Session Notifications Source: https://context7.com/agprojects/python3-sipsimple/llms.txt A generic handler for notifications received from SIP sessions. It dispatches notifications to specific handler methods based on the notification name. ```python def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, None) if handler: handler(notification) ``` -------------------------------- ### Handle SIP Presence Watcher Info Notifications Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Process incoming watcher info notifications, which indicate who is subscribing to your presence status. This handler lists watchers and their status. ```python def _NH_SIPAccountGotPresenceWinfo(self, notification): """Received watcher info (who is watching us)""" watcher_list = notification.data.watcher_list for watcher in watcher_list: print(f"Watcher: {watcher.sipuri}") print(f" Status: {watcher.status}") # pending, active, terminated print(f" Event: {watcher.event}") ``` -------------------------------- ### Synchronous SIP Proxy DNS Lookup Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Perform a synchronous DNS lookup for SIP proxy routes within a green thread. This method returns routes directly or raises a DNSLookupError on failure. ```python from sipsimple.threading.green import run_in_green_thread @run_in_green_thread def sync_lookup(target): lookup = DNSLookup() try: # lookup_sip_proxy returns routes when used in green thread uri = SIPURI.parse(f'sip:{target}') routes = lookup.lookup_sip_proxy(uri, ['udp', 'tcp', 'tls']) return routes except DNSLookupError as e: print(f"Lookup failed: {e}") return [] ``` -------------------------------- ### Answer Incoming SIP Call Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Answers an incoming call, optionally specifying streams to accept. Defaults to accepting an audio stream if none are provided. ```python def answer_call(self, session, streams=None): """Answer an incoming call""" self.session = session self.notification_center.add_observer(self, sender=session) if streams is None: # Accept with audio stream by default streams = [AudioStream()] session.accept(streams) ``` -------------------------------- ### Parse CPIM Message Source: https://context7.com/agprojects/python3-sipsimple/llms.txt This snippet demonstrates how to parse an incoming CPIM (Common Presence and Instant Messaging) message using the CPIMPayload class. It extracts and prints the sender and content of the message. ```python # Parse incoming CPIM message cpim_data = b'''From: To: DateTime: 2024-01-15T10:30:00Z Content-Type: text/plain Hello, Bob!''' payload = CPIMPayload.decode(cpim_data) print(f"Sender: {payload.sender}") print(f"Content: {payload.content}") ``` -------------------------------- ### Perform SIP Proxy DNS Lookup Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Use DNSLookup to perform RFC 3263 compliant DNS resolution for SIP proxy routes. Specify transport protocols and timeouts for the lookup. ```python from sipsimple.lookup import DNSLookup, DNSLookupError from sipsimple.core import SIPURI, Route from application.notification import NotificationCenter, IObserver from zope.interface import implementer @implementer(IObserver) class DNSHandler: def __init__(self): self.notification_center = NotificationCenter() self.routes = None def lookup_sip_proxy(self, uri, transport_list=None): """Perform DNS lookup for SIP routing""" lookup = DNSLookup() self.notification_center.add_observer(self, sender=lookup) if transport_list is None: transport_list = ['tls', 'tcp', 'udp'] # Perform the lookup asynchronously lookup.lookup_sip_proxy( uri, transport_list, timeout=3.0, # DNS query timeout lifetime=15.0 # Total lookup lifetime ) def lookup_stun_server(self, uri): """Lookup STUN server for NAT traversal""" lookup = DNSLookup() self.notification_center.add_observer(self, sender=lookup) lookup.lookup_service(uri, 'stun') def lookup_msrp_relay(self, uri): """Lookup MSRP relay server""" lookup = DNSLookup() self.notification_center.add_observer(self, sender=lookup) lookup.lookup_service(uri, 'msrprelay') def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, None) if handler: handler(notification) def _NH_DNSLookupDidSucceed(self, notification): result = notification.data.result print("DNS lookup succeeded:") for route in result: if isinstance(route, Route): print(f" Route: {route.address}:{route.port} ({route.transport})") else: # Service lookup result (address, port tuple) print(f" Server: {route[0]}:{route[1]}") self.routes = result def _NH_DNSLookupDidFail(self, notification): error = notification.data.error print(f"DNS lookup failed: {error}") self.routes = None ``` -------------------------------- ### Implement File Transfer Handler Source: https://context7.com/agprojects/python3-sipsimple/llms.txt This class handles sending and receiving files using FileTransferStream. It requires an account object and manages session and transfer stream lifecycles. Observers are added to the notification center for session and stream events. ```python from sipsimple.streams.msrp.filetransfer import FileTransferStream, FileSelector from sipsimple.session import Session from sipsimple.core import SIPURI, ToHeader from application.notification import NotificationCenter, IObserver from zope.interface import implementer @implementer(IObserver) class FileTransferHandler: def __init__(self, account): self.account = account self.session = None self.transfer_stream = None self.notification_center = NotificationCenter() def send_file(self, target_uri, file_path, routes): """Send a file to a remote party""" # Create file selector with file metadata file_selector = FileSelector.for_file(file_path) # Calculate file hash for integrity verification import hashlib with open(file_path, 'rb') as f: file_hash = hashlib.sha1() while chunk := f.read(8192): file_hash.update(chunk) file_selector.hash = file_hash # Create transfer stream self.transfer_stream = FileTransferStream(file_selector, 'sendonly') # Create session self.session = Session(self.account) self.notification_center.add_observer(self, sender=self.session) self.notification_center.add_observer(self, sender=self.transfer_stream) target = SIPURI.parse(f'sip:{target_uri}') self.session.connect(ToHeader(target), routes, [self.transfer_stream]) def accept_incoming_file(self, session, stream, save_path): """Accept an incoming file transfer""" self.session = session self.transfer_stream = stream # Set the path where the file will be saved stream.file_selector.fd = open(save_path, 'wb') self.notification_center.add_observer(self, sender=session) self.notification_center.add_observer(self, sender=stream) session.accept([stream]) def cancel_transfer(self): """Cancel ongoing file transfer""" if self.session: self.session.end() def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, None) if handler: handler(notification) def _NH_SIPSessionNewIncoming(self, notification): session = notification.sender streams = notification.data.streams for stream in streams: if isinstance(stream, FileTransferStream): file_selector = stream.file_selector print(f"Incoming file transfer:") print(f" Name: {file_selector.name}") print(f" Size: {file_selector.size} bytes") print(f" Type: {file_selector.type}") print(f" Hash: {file_selector.hash}") def _NH_FileTransferStreamDidStart(self, notification): stream = notification.sender print(f"File transfer started") print(f" Direction: {stream.direction}") def _NH_FileTransferStreamProgress(self, notification): stream = notification.sender transferred = notification.data.transferred_bytes total = notification.data.file_size if total > 0: progress = (transferred / total) * 100 print(f"Progress: {progress:.1f}% ({transferred}/{total} bytes)") def _NH_FileTransferStreamDidFinish(self, notification): stream = notification.sender print("File transfer completed successfully") # Verify hash if available if stream.file_selector.hash: print(f"File hash verified: {stream.file_selector.hash}") def _NH_FileTransferStreamDidFail(self, notification): reason = notification.data.reason print(f"File transfer failed: {reason}") ``` -------------------------------- ### Handle Audio Stream Recording Stop Notification Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Callback for when the AudioStream stops recording. Logs the filename. ```python def _NH_AudioStreamDidStopRecording(self, notification): print(f"Recording stopped: {notification.data.filename}") ``` -------------------------------- ### Handle Incoming Notifications for AudioStream Events Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Process notifications sent by the AudioStream. This method dispatches notifications to specific handler methods based on their names. ```python def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, None) if handler: handler(notification) ``` -------------------------------- ### Handle SIP Session Failure Notification Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Processes notifications when a SIP session fails, printing the failure reason and associated error code. ```python def _NH_SIPSessionDidFail(self, notification): print(f"Call failed: {notification.data.failure_reason}") print(f"Code: {notification.data.code}") ``` -------------------------------- ### Send Composing Indication Source: https://context7.com/agprojects/python3-sipsimple/llms.txt This method sends a typing indicator to the peer, indicating whether the user is actively typing or has stopped. The refresh parameter sets the interval for the indication. ```python def send_composing_indication(self, is_composing): """Send typing indicator""" if self.chat_stream: state = 'active' if is_composing else 'idle' self.chat_stream.send_composing_indication( state=state, refresh=60 # Refresh interval in seconds ) ``` -------------------------------- ### Handle SIP Session Connected Notification Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Processes notifications when a SIP session successfully connects, printing the remote identity and active stream types. ```python def _NH_SIPSessionDidStart(self, notification): session = notification.sender print(f"Call connected with {session.remote_identity}") for stream in session.streams: print(f" Active stream: {stream.type}") ``` -------------------------------- ### Handle SIP Presence State Notifications Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Process incoming presence state notifications from contacts. This handler parses PIDF documents to extract status information. ```python def _NH_SIPAccountGotPresenceState(self, notification): """Received presence state from contacts""" version = notification.data.version full_state = notification.data.full_state resource_map = notification.data.resource_map for uri, resource in resource_map.items(): print(f"Contact: {uri}") # Parse PIDF document from resource if resource.pidf: for service in resource.pidf.services: if service.status and service.status.basic: print(f" Status: {service.status.basic}") ``` -------------------------------- ### Handle OTR Encryption State Change Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Processes notifications when the OTR encryption state changes. It prints the new state and, if active, the peer's fingerprint and verification status. ```python def _NH_ChatStreamOTREncryptionStateChanged(self, notification): state = notification.data.state print(f"OTR encryption state: {state}") if self.chat_stream.encryption.active: print(f"Peer fingerprint: {self.chat_stream.encryption.peer_fingerprint}") print(f"Verified: {self.chat_stream.encryption.verified}") ``` -------------------------------- ### Handle DTMF Notification Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Callback for when an AudioStream receives a DTMF digit. Logs the received digit. ```python def _NH_AudioStreamGotDTMF(self, notification): digit = notification.data.digit print(f"Received DTMF digit: {digit}") ``` -------------------------------- ### Handle RTP Hold State Change Notification Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Callback for when the RTP stream's hold state changes. Logs the new hold state and the originator. ```python def _NH_RTPStreamDidChangeHoldState(self, notification): on_hold = notification.data.on_hold originator = notification.data.originator print(f"Audio {originator} hold state: {on_hold}") ``` -------------------------------- ### Verify OTR Peer Identity Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Verifies the peer's identity using the Simple Messaging Protocol (SMP) with a shared secret and an optional question. This is part of OTR encryption. ```python def verify_otr_peer(self, secret, question=None): """Verify peer identity using SMP""" if self.chat_stream and self.chat_stream.encryption.active: self.chat_stream.encryption.smp_verify(secret, question) ``` -------------------------------- ### Send DTMF Tones Source: https://context7.com/agprojects/python3-sipsimple/llms.txt Sends DTMF tones through active media streams that support DTMF functionality. ```python def send_dtmf(self, digit): """Send DTMF tone""" for stream in self.session.streams: if hasattr(stream, 'send_dtmf'): stream.send_dtmf(digit) ```