mirror of
https://gitlab.freedesktop.org/gstreamer/gstreamer.git
synced 2024-09-22 03:50:12 +00:00
285 lines
12 KiB
Python
285 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# Copyright (c) 2018, Matthew Waters <matthew@centricular.com>
|
|
#
|
|
# This program is free software; you can redistribute it and/or
|
|
# modify it under the terms of the GNU Lesser General Public
|
|
# License as published by the Free Software Foundation; either
|
|
# version 2.1 of the License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
|
# Lesser General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Lesser General Public
|
|
# License along with this program; if not, write to the
|
|
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
|
|
# Boston, MA 02110-1301, USA.
|
|
|
|
import os
|
|
import sys
|
|
import argparse
|
|
import json
|
|
import logging
|
|
|
|
from signalling import WebRTCSignallingClient, RemoteWebRTCObserver
|
|
from actions import ActionObserver
|
|
from client import WebRTCClient
|
|
from browser import Browser, create_driver
|
|
from enums import SignallingState, NegotiationState, DataChannelState, Actions
|
|
|
|
import gi
|
|
gi.require_version("GLib", "2.0")
|
|
from gi.repository import GLib
|
|
gi.require_version("Gst", "1.0")
|
|
from gi.repository import Gst
|
|
gi.require_version("GstWebRTC", "1.0")
|
|
from gi.repository import GstWebRTC
|
|
gi.require_version("GstSdp", "1.0")
|
|
from gi.repository import GstSdp
|
|
gi.require_version("GstValidate", "1.0")
|
|
from gi.repository import GstValidate
|
|
|
|
FORMAT = '%(asctime)-23s %(levelname)-7s %(thread)d %(name)-24s\t%(funcName)-24s %(message)s'
|
|
LEVEL = os.environ.get("LOGLEVEL", "DEBUG")
|
|
logging.basicConfig(level=LEVEL, format=FORMAT)
|
|
l = logging.getLogger(__name__)
|
|
|
|
class WebRTCApplication(object):
|
|
def __init__(self, server, id_, peerid, scenario_name, browser_name, html_source, test_name=None):
|
|
self.server = server
|
|
self.peerid = peerid
|
|
self.html_source = html_source
|
|
self.id = id_
|
|
self.scenario_name = scenario_name
|
|
self.browser_name = browser_name
|
|
self.test_name = test_name
|
|
|
|
def _init_validate(self, scenario_file):
|
|
self.runner = GstValidate.Runner.new()
|
|
self.monitor = GstValidate.Monitor.factory_create(
|
|
self.client.pipeline, self.runner, None)
|
|
self._scenario = GstValidate.Scenario.factory_create(
|
|
self.runner, self.client.pipeline, self.scenario_name)
|
|
self._scenario.connect("done", self._on_scenario_done)
|
|
self._scenario.props.execute_on_idle = True
|
|
if not self._scenario.props.handles_states:
|
|
self.client.pipeline.set_state(Gst.State.PLAYING)
|
|
|
|
def _on_scenario_done(self, scenario):
|
|
l.error ('scenario done')
|
|
GLib.idle_add(self.quit)
|
|
|
|
def _connect_actions(self, actions):
|
|
def on_action(atype, action):
|
|
"""
|
|
From a validate action, perform the action as required
|
|
"""
|
|
if atype == Actions.CREATE_OFFER:
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
c.create_offer()
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.CREATE_ANSWER:
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
c.create_answer()
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.WAIT_FOR_NEGOTIATION_STATE:
|
|
states = [NegotiationState(action.structure["state"]), NegotiationState.ERROR]
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
state = c.wait_for_negotiation_states(states)
|
|
return GstValidate.ActionReturn.OK if state != NegotiationState.ERROR else GstValidate.ActionReturn.ERROR
|
|
elif atype == Actions.ADD_STREAM:
|
|
self.client.add_stream(action.structure["pipeline"])
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.ADD_DATA_CHANNEL:
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
c.add_data_channel(action.structure["id"])
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.SEND_DATA_CHANNEL_STRING:
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
channel = c.find_channel (action.structure["id"])
|
|
channel.send_string (action.structure["msg"])
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.WAIT_FOR_DATA_CHANNEL_STATE:
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
states = [DataChannelState(action.structure["state"]), DataChannelState.ERROR]
|
|
channel = c.find_channel (action.structure["id"])
|
|
state = channel.wait_for_states(states)
|
|
return GstValidate.ActionReturn.OK if state != DataChannelState.ERROR else GstValidate.ActionReturn.ERROR
|
|
elif atype == Actions.CLOSE_DATA_CHANNEL:
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
channel = c.find_channel (action.structure["id"])
|
|
channel.close()
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.WAIT_FOR_DATA_CHANNEL:
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
state = c.wait_for_data_channel(action.structure["id"])
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.WAIT_FOR_DATA_CHANNEL_STRING:
|
|
assert action.structure["which"] in ("local", "remote")
|
|
c = self.client if action.structure["which"] == "local" else self.remote_client
|
|
channel = c.find_channel (action.structure["id"])
|
|
channel.wait_for_message(action.structure["msg"])
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.WAIT_FOR_NEGOTIATION_NEEDED:
|
|
self.client.wait_for_negotiation_needed(action.structure["generation"])
|
|
return GstValidate.ActionReturn.OK
|
|
elif atype == Actions.SET_WEBRTC_OPTIONS:
|
|
self.client.set_options (action.structure)
|
|
self.remote_client.set_options (action.structure)
|
|
return GstValidate.ActionReturn.OK
|
|
else:
|
|
assert "Not reached" == ""
|
|
|
|
actions.action.connect (on_action)
|
|
|
|
def _connect_client_observer(self):
|
|
def on_offer_created(offer):
|
|
text = offer.sdp.as_text()
|
|
msg = json.dumps({'sdp': {'type': 'offer', 'sdp': text}})
|
|
self.signalling.send(msg)
|
|
self.client.on_offer_created.connect(on_offer_created)
|
|
|
|
def on_answer_created(answer):
|
|
text = answer.sdp.as_text()
|
|
msg = json.dumps({'sdp': {'type': 'answer', 'sdp': text}})
|
|
self.signalling.send(msg)
|
|
self.client.on_answer_created.connect(on_answer_created)
|
|
|
|
def on_ice_candidate(mline, candidate):
|
|
msg = json.dumps({'ice': {'sdpMLineIndex': str(mline), 'candidate' : candidate}})
|
|
self.signalling.send(msg)
|
|
self.client.on_ice_candidate.connect(on_ice_candidate)
|
|
|
|
def on_pad_added(pad):
|
|
if pad.get_direction() != Gst.PadDirection.SRC:
|
|
return
|
|
self.client.add_stream_with_pad('fakesink', pad)
|
|
self.client.on_pad_added.connect(on_pad_added)
|
|
|
|
def _connect_signalling_observer(self):
|
|
def have_json(msg):
|
|
if 'sdp' in msg:
|
|
sdp = msg['sdp']
|
|
res, sdpmsg = GstSdp.SDPMessage.new_from_text(sdp['sdp'])
|
|
sdptype = GstWebRTC.WebRTCSDPType.ANSWER if sdp['type'] == 'answer' else GstWebRTC.WebRTCSDPType.OFFER
|
|
desc = GstWebRTC.WebRTCSessionDescription.new(sdptype, sdpmsg)
|
|
self.client.set_remote_description(desc)
|
|
elif 'ice' in msg:
|
|
ice = msg['ice']
|
|
candidate = ice['candidate']
|
|
sdpmlineindex = ice['sdpMLineIndex']
|
|
self.client.add_ice_candidate(sdpmlineindex, candidate)
|
|
self.signalling.have_json.connect(have_json)
|
|
|
|
def error(msg):
|
|
# errors are unexpected
|
|
l.error ('Unexpected error: ' + msg)
|
|
GLib.idle_add(self.quit)
|
|
GLib.idle_add(sys.exit, -20)
|
|
self.signalling.error.connect(error)
|
|
|
|
def _init(self):
|
|
self.main_loop = GLib.MainLoop()
|
|
|
|
self.client = WebRTCClient()
|
|
self._connect_client_observer()
|
|
|
|
self.signalling = WebRTCSignallingClient(self.server, self.id)
|
|
self.remote_client = RemoteWebRTCObserver (self.signalling)
|
|
self._connect_signalling_observer()
|
|
|
|
actions = ActionObserver()
|
|
actions.register_action_types()
|
|
self._connect_actions(actions)
|
|
|
|
# wait for the signalling server to start up before creating the browser
|
|
self.signalling.wait_for_states([SignallingState.OPEN])
|
|
self.signalling.hello()
|
|
|
|
self.browser = Browser(create_driver(self.browser_name))
|
|
self.browser.open(self.html_source)
|
|
|
|
browser_id = self.browser.get_peer_id ()
|
|
assert browser_id == self.peerid
|
|
|
|
self.signalling.create_session(self.peerid)
|
|
test_name = self.test_name if self.test_name else self.scenario_name
|
|
self.remote_client.set_title (test_name)
|
|
|
|
self._init_validate(self.scenario_name)
|
|
|
|
def quit(self):
|
|
# Stop signalling first so asyncio doesn't keep us alive on weird failures
|
|
l.info('quiting')
|
|
self.signalling.stop()
|
|
l.info('signalling stopped')
|
|
self.main_loop.quit()
|
|
l.info('main loop stopped')
|
|
self.client.stop()
|
|
l.info('client stopped')
|
|
self.browser.driver.quit()
|
|
l.info('browser exitted')
|
|
|
|
def run(self):
|
|
try:
|
|
self._init()
|
|
l.info("app initialized")
|
|
self.main_loop.run()
|
|
l.info("loop exited")
|
|
except:
|
|
l.exception("Fatal error")
|
|
self.quit()
|
|
raise
|
|
|
|
def parse_options():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('id', help='ID of this client', type=int)
|
|
parser.add_argument('--peer-id', help='ID of the peer to connect to', type=int)
|
|
parser.add_argument('--server', help='Signalling server to connect to, eg "wss://127.0.0.1:8443"')
|
|
parser.add_argument('--html-source', help='HTML page to open in the browser', default=None)
|
|
parser.add_argument('--scenario', help='Scenario file to execute', default=None)
|
|
parser.add_argument('--browser', help='Browser name to use', default=None)
|
|
parser.add_argument('--name', help='Name of the test', default=None)
|
|
return parser.parse_args()
|
|
|
|
def init():
|
|
Gst.init(None)
|
|
GstValidate.init()
|
|
|
|
args = parse_options()
|
|
if not args.scenario:
|
|
args.scenario = os.environ.get('GST_VALIDATE_SCENARIO', None)
|
|
# if we have both manual scenario creation and env, then the scenario
|
|
# is executed twice...
|
|
if 'GST_VALIDATE_SCENARIO' in os.environ:
|
|
del os.environ['GST_VALIDATE_SCENARIO']
|
|
if not args.scenario:
|
|
raise ValueError("No scenario file provided")
|
|
if not args.server:
|
|
raise ValueError("No server location provided")
|
|
if not args.peer_id:
|
|
raise ValueError("No peer id provided")
|
|
if not args.html_source:
|
|
raise ValueError("No HTML page provided")
|
|
if not args.browser:
|
|
raise ValueError("No Browser provided")
|
|
|
|
return args
|
|
|
|
def run():
|
|
args = init()
|
|
w = WebRTCApplication (args.server, args.id, args.peer_id, args.scenario, args.browser, args.html_source, test_name=args.name)
|
|
return w.run()
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(run())
|