mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-12-11 10:56:38 +00:00
Code Issues 1 Projects Releases Wiki Activity
gstreamer/subprojects/gst-devtools/validate/launcher/utils.py
Thibault Saunier a56abe2c9d validate: launcher: Lower some 'coredumpctl' info messages 
Those are verbose and are not really useful in our context

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/5666>
2023-11-15 19:07:36 +00:00

1845 lines
64 KiB
Python
Raw Blame History

#!/usr/bin/env python3
#
# Copyright (c) 2013,Thibault Saunier <thibault.saunier@collabora.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.
""" Some utilies. """
try:
import config
except ImportError:
from . import config
import json
import numbers
import os
import platform
import re
import shutil
import shlex
import signal
import subprocess
import sys
import tempfile
import time
import urllib.request
import urllib.error
import urllib.parse
from .loggable import Loggable, warning
from operator import itemgetter
from xml.etree import ElementTree
from collections import defaultdict
from fractions import Fraction
GST_SECOND = int(1000000000)
DEFAULT_TIMEOUT = 30
DEFAULT_MAIN_DIR = os.path.join(config.BUILDDIR, "subprojects", "gst-integration-testsuites")
DEFAULT_GST_QA_ASSETS = os.path.join(config.SRCDIR, "subprojects", "gst-integration-testsuites")
USING_SUBPROJECT = os.path.exists(os.path.join(config.BUILDDIR, "subprojects", "gst-integration-testsuites"))
if not USING_SUBPROJECT:
DEFAULT_MAIN_DIR = os.path.join(os.path.expanduser("~"), "gst-validate")
DEFAULT_GST_QA_ASSETS = os.path.join(DEFAULT_MAIN_DIR, "gstreamer", "subprojects", "gst-integration-testsuites")
DEFAULT_MAIN_DIR = os.environ.get('GST_VALIDATE_LAUNCHER_MAIN_DIR', DEFAULT_MAIN_DIR)
DEFAULT_TESTSUITES_DIRS = [os.path.join(DEFAULT_GST_QA_ASSETS, "testsuites")]
DISCOVERER_COMMAND = "gst-discoverer-1.0"
# Use to set the duration from which a test is considered as being 'long'
LONG_TEST = 40
class Result(object):
NOT_RUN = "Not run"
FAILED = "Failed"
TIMEOUT = "Timeout"
PASSED = "Passed"
SKIPPED = "Skipped"
KNOWN_ERROR = "Known error"
class Protocols(object):
HTTP = "http"
FILE = "file"
PUSHFILE = "pushfile"
HLS = "hls"
DASH = "dash"
RTSP = "rtsp"
IMAGESEQUENCE = "imagesequence"
@staticmethod
def needs_clock_sync(protocol):
if protocol in [Protocols.HLS, Protocols.DASH]:
return True
return False
def is_tty():
return hasattr(sys.stdout, 'isatty') and sys.stdout.isatty()
def supports_ansi_colors():
if 'GST_VALIDATE_LAUNCHER_FORCE_COLORS' in os.environ:
return True
platform = sys.platform
supported_platform = platform != 'win32' or 'ANSICON' in os.environ
if not supported_platform or not is_tty():
return False
return True
class Colors(object):
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
def desactivate_colors():
Colors.HEADER = ''
Colors.OKBLUE = ''
Colors.OKGREEN = ''
Colors.WARNING = ''
Colors.FAIL = ''
Colors.ENDC = ''
if not supports_ansi_colors():
desactivate_colors()
def mkdir(directory):
try:
os.makedirs(directory)
except os.error:
pass
def which(name, extra_path=None):
exts = [_f for _f in os.environ.get('PATHEXT', '').split(os.pathsep) if _f]
path = os.environ.get('PATH', '')
if extra_path:
path = extra_path + os.pathsep + path
if not path:
return []
for p in path.split(os.pathsep):
p = os.path.join(p, name)
if os.access(p, os.X_OK):
return p
for e in exts:
pext = p + e
if os.access(pext, os.X_OK):
return pext
return None
def get_color_for_result(result):
if result is Result.FAILED:
color = Colors.FAIL
elif result is Result.TIMEOUT:
color = Colors.WARNING
elif result is Result.PASSED:
color = Colors.OKGREEN
else:
color = Colors.OKBLUE
return color
last_carriage_return_len = 0
def printc(message, color="", title=False, title_char='', end="\n"):
global last_carriage_return_len
if title or title_char:
length = 0
for line in message.split("\n"):
if len(line) > length:
length = len(line)
if length == 0:
length = len(message)
needed_spaces = ' ' * max(0, last_carriage_return_len - length)
if title is True:
message = length * "=" + needed_spaces + "\n" \
+ str(message) + "\n" + length * '='
else:
message = str(message) + needed_spaces + "\n" + \
length * title_char
if hasattr(message, "result") and color == '':
color = get_color_for_result(message.result)
if not is_tty():
end = "\n"
message = str(message)
message += ' ' * max(0, last_carriage_return_len - len(message))
if end == '\r':
term_width = shutil.get_terminal_size((80, 20))[0]
if len(message) > term_width:
message = message[0:term_width - 2] + '…'
last_carriage_return_len = len(message)
else:
last_carriage_return_len = 0
sys.stdout.write(color + str(message) + Colors.ENDC + end)
sys.stdout.flush()
def launch_command(command, color=None, fails=False):
printc(command, Colors.OKGREEN, True)
res = os.system(command)
if res != 0 and fails is True:
raise subprocess.CalledProcessError(res, "%s failed" % command)
def path2url(path):
return urllib.parse.urljoin('file:', urllib.request.pathname2url(path))
def is_windows():
platname = platform.system().lower()
return platname == 'windows' or 'mingw' in platname
def url2path(url):
path = urllib.parse.urlparse(url).path
if "win32" in sys.platform:
if path[0] == '/':
return path[1:] # We need to remove the first '/' on windows
path = urllib.parse.unquote(path)
return path
def isuri(string):
url = urllib.parse.urlparse(string)
if url.scheme != "" and url.scheme != "":
return True
return False
def touch(fname, times=None):
with open(fname, 'a'):
os.utime(fname, times)
def get_subclasses(klass, env):
subclasses = []
for symb in env.items():
try:
if issubclass(symb[1], klass) and not symb[1] is klass:
subclasses.append(symb[1])
except TypeError:
pass
return subclasses
def TIME_ARGS(time):
return "%u:%02u:%02u.%09u" % (time / (GST_SECOND * 60 * 60),
(time / (GST_SECOND * 60)) % 60,
(time / GST_SECOND) % 60,
time % GST_SECOND)
def look_for_file_in_source_dir(subdir, name):
root_dir = os.path.abspath(os.path.dirname(
os.path.join(os.path.dirname(os.path.abspath(__file__)))))
p = os.path.join(root_dir, subdir, name)
if os.path.exists(p):
return p
return None
# Returns the path $top_src_dir/@subdir/@name if running from source, or
# $DATADIR/gstreamer-1.0/validate/@name if not
def get_data_file(subdir, name):
# Are we running from sources?
p = look_for_file_in_source_dir(subdir, name)
if p:
return p
# Look in system data dirs
p = os.path.join(config.DATADIR, 'gstreamer-1.0', 'validate', name)
if os.path.exists(p):
return p
return None
#
# Some utilities to parse gst-validate output #
#
def gsttime_from_tuple(stime):
return int((int(stime[0]) * 3600 + int(stime[1]) * 60 + int(stime[2])) * GST_SECOND + int(stime[3]))
timeregex = re.compile(r'(?P<_0>.+):(?P<_1>.+):(?P<_2>.+)\.(?P<_3>.+)')
def parse_gsttimeargs(time):
stime = list(map(itemgetter(1), sorted(
timeregex.match(time).groupdict().items())))
return int((int(stime[0]) * 3600 + int(stime[1]) * 60 + int(stime[2])) * GST_SECOND + int(stime[3]))
def get_duration(media_file):
duration = 0
res = ''
try:
res = subprocess.check_output(
[DISCOVERER_COMMAND, media_file]).decode()
except subprocess.CalledProcessError:
# gst-media-check returns !0 if seeking is not possible, we do not care
# in that case.
pass
for line in res.split('\n'):
if "Duration: " in line:
duration = parse_gsttimeargs(line.replace("Duration: ", ""))
break
return duration
def get_scenarios():
GST_VALIDATE_COMMAND = "gst-validate-1.0"
os.system("%s --scenarios-defs-output-file %s" % (GST_VALIDATE_COMMAND,
))
def get_gst_build_valgrind_suppressions():
if hasattr(get_gst_build_valgrind_suppressions, "data"):
return get_gst_build_valgrind_suppressions.data
get_gst_build_valgrind_suppressions.data = []
if not os.path.exists(os.path.join(config.SRCDIR, "subprojects")):
return get_gst_build_valgrind_suppressions.data
for suppression_path in ["gstreamer/tests/check/gstreamer.supp",
"gst-plugins-base/tests/check/gst-plugins-base.supp",
"gst-plugins-good/tests/check/gst-plugins-good.supp",
"gst-plugins-bad/tests/check/gst-plugins-bad.supp",
"gst-plugins-ugly/tests/check/gst-plugins-ugly.supp",
"gst-libav/tests/check/gst-libav.supp",
"gst-devtools/validate/data/gstvalidate.supp",
"libnice/tests/libnice.supp",
"libsoup/tests/libsoup.supp",
"glib/glib.supp",
"gst-python/testsuite/gstpython.supp",
"gst-python/testsuite/python.supp",
]:
suppression = os.path.join(config.SRCDIR, "subprojects", suppression_path)
if os.path.exists(suppression):
get_gst_build_valgrind_suppressions.data.append(suppression)
return get_gst_build_valgrind_suppressions.data
class BackTraceGenerator(Loggable):
__instance = None
_command_line_regex = re.compile(r'Command Line: (.*)\n')
_timestamp_regex = re.compile(r'Timestamp: .*\((\d*)s ago\)')
_pid_regex = re.compile(r'PID: (\d+) \(.*\)')
def __init__(self):
Loggable.__init__(self)
self.in_flatpak = os.path.exists("/usr/manifest.json")
if self.in_flatpak:
coredumpctl = ['flatpak-spawn', '--host', 'coredumpctl']
else:
coredumpctl = ['coredumpctl']
coredumpctl.append('-q')
try:
subprocess.check_output(coredumpctl)
self.coredumpctl = coredumpctl
except Exception as e:
self.warning(e)
self.coredumpctl = None
self.gdb = shutil.which('gdb')
@classmethod
def get_default(cls):
if not cls.__instance:
cls.__instance = BackTraceGenerator()
return cls.__instance
def get_trace(self, test):
if not test.process.returncode:
return self.get_trace_on_running_process(test)
if self.coredumpctl:
return self.get_trace_from_systemd(test)
self.debug("coredumpctl not present, and it is the only"
" supported way to get backtraces for now.")
return None
def get_trace_on_running_process(self, test):
if not self.gdb:
return "Can not generate stack trace as `gdb` is not" \
"installed."
gdb = ['gdb', '-ex', 't a a bt', '-batch',
'-p', str(test.process.pid)]
try:
return subprocess.check_output(
gdb, stderr=subprocess.STDOUT, timeout=30).decode()
except Exception as e:
return "Could not run `gdb` on process (pid: %d):\n%s" % (
test.process.pid, e)
def get_trace_from_systemd(self, test):
for ntry in range(10):
if ntry != 0:
# Loopping, it means we conceder the logs might not be ready
# yet.
time.sleep(1)
if not self.in_flatpak:
coredumpctl = self.coredumpctl + ['info', str(test.process.pid)]
else:
newer_than = time.strftime("%a %Y-%m-%d %H:%M:%S %Z", time.localtime(test._starting_time))
coredumpctl = self.coredumpctl + ['info', os.path.basename(test.command[0]),
'--since', newer_than]
try:
info = subprocess.check_output(coredumpctl, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError:
# The trace might not be ready yet
time.sleep(1)
continue
info = info.decode()
try:
pid = self._pid_regex.findall(info)[0]
except IndexError:
self.debug("Backtrace could not be found yet, trying harder.")
continue
application = test.process.args[0]
command_line = BackTraceGenerator._command_line_regex.findall(info)[0]
if shlex.split(command_line)[0] != application:
self.debug("PID: %s -- executable %s != test application: %s" % (
pid, command_line[0], test.application))
# The trace might not be ready yet
continue
if not BackTraceGenerator._timestamp_regex.findall(info):
self.debug("Timestamp %s is more than 1min old",
re.findall(r'Timestamp: .*', info))
# The trace might not be ready yet
continue
bt_all = None
if self.gdb:
try:
with tempfile.NamedTemporaryFile() as stderr:
coredump = subprocess.check_output(self.coredumpctl + ['dump', pid],
stderr=stderr)
with tempfile.NamedTemporaryFile() as tf:
tf.write(coredump)
tf.flush()
gdb = ['gdb', '-ex', 't a a bt', '-ex', 'quit', application, tf.name]
bt_all = subprocess.check_output(
gdb, stderr=subprocess.STDOUT).decode()
info += "\nThread apply all bt:\n\n%s" % (
bt_all.replace('\n', '\n' + 15 * ' '))
except Exception as e:
self.error("Could not get backtrace from gdb: %s" % e)
return info
return None
ALL_GITLAB_ISSUES = defaultdict(list)
def check_bugs_resolution(bugs_definitions):
bugz = {}
gitlab_issues = defaultdict(list)
regexes = {}
mr_id = os.environ.get('CI_MERGE_REQUEST_IID')
mr_closes_issues = []
if mr_id:
gitlab_url = f"{os.environ['CI_API_V4_URL']}/projects/{os.environ['CI_MERGE_REQUEST_PROJECT_ID']}/merge_requests/{mr_id}/closes_issues"
for issue in json.load(urllib.request.urlopen(gitlab_url)):
mr_closes_issues.append(issue)
res = True
for regex, bugs in bugs_definitions:
if isinstance(bugs, str):
bugs = [bugs]
for bug in bugs:
url = urllib.parse.urlparse(bug)
if "gitlab" in url.netloc:
components = [c for c in url.path.split('/') if c]
if len(components) not in [4, 5]:
printc("\n + %s \n --> bug: %s\n --> Status: Not a proper gitlab report" % (regex, bug),
Colors.WARNING)
continue
project_id = components[0] + '%2F' + components[1]
issue_id = int(components[-1])
for issue in mr_closes_issues:
url = urllib.parse.urlparse(issue['web_url'])
closing_issue_project = '%2F'.join([c for c in url.path.split('/') if c][0:2])
if project_id == closing_issue_project and issue['iid'] == issue_id:
res = False
printc("\n + %s \n --> %s: '%s'\n ==> Will be closed by current MR %s\n\n===> Remove blacklisting before merging." % (
regex, issue['web_url'], issue['title'], issue['state']), Colors.FAIL)
gitlab_url = "https://%s/api/v4/projects/%s/issues/%s" % (url.hostname, project_id, issue_id)
if gitlab_url in ALL_GITLAB_ISSUES:
continue
gitlab_issues[gitlab_url].append(regex)
ALL_GITLAB_ISSUES[gitlab_url].append(regex)
continue
if "bugzilla" not in url.netloc:
continue
query = urllib.parse.parse_qs(url.query)
_id = query.get('id')
if not _id:
printc("\n + '%s' -- Can't check bug '%s'" %
(regex, bug), Colors.WARNING)
continue
if isinstance(_id, list):
_id = _id[0]
regexes[_id] = (regex, bug)
url_parts = tuple(list(url)[:3] + ['', '', ''])
ids = bugz.get(url_parts, [])
ids.append(_id)
bugz[url_parts] = ids
for gitlab_url, regexe in gitlab_issues.items():
try:
issue = json.load(urllib.request.urlopen(gitlab_url))
except Exception as e:
printc("\n + Could not properly check bugs status for: %s (%s)"
% (gitlab_url, e), Colors.FAIL)
continue
if issue['state'] in ['closed']:
printc("\n + %s \n --> %s: '%s'\n ==> Bug CLOSED already (status: %s)" % (
regexe, issue['web_url'], issue['title'], issue['state']), Colors.FAIL)
res = False
for url_parts, ids in bugz.items():
url_parts = list(url_parts)
query = {'id': ','.join(ids)}
query['ctype'] = 'xml'
url_parts[4] = urllib.parse.urlencode(query)
try:
res = urllib.request.urlopen(urllib.parse.urlunparse(url_parts))
except Exception as e:
printc("\n + Could not properly check bugs status for: %s (%s)"
% (urllib.parse.urlunparse(url_parts), e), Colors.FAIL)
continue
root = ElementTree.fromstring(res.read())
bugs = root.findall('./bug')
if len(bugs) != len(ids):
printc("\n + Could not properly check bugs status on server %s" %
urllib.parse.urlunparse(url_parts), Colors.FAIL)
continue
for bugelem in bugs:
status = bugelem.findtext('./bug_status')
bugid = bugelem.findtext('./bug_id')
regex, bug = regexes[bugid]
desc = bugelem.findtext('./short_desc')
if not status:
printc("\n + %s \n --> bug: %s\n --> Status: UNKNOWN" % (regex, bug),
Colors.WARNING)
continue
if not status.lower() in ['new', 'verified']:
printc("\n + %s \n --> bug: #%s: '%s'\n ==> Bug CLOSED already (status: %s)" % (
regex, bugid, desc, status), Colors.WARNING)
res = False
printc("\n + %s \n --> bug: #%s: '%s'\n --> Status: %s" % (
regex, bugid, desc, status), Colors.OKGREEN)
if not res:
printc("\n==> Some bugs marked as known issues have been (or will be) closed!", Colors.FAIL)
return res
def kill_subprocess(owner, process, timeout, subprocess_ids=None):
if process is None:
return
stime = time.time()
res = process.poll()
waittime = 0.05
killsig = None
if not is_windows():
killsig = signal.SIGINT
while res is None:
try:
owner.debug("Subprocess is still alive, sending KILL signal")
if is_windows():
subprocess.call(
['taskkill', '/F', '/T', '/PID', str(process.pid)])
else:
if subprocess_ids:
for subprocess_id in subprocess_ids:
os.kill(subprocess_id, killsig)
else:
process.send_signal(killsig)
time.sleep(waittime)
waittime *= 2
except OSError:
pass
if not is_windows() and time.time() - stime > timeout / 4:
killsig = signal.SIGKILL
if time.time() - stime > timeout:
printc("Could not kill %s subprocess after %s second"
" Something is really wrong, => EXITING"
% (owner, timeout), Colors.FAIL)
return
res = process.poll()
return res
def format_config_template(extra_data, config_text, test_name):
# Variables available for interpolation inside config blocks.
extra_vars = extra_data.copy()
if 'validate-flow-expectations-dir' in extra_vars and \
'validate-flow-actual-results-dir' in extra_vars:
expectations_dir = os.path.join(extra_vars['validate-flow-expectations-dir'],
test_name.replace('.', os.sep))
actual_results_dir = os.path.join(extra_vars['validate-flow-actual-results-dir'],
test_name.replace('.', os.sep))
extra_vars['validateflow'] = "validateflow, expectations-dir=\"%s\", actual-results-dir=\"%s\"" % (
expectations_dir, actual_results_dir)
if 'ssim-results-dir' in extra_vars:
ssim_results = extra_vars['ssim-results-dir']
extra_vars['ssim'] = "validatessim, result-output-dir=\"%s\", output-dir=\"%s\"" % (
os.path.join(ssim_results, test_name.replace('.', os.sep), 'diff-images'),
os.path.join(ssim_results, test_name.replace('.', os.sep), 'images'),
)
return config_text % extra_vars
def get_fakesink_for_media_type(media_type, needs_clock=False):
extra = ""
if media_type == "video" and needs_clock:
extra = 'max-lateness=20000000'
return f"fake{media_type}sink sync={needs_clock} {extra}"
class InvalidValueError(ValueError):
"""Received value is invalid"""
def __init__(self, name, value, expect):
ValueError.__init__(
self, "Invalid value {!r} for {}. Expect {}.".format(
value, name, expect))
def wrong_type_for_arg(val, expect_type_name, arg_name):
"""Raise exception in response to a wrong argument type"""
raise TypeError(
"Expect a {} type for the '{}' argument. Received a {} type."
"".format(expect_type_name, arg_name, type(val).__name__))
class DeserializeError(Exception):
"""Receive an incorrectly serialized value"""
MAX_LEN = 20
def __init__(self, read, reason):
if len(read) > self.MAX_LEN:
read = read[:self.MAX_LEN] + "..."
Exception.__init__(
self, "Could not deserialize the string ({}) because it {}."
"".format(read, reason))
class GstStructure(Loggable):
"""
This implementation has been copied from OpenTimelineIO.
Note that the types are to correspond to GStreamer/GES GTypes,
rather than python types.
Current supported GTypes:
GType Associated Accepted
Python type aliases
======================================
gint int int, i
glong int
gint64 int
guint int uint, u
gulong int
guint64 int
gfloat float float, f
gdouble float double, d
gboolean bool boolean,
bool, b
string str or None str, s
GstFraction str or fraction
Fraction
GstStructure GstStructure structure
schema
GstCaps GstCaps
schema
Note that other types can be given: these must be given as strings
and the user will be responsible for making sure they are already in
a serialized form.
"""
INT_TYPES = ("int", "glong", "gint64")
UINT_TYPES = ("uint", "gulong", "guint64")
FLOAT_TYPES = ("float", "double")
BOOLEAN_TYPE = "boolean"
FRACTION_TYPE = "fraction"
STRING_TYPE = "string"
STRUCTURE_TYPE = "structure"
CAPS_TYPE = "GstCaps"
KNOWN_TYPES = INT_TYPES + UINT_TYPES + FLOAT_TYPES + (
BOOLEAN_TYPE, FRACTION_TYPE, STRING_TYPE, STRUCTURE_TYPE,
CAPS_TYPE)
TYPE_ALIAS = {
"i": "int",
"gint": "int",
"u": "uint",
"guint": "uint",
"f": "float",
"gfloat": "float",
"d": "double",
"gdouble": "double",
"b": BOOLEAN_TYPE,
"bool": BOOLEAN_TYPE,
"gboolean": BOOLEAN_TYPE,
"GstFraction": FRACTION_TYPE,
"str": STRING_TYPE,
"s": STRING_TYPE,
"GstStructure": STRUCTURE_TYPE
}
def __init__(self, name=None, fields=None):
if name is None:
name = "Unnamed"
if fields is None:
fields = {}
if type(name) is not str:
wrong_type_for_arg(name, "str", "name")
self._check_name(name)
self.name = name
try:
fields = dict(fields)
except (TypeError, ValueError):
wrong_type_for_arg(fields, "dict", "fields")
self.fields = {}
for key in fields:
entry = fields[key]
if type(entry) is not tuple:
try:
entry = tuple(entry)
except (TypeError, ValueError):
raise TypeError(
"Expect dict to be filled with tuple-like "
"entries")
if len(entry) != 2:
raise TypeError(
"Expect dict to be filled with 2-entry tuples")
self.set(key, *entry)
def __repr__(self):
return "GstStructure({!r}, {!r})".format(self.name, self.fields)
UNKNOWN_PREFIX = "[UNKNOWN]"
@classmethod
def _make_type_unknown(cls, _type):
return cls.UNKNOWN_PREFIX + _type
# note the sqaure brackets make the type break the TYPE_FORMAT
@classmethod
def _is_unknown_type(cls, _type):
return _type[:len(cls.UNKNOWN_PREFIX)] == cls.UNKNOWN_PREFIX
@classmethod
def _get_unknown_type(cls, _type):
return _type[len(cls.UNKNOWN_PREFIX):]
def _field_to_str(self, key):
"""Return field in a serialized form"""
_type, value = self.fields[key]
if type(key) is not str:
raise TypeError("Found a key that is not a str type")
if type(_type) is not str:
raise TypeError(
"Found a type name that is not a str type")
self._check_key(key)
_type = self.TYPE_ALIAS.get(_type, _type)
if self._is_unknown_type(_type):
_type = self._get_unknown_type(_type)
self._check_type(_type)
self._check_unknown_typed_value(value)
# already in serialized form
else:
self._check_type(_type)
value = self.serialize_value(_type, value)
return "{}=({}){}".format(key, _type, value)
def _fields_to_str(self):
write = []
for key in self.fields:
write.append(", {}".format(self._field_to_str(key)))
return "".join(write)
def _name_to_str(self):
"""Return the name in a serialized form"""
return self._check_name(self.name)
def __str__(self):
"""Emulates gst_structure_to_string"""
return "{}{};".format(self._name_to_str(), self._fields_to_str())
def get_type_name(self, key):
"""Return the field type"""
_type = self.fields[key][0]
return _type
def get_value(self, key):
"""Return the field value"""
value = self.fields[key][1]
return value
def __getitem__(self, key):
return self.get_value(key)
def __len__(self):
return len(self.fields)
@staticmethod
def _val_type_err(typ, val, expect):
raise TypeError(
"Received value ({!s}) is a {} rather than a {}, even "
"though the {} type was given".format(
val, type(val).__name__, expect, typ))
def set(self, key, _type, value):
"""Set a field to the given typed value"""
if type(key) is not str:
wrong_type_for_arg(key, "str", "key")
if type(_type) is not str:
wrong_type_for_arg(_type, "str", "_type")
_type = self.TYPE_ALIAS.get(_type, _type)
if self.fields.get(key) == (_type, value):
return
self._check_key(key)
type_is_unknown = True
if self._is_unknown_type(_type):
# this can happen if the user is setting a GstStructure
# using a preexisting GstStructure, the type will then
# be passed and marked as unknown
_type = self._get_unknown_type(_type)
self._check_type(_type)
else:
self._check_type(_type)
if _type in self.INT_TYPES:
type_is_unknown = False
if not isinstance(value, int):
self._val_type_err(_type, value, "int")
elif _type in self.UINT_TYPES:
type_is_unknown = False
if not isinstance(value, int):
self._val_type_err(_type, value, "int")
if value < 0:
raise InvalidValueError(
"value", value, "a positive integer for {} "
"types".format(_type))
elif _type in self.FLOAT_TYPES:
type_is_unknown = False
if type(value) is not float:
self._val_type_err(_type, value, "float")
elif _type == self.BOOLEAN_TYPE:
type_is_unknown = False
if type(value) is not bool:
self._val_type_err(_type, value, "bool")
elif _type == self.FRACTION_TYPE:
type_is_unknown = False
if type(value) is Fraction:
value = value
elif type(value) is str:
try:
Fraction(value)
except ValueError:
raise InvalidValueError(
"value", value, "a fraction for the {} "
"types".format(_type))
else:
self._val_type_err(_type, value, "Fraction or str")
elif _type == self.STRING_TYPE:
type_is_unknown = False
if value is not None and type(value) is not str:
self._val_type_err(_type, value, "str or None")
elif _type == self.STRUCTURE_TYPE:
type_is_unknown = False
if not isinstance(value, GstStructure):
self._val_type_err(_type, value, "GstStructure")
elif _type == self.CAPS_TYPE:
type_is_unknown = False
if not isinstance(value, GstCaps):
self._val_type_err(_type, value, "GstCaps")
if type_is_unknown:
self._check_unknown_typed_value(value)
warning('GstStructure',
"The GstStructure type {} with the value ({}) is "
"unknown. The value will be stored and serialized as "
"given.".format(_type, value))
_type = self._make_type_unknown(_type)
self.fields[key] = (_type, value)
def get(self, key, default=None):
"""Return the raw value associated with key"""
if key in self.fields:
value = self.get_value(key)
return value
return default
def get_typed(self, key, expect_type, default=None):
"""
Return the raw value associated with key if its type matches.
Raises a warning if a value exists under key but is of the
wrong type.
"""
if type(expect_type) is not str:
wrong_type_for_arg(expect_type, "str", "expect_type")
expect_type = self.TYPE_ALIAS.get(expect_type, expect_type)
if key in self.fields:
type_name = self.get_type_name(key)
if expect_type == type_name:
value = self.get_value(key)
return value
warning('GstStructure',
"The structure {} contains a value under {}, but is "
"a {}, rather than the expected {} type".format(
self.name, key, type_name, expect_type))
return default
def values(self):
"""Return a list of all values contained in the structure"""
return [self.get_value(key) for key in self.fields]
def values_of_type(self, _type):
"""
Return a list of all values contained of the given type in the
structure
"""
if type(_type) is not str:
wrong_type_for_arg(_type, "str", "_type")
_type = self.TYPE_ALIAS.get(_type, _type)
return [self.get_value(key) for key in self.fields
if self.get_type_name(key) == _type]
ASCII_SPACES = r"(\\?[ \t\n\r\f\v])*"
END_FORMAT = r"(?P<end>" + ASCII_SPACES + r")"
NAME_FORMAT = r"(?P<name>[a-zA-Z][a-zA-Z0-9/_.:-]*)"
# ^Format requirement for the name of a GstStructure
SIMPLE_STRING = r"[a-zA-Z0-9_+/:.-]+"
# see GST_ASCII_CHARS (below)
KEY_FORMAT = r"(?P<key>" + SIMPLE_STRING + r")"
# NOTE: GstStructure technically allows more general keys, but
# these can break the parsing.
TYPE_FORMAT = r"(?P<type>" + SIMPLE_STRING + r")"
BASIC_VALUE_FORMAT = \
r'(?P<value>("(\\.|[^"])*")|(' + SIMPLE_STRING + r'))'
# consume simple string or a string between quotes. Second will
# consume anything that is escaped, including a '"'
# NOTE: \\. is used rather than \\" since:
# + '"start\"end;"' should be captured as '"start\"end"' since
# the '"' is escaped.
# + '"start\\"end;"' should be captured as '"start\\"' since the
# '\' is escaped, not the '"'
# In the fist case \\. will consume '\"', and in the second it will
# consumer '\\', as desired. The second would not work with just \\"
@staticmethod
def _check_against_regex(check, regex, name):
if not regex.fullmatch(check):
raise InvalidValueError(
name, check, "to match the regular expression {}"
"".format(regex.pattern))
return check
NAME_REGEX = re.compile(NAME_FORMAT)
KEY_REGEX = re.compile(KEY_FORMAT)
TYPE_REGEX = re.compile(TYPE_FORMAT)
@classmethod
def _check_name(cls, name):
return cls._check_against_regex(name, cls.NAME_REGEX, "name")
@classmethod
def _check_key(cls, key):
return cls._check_against_regex(key, cls.KEY_REGEX, "key")
@classmethod
def _check_type(cls, _type):
return cls._check_against_regex(_type, cls.TYPE_REGEX, "type")
@classmethod
def _check_unknown_typed_value(cls, value):
if type(value) is not str:
cls._val_type_err("unknown", value, "string")
try:
# see if the value could be successfully parsed in again
ret_type, ret_val, _ = cls._parse_value(value, False)
except DeserializeError as err:
raise InvalidValueError(
"value", value, "unknown-typed values to be in a "
"serialized format ({!s})".format(err))
else:
if ret_type is not None:
raise InvalidValueError(
"value", value, "unknown-typed values to *not* "
"start with a type specification, only the "
"serialized value should be given")
if ret_val != value:
raise InvalidValueError(
"value", value, "unknown-typed values to be the "
"same as its parsed value {}".format(ret_val))
PARSE_NAME_REGEX = re.compile(
ASCII_SPACES + NAME_FORMAT + END_FORMAT)
@classmethod
def _parse_name(cls, read):
match = cls.PARSE_NAME_REGEX.match(read)
if match is None:
raise DeserializeError(
read, "does not start with a correct name")
name = match.group("name")
read = read[match.end("end"):]
return name, read
@classmethod
def _parse_range_list_array(cls, read):
start = read[0]
end = {'[': ']', '{': '}', '<': '>'}.get(start)
read = read[1:]
values = [start, ' ']
first = True
while read and read[0] != end:
if first:
first = False
else:
if read and read[0] != ',':
DeserializeError(
read, "does not contain a comma between listed "
"items")
values.append(", ")
read = read[1:]
_type, value, read = cls._parse_value(read, False)
if _type is not None:
if cls._is_unknown_type(_type):
# remove unknown marker for serialization
_type = cls._get_unknown_type(_type)
values.extend(('(', _type, ')'))
values.append(value)
if not read:
raise DeserializeError(
read, "ended before {} could be found".format(end))
read = read[1:] # skip past 'end'
match = cls.END_REGEX.match(read) # skip whitespace
read = read[match.end("end"):]
# NOTE: we are ignoring the incorrect cases where a range
# has 0, 1 or 4+ values! This is the users responsiblity.
values.extend((' ', end))
return "".join(values), read
FIELD_START_REGEX = re.compile(
ASCII_SPACES + KEY_FORMAT + ASCII_SPACES + r"=" + END_FORMAT)
FIELD_TYPE_REGEX = re.compile(
ASCII_SPACES + r"(\(" + ASCII_SPACES + TYPE_FORMAT
+ ASCII_SPACES + r"\))?" + END_FORMAT)
FIELD_VALUE_REGEX = re.compile(
ASCII_SPACES + BASIC_VALUE_FORMAT + END_FORMAT)
END_REGEX = re.compile(END_FORMAT)
@classmethod
def _parse_value(cls, read, deserialize=True):
match = cls.FIELD_TYPE_REGEX.match(read)
# match shouldn't be None since the (TYPE_FORMAT) is optional
# and the rest is just ASCII_SPACES
_type = match.group("type")
if _type is None and deserialize:
# if deserialize is False, the (type) is optional
raise DeserializeError(
read, "does not contain a valid '(type)' format")
_type = cls.TYPE_ALIAS.get(_type, _type)
type_is_unknown = True
read = read[match.end("end"):]
if read and read[0] in ('[', '{', '<'):
# range/list/array types
# this is an unknown type, even though _type itself may
# be known. e.g. a list on integers will have _type as 'int'
# but the corresponding value can not be deserialized as an
# integer
value, read = cls._parse_range_list_array(read)
if deserialize:
# prevent printing on subsequent calls if we find a
# list within a list, etc.
warning('GstStructure',
"GstStructure received a range/list/array of type "
"{}, which can not be deserialized. Storing the "
"value as {}.".format(_type, value))
else:
match = cls.FIELD_VALUE_REGEX.match(read)
if match is None:
raise DeserializeError(
read, "does not have a valid value format")
read = read[match.end("end"):]
value = match.group("value")
if deserialize:
if _type in cls.KNOWN_TYPES:
type_is_unknown = False
try:
value = cls.deserialize_value(_type, value)
except DeserializeError as err:
raise DeserializeError(
read, "contains an invalid typed value "
"({!s})".format(err))
else:
warning('GstStructure',
"GstStructure found a type {} that is unknown. "
"The corresponding value ({}) will not be "
"deserialized and will be stored as given."
"".format(_type, value))
if type_is_unknown and _type is not None:
_type = cls._make_type_unknown(_type)
return _type, value, read
@classmethod
def _parse_field(cls, read):
match = cls.FIELD_START_REGEX.match(read)
if match is None:
raise DeserializeError(
read, "does not have a valid 'key=...' format")
key = match.group("key")
read = read[match.end("end"):]
_type, value, read = cls._parse_value(read)
return key, _type, value, read
@classmethod
def _parse_fields(cls, read):
if type(read) is not str:
wrong_type_for_arg(read, "str", "read")
fields = {}
while read and read[0] != ';':
if read and read[0] != ',':
DeserializeError(
read, "does not separate fields with commas")
read = read[1:]
key, _type, value, read = cls._parse_field(read)
fields[key] = (_type, value)
if read:
# read[0] == ';'
read = read[1:]
return fields, read
@classmethod
def new_from_str(cls, read):
"""
Returns a new instance of GstStructure, based on the Gst library
function gst_structure_from_string.
Strings obtained from the GstStructure str() method can be
parsed in to recreate the original GstStructure.
"""
if type(read) is not str:
wrong_type_for_arg(read, "str", "read")
name, read = cls._parse_name(read)
fields = cls._parse_fields(read)[0]
return GstStructure(name=name, fields=fields)
@staticmethod
def _val_read_err(typ, val):
raise DeserializeError(
val, "does not translated to the {} type".format(typ))
@classmethod
def deserialize_value(cls, _type, value):
"""Return the value as the corresponding type"""
if type(_type) is not str:
wrong_type_for_arg(_type, "str", "_type")
if type(value) is not str:
wrong_type_for_arg(value, "str", "value")
_type = cls.TYPE_ALIAS.get(_type, _type)
if _type in cls.INT_TYPES or _type in cls.UINT_TYPES:
try:
value = int(value)
except ValueError:
cls._val_read_err(_type, value)
if _type in cls.UINT_TYPES and value < 0:
cls._val_read_err(_type, value)
elif _type in cls.FLOAT_TYPES:
try:
value = float(value)
except ValueError:
cls._val_read_err(_type, value)
elif _type == cls.BOOLEAN_TYPE:
try:
value = cls.deserialize_boolean(value)
except DeserializeError:
cls._val_read_err(_type, value)
elif _type == cls.FRACTION_TYPE:
try:
value = Fraction(value)
except ValueError:
cls._val_read_err(_type, value)
elif _type == cls.STRING_TYPE:
try:
value = cls.deserialize_string(value)
except DeserializeError as err:
raise DeserializeError(
value, "does not translate to a string ({!s})"
"".format(err))
elif _type == cls.STRUCTURE_TYPE:
try:
value = cls.deserialize_structure(value)
except DeserializeError as err:
raise DeserializeError(
value, "does not translate to a GstStructure ({!s})"
"".format(err))
elif _type == cls.CAPS_TYPE:
try:
value = cls.deserialize_caps(value)
except DeserializeError as err:
raise DeserializeError(
value, "does not translate to a GstCaps ({!s})"
"".format(err))
else:
raise ValueError(
"The type {} is unknown, so the value ({}) can not "
"be deserialized.".format(_type, value))
return value
@classmethod
def serialize_value(cls, _type, value):
"""Serialize the typed value as a string"""
if type(_type) is not str:
wrong_type_for_arg(_type, "str", "_type")
_type = cls.TYPE_ALIAS.get(_type, _type)
if _type in cls.INT_TYPES + cls.UINT_TYPES + cls.FLOAT_TYPES \
+ (cls.FRACTION_TYPE, ):
return str(value)
if _type == cls.BOOLEAN_TYPE:
return cls.serialize_boolean(value)
if _type == cls.STRING_TYPE:
return cls.serialize_string(value)
if _type == cls.STRUCTURE_TYPE:
return cls.serialize_structure(value)
if _type == cls.CAPS_TYPE:
return cls.serialize_caps(value)
raise ValueError(
"The type {} is unknown, so the value ({}) can not be "
"serialized.".format(_type, str(value)))
# see GST_ASCII_IS_STRING in gst_private.h
GST_ASCII_CHARS = [
ord(line) for line in "abcdefghijklmnopqrstuvwxyz"
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"0123456789"
"_-+/:."
]
LEADING_OCTAL_CHARS = [ord(line) for line in "0123"]
OCTAL_CHARS = [ord(line) for line in "01234567"]
@classmethod
def serialize_string(cls, value):
"""
Emulates gst_value_serialize_string.
Accepts a bytes, str or None type.
Returns a str type.
"""
if value is not None and type(value) is not str:
wrong_type_for_arg(value, "None or str", "value")
return cls._wrap_string(value)
@classmethod
def _wrap_string(cls, read):
if read is None:
return "NULL"
if read == "NULL":
return "\"NULL\""
if type(read) is bytes:
pass
elif type(read) is str:
read = read.encode()
else:
wrong_type_for_arg(read, "None, str, or bytes", "read")
if not read:
return '""'
added_wrap = False
ser_string_list = []
for byte in read:
if byte in cls.GST_ASCII_CHARS:
ser_string_list.append(chr(byte))
elif byte < 0x20 or byte >= 0x7f:
ser_string_list.append("\\{:03o}".format(byte))
added_wrap = True
else:
ser_string_list.append("\\" + chr(byte))
added_wrap = True
if added_wrap:
ser_string_list.insert(0, '"')
ser_string_list.append('"')
return "".join(ser_string_list)
@classmethod
def deserialize_string(cls, read):
"""
Emulates gst_value_deserialize_string.
Accepts a str type.
Returns a str or None type.
"""
if type(read) is not str:
wrong_type_for_arg(read, "str", "read")
if read == "NULL":
return None
if not read:
return ""
if read[0] != '"' or read[-1] != '"':
return read
return cls._unwrap_string(read)
@classmethod
def _unwrap_string(cls, read):
"""Emulates gst_string_unwrap"""
if type(read) is not bytes:
read_array = read.encode()
byte_list = []
bytes_iter = iter(read_array)
def next_byte():
try:
return next(bytes_iter)
except StopIteration:
raise DeserializeError(read, "end unexpectedly")
byte = next_byte()
if byte != ord('"'):
raise DeserializeError(
read, "does not start with '\"', but ends with '\"'")
while True:
byte = next_byte()
if byte in cls.GST_ASCII_CHARS:
byte_list.append(byte)
elif byte == ord('"'):
try:
next(bytes_iter)
except StopIteration:
# expect there to be no more bytes
break
raise DeserializeError(
read, "contains an un-escaped '\"' before the end")
elif byte == ord('\\'):
byte = next_byte()
if byte in cls.LEADING_OCTAL_CHARS:
# could be the start of an octal
byte2 = next_byte()
byte3 = next_byte()
if byte2 in cls.OCTAL_CHARS and byte3 in cls.OCTAL_CHARS:
nums = [b - ord('0') for b in (byte, byte2, byte3)]
byte = (nums[0] << 6) + (nums[1] << 3) + nums[2]
byte_list.append(byte)
else:
raise DeserializeError(
read, "contains the start of an octal "
"sequence but not the end")
else:
if byte == 0:
raise DeserializeError(
read, "contains a null byte after an escape")
byte_list.append(byte)
else:
raise DeserializeError(
read, "contains an unexpected un-escaped character")
out_str = bytes(bytearray(byte_list))
try:
return out_str.decode()
except (UnicodeError, ValueError):
raise DeserializeError(
read, "contains invalid utf-8 byte sequences")
@staticmethod
def serialize_boolean(value):
"""
Emulates gst_value_serialize_boolean.
Accepts bool type.
Returns a str type.
"""
if type(value) is not bool:
wrong_type_for_arg(value, "bool", "value")
if value:
return "true"
return "false"
@staticmethod
def deserialize_boolean(read):
"""
Emulates gst_value_deserialize_boolean.
Accepts str type.
Returns a bool type.
"""
if type(read) is not str:
wrong_type_for_arg(read, "str", "read")
if read.lower() in ("true", "t", "yes", "1"):
return True
if read.lower() in ("false", "f", "no", "0"):
return False
raise DeserializeError(read, "is an unknown boolean value")
@classmethod
def serialize_structure(cls, value):
"""
Emulates gst_value_serialize_structure.
Accepts a GstStructure.
Returns a str type.
"""
if not isinstance(value, GstStructure):
wrong_type_for_arg(value, "GstStructure", "value")
return cls._wrap_string(str(value))
@classmethod
def deserialize_structure(cls, read):
"""
Emulates gst_value_serialize_structure.
Accepts a str type.
Returns a GstStructure.
"""
if type(read) is not str:
wrong_type_for_arg(read, "str", "read")
if read[0] == '"':
# NOTE: since all GstStructure strings end with ';', we
# don't ever expect the above to *not* be true, but the
# GStreamer library allows for this case
try:
read = cls._unwrap_string(read)
# NOTE: in the GStreamer library, serialized
# GstStructure and GstCaps strings are sent to
# _priv_gst_value_parse_string with unescape set to
# TRUE. What this essentially does is replace "\x" with
# just "x". Since caps and structure strings should only
# contain printable ascii characters before they are
# passed to _wrap_string, this should be equivalent to
# calling _unwrap_string. Our method is more clearly a
# reverse of the serialization method.
except DeserializeError as err:
raise DeserializeError(
read, "could not be unwrapped as a string ({!s})"
"".format(err))
return GstStructure.new_from_str(read)
@classmethod
def serialize_caps(cls, value):
"""
Emulates gst_value_serialize_caps.
Accepts a GstCaps.
Returns a str type.
"""
if not isinstance(value, GstCaps):
wrong_type_for_arg(value, "GstCaps", "value")
return cls._wrap_string(str(value))
@classmethod
def deserialize_caps(cls, read):
"""
Emulates gst_value_serialize_caps.
Accepts a str type.
Returns a GstCaps.
"""
if type(read) is not str:
wrong_type_for_arg(read, "str", "read")
if read[0] == '"':
# can be not true if a caps only contains a single empty
# structure, or is ALL or NONE
try:
read = cls._unwrap_string(read)
except DeserializeError as err:
raise DeserializeError(
read, "could not be unwrapped as a string ({!s})"
"".format(err))
return GstCaps.new_from_str(read)
@staticmethod
def _escape_string(read):
"""
Emulates some of g_strescape's behaviour in
ges_marker_list_serialize
"""
# NOTE: in the original g_strescape, all the special characters
# '\b', '\f', '\n', '\r', '\t', '\v', '\' and '"' are escaped,
# and all characters in the range 0x01-0x1F and non-ascii
# characters are replaced by an octal sequence
# (similar to _wrap_string).
# However, a caps string should only contain printable ascii
# characters, so it should be sufficient to simply escape '\'
# and '"'.
escaped = ['"']
for character in read:
if character in ('"', '\\'):
escaped.append('\\')
escaped.append(character)
escaped.append('"')
return "".join(escaped)
@staticmethod
def _unescape_string(read):
"""
Emulates behaviour of _priv_gst_value_parse_string with
unescape set to TRUE. This should undo _escape_string
"""
if read[0] != '"':
return read
character_iter = iter(read)
def next_char():
try:
return next(character_iter)
except StopIteration:
raise DeserializeError(read, "ends unexpectedly")
next_char() # skip '"'
unescaped = []
while True:
character = next_char()
if character == '"':
break
if character == '\\':
unescaped.append(next_char())
else:
unescaped.append(character)
return "".join(unescaped)
class GstCapsFeatures():
"""
Mimicking a GstCapsFeatures.
"""
def __init__(self, *features):
"""
Initialize the GstCapsFeatures.
'features' should be a series of feature names as strings.
"""
self.is_any = False
self.features = []
for feature in features:
if type(feature) is not str:
wrong_type_for_arg(feature, "strs", "features")
self._check_feature(feature)
self.features.append(feature)
# NOTE: if 'features' is a str, rather than a list of strs
# then this will iterate through all of its characters! But,
# a single character can not match the feature regular
# expression.
def __getitem__(self, index):
return self.features[index]
def __len__(self):
return len(self.features)
@classmethod
def new_any(cls):
features = cls()
features.is_any = True
return features
# Based on gst_caps_feature_name_is_valid
FEATURE_FORMAT = r"(?P<feature>[a-zA-Z]*:[a-zA-Z][a-zA-Z0-9]*)"
FEATURE_REGEX = re.compile(FEATURE_FORMAT)
@classmethod
def _check_feature(cls, feature):
if not cls.FEATURE_REGEX.fullmatch(feature):
raise InvalidValueError(
"feature", feature, "to match the regular expression "
"{}".format(cls.FEATURE_REGEX.pattern))
PARSE_FEATURE_REGEX = re.compile(
r" *" + FEATURE_FORMAT + "(?P<end>)")
@classmethod
def new_from_str(cls, read):
"""
Returns a new instance of GstCapsFeatures, based on the Gst
library function gst_caps_features_from_string.
Strings obtained from the GstCapsFeatures str() method can be
parsed in to recreate the original GstCapsFeatures.
"""
if type(read) is not str:
wrong_type_for_arg(read, "str", "read")
if read == "ANY":
return cls.new_any()
first = True
features = []
while read:
if first:
first = False
else:
if read[0] != ',':
DeserializeError(
read, "does not separate features with commas")
read = read[1:]
match = cls.PARSE_FEATURE_REGEX.match(read)
if match is None:
raise DeserializeError(
read, "does not match the regular expression {}"
"".format(cls.PARSE_FEATURE_REGEX.pattern))
features.append(match.group("feature"))
read = read[match.end("end"):]
return cls(*features)
def __repr__(self):
if self.is_any:
return "GstCapsFeatures.new_any()"
write = ["GstCapsFeatures("]
first = True
for feature in self.features:
if first:
first = False
else:
write.append(", ")
write.append(repr(feature))
write.append(")")
return "".join(write)
def __str__(self):
"""Emulate gst_caps_features_to_string"""
if not self.features and self.is_any:
return "ANY"
write = []
first = True
for feature in self.features:
if type(feature) is not str:
raise TypeError(
"Found a feature that is not a str type")
if first:
first = False
else:
write.append(", ")
write.append(feature)
return "".join(write)
class GstCaps:
GST_CAPS_FLAG_ANY = 1 << 4
# from GST_MINI_OBJECT_FLAG_LAST
def __init__(self, *structs):
"""
Initialize the GstCaps.
'structs' should be a series of GstStructures, and
GstCapsFeatures pairs:
struct0, features0, struct1, features1, ...
None may be given in place of a GstCapsFeatures, in which case
an empty features is assigned to the structure.
Note, this instance will need to take ownership of any given
GstStructure or GstCapsFeatures.
"""
if len(structs) % 2:
raise InvalidValueError(
"*structs", structs, "an even number of arguments")
self.flags = 0
self.structs = []
struct = None
for index, arg in enumerate(structs):
if index % 2 == 0:
struct = arg
else:
self.append(struct, arg)
def get_structure(self, index):
"""Return the GstStructure at the given index"""
return self.structs[index][0]
def get_features(self, index):
"""Return the GstStructure at the given index"""
return self.structs[index][1]
def __getitem__(self, index):
return self.get_structure(index)
def __len__(self):
return len(self.structs)
def __iter__(self):
for s in self.structs:
yield s
@classmethod
def new_any(cls):
caps = cls()
caps.flags = cls.GST_CAPS_FLAG_ANY
return caps
def is_any(self):
return self.flags & self.GST_CAPS_FLAG_ANY != 0
FEATURES_FORMAT = r"\((?P<features>[^)]*)\)"
NAME_FEATURES_REGEX = re.compile(
GstStructure.ASCII_SPACES + GstStructure.NAME_FORMAT
+ r"(" + FEATURES_FORMAT + r")?" + GstStructure.END_FORMAT)
@classmethod
def new_from_str(cls, read):
"""
Returns a new instance of GstCaps, based on the Gst library
function gst_caps_from_string.
Strings obtained from the GstCaps str() method can be parsed in
to recreate the original GstCaps.
"""
if type(read) is not str:
wrong_type_for_arg(read, "str", "read")
if read == "ANY":
return cls.new_any()
if read in ("EMPTY", "NONE"):
return cls()
structs = []
# restriction-caps is otherwise serialized in the format:
# "struct-name-nums(feature), "
# "field1=(type1)val1, field2=(type2)val2; "
# "struct-name-alphas(feature), "
# "fieldA=(typeA)valA, fieldB=(typeB)valB"
# Note the lack of ';' for the last structure, and the
# '(feature)' is optional.
#
# NOTE: gst_caps_from_string also accepts:
# "struct-name(feature"
# without the final ')', but this must be the end of the string,
# but we will require that this final ')' is still given
while read:
match = cls.NAME_FEATURES_REGEX.match(read)
if match is None:
raise DeserializeError(
read, "does not match the regular expression {}"
"".format(cls.NAME_FEATURE_REGEX.pattern))
read = read[match.end("end"):]
name = match.group("name")
features = match.group("features")
# NOTE: features may be None since the features part of the
# regular expression is optional
if features is None:
features = GstCapsFeatures()
else:
features = GstCapsFeatures.new_from_str(features)
fields, read = GstStructure._parse_fields(read)
structs.append(GstStructure(name, fields))
structs.append(features)
return cls(*structs)
def __repr__(self):
if self.is_any():
return "GstCaps.new_any()"
write = ["GstCaps("]
first = True
for struct in self.structs:
if first:
first = False
else:
write.append(", ")
write.append(repr(struct[0]))
write.append(", ")
write.append(repr(struct[1]))
write.append(")")
return "".join(write)
def __str__(self):
"""Emulate gst_caps_to_string"""
if self.is_any():
return "ANY"
if not self.structs:
return "EMPTY"
first = True
write = []
for struct, features in self.structs:
if first:
first = False
else:
write.append("; ")
write.append(struct._name_to_str())
if features.is_any or features.features:
# NOTE: is gst_caps_to_string, the feature will not
# be written if it only contains the
# GST_FEATURE_MEMORY_SYSTEM_MEMORY feature, since this
# considered equal to being an empty features.
# We do not seem to require this behaviour
write.append("({!s})".format(features))
write.append(struct._fields_to_str())
return "".join(write)
def append(self, structure, features=None):
"""Append a structure with the given features"""
if not isinstance(structure, GstStructure):
wrong_type_for_arg(structure, "GstStructure", "structure")
if features is None:
features = GstCapsFeatures()
if not isinstance(features, GstCapsFeatures):
wrong_type_for_arg(
features, "GstCapsFeatures or None", "features")
self.structs.append((structure, features))
Reference in a new issue View git blame Copy permalink