| #!/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)) |