KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _dumper method in Airtest

Best Python code snippet using Airtest

vve_service.py

Source:vve_service.py

1import sys2from abc import ABCMeta, abstractmethod3from json import JSONDecodeError4from pathlib import Path5from typing import Any, Dict, List, Optional67import requests8from requests.models import Response910from vve_cli.dumper import TaggedDumper11from vve_cli.main import IntervalTimer121314class VveClient:15    __urlorigin: str1617    def __init__(self, host: str, port: int) -> None:18        self.__urlorigin = "http://{}:{:d}".format(host, port)1920    def get(self, url):21        return requests.get(self.__urlorigin + url)2223    def post(self, url, json=None, params=None, headers=None):24        return requests.post(25            self.__urlorigin + url, json=json, params=params, headers=headers26        )272829class MetaEndPoint(metaclass=ABCMeta):30    def run(self, client, **kwargs) -> Any:31        t = IntervalTimer()3233        response = self._request(client, **kwargs)3435        response_time = t.elapsed()36        self._put_log(response_time, response, **kwargs)3738        return self._set_content(response, **kwargs)3940    @abstractmethod41    def _request(self, client, **kwargs) -> Response:42        pass4344    @abstractmethod45    def _put_log(self, response_time: float, response: Response, **kwargs) -> None:46        pass4748    @abstractmethod49    def _set_content(self, response: Response, **kwargs) -> Any:50        pass515253class EndPoint(MetaEndPoint):54    def __init__(self, api_name: str, dump_dir: Optional[Path] = None) -> None:55        self._api_name = api_name56        self._dump_dir = dump_dir57        self._dumper: Optional[TaggedDumper] = None5859    def _put_log(self, response_time: float, response: Response, **kwargs) -> None:60        print(61            f"{self._api_name:>18}: {response_time:7.3f} [sec]",62            file=sys.stderr,63        )646566class InformationQueryAPI(EndPoint):67    def _request(self, client, **kwargs) -> Response:68        return client.get(f"/{self._api_name}")6970    def _set_content(self, response: Response, **kwargs) -> Any:71        if self._dump_dir is not None:72            if self._dumper is None:73                self._dumper = TaggedDumper(self._dump_dir / self._api_name, "json")74            self._dumper.dump(response.text)7576        try:77            json_response = response.json()78        except JSONDecodeError:79            json_response = {}80        return json_response818283class TextToAudioQueryAPI(EndPoint):84    def _request(self, client, text: str, speaker_id: int, **kwargs) -> Response:85        return client.post(86            f"/{self._api_name}", params={"text": text, "speaker": speaker_id}87        )8889    def _put_log(90        self, response_time: float, response: Response, text: str, **kwargs91    ) -> None:92        print(93            (94                f"{self._api_name:>18}: {response_time:7.3f} [sec]"95                f" : {len(text):3d} : {text}"96            ),97            file=sys.stderr,98        )99100    def _set_content(101        self, response: Response, speaker_id: int, tag: str = "dump", **kwargs102    ) -> Any:103        if self._dump_dir is not None:104            if self._dumper is None:105                self._dumper = TaggedDumper(106                    self._dump_dir / self._api_name, "json", is_indexed=True107                )108            self._dumper.dump(response.text, tag + f"_s{speaker_id:02d}")109110        try:111            json_response = response.json()112        except JSONDecodeError:113            json_response = {}114        return json_response115116117class SynthesisAPI(EndPoint):118    def _request(119        self, client, audio_query: Dict[str, Any], speaker_id: int, **kwargs120    ) -> Response:121        return client.post(122            f"/{self._api_name}",123            json=audio_query,124            params={"speaker": speaker_id},125            headers={"Content-Type": "application/json"},126        )127128    def _put_log(129        self,130        response_time: float,131        response: Response,132        audio_query: Dict[str, Any],133        **kwargs,134    ) -> None:135        speech_text = "".join(136            [137                mora["text"]138                for accent_phrase in audio_query["accent_phrases"]139                for mora in accent_phrase["moras"]140                + (141                    [accent_phrase["pause_mora"]]142                    if "pause_mora" in accent_phrase143                    and accent_phrase["pause_mora"] is not None144                    else []145                )146            ]147        )148        print(149            (150                f"{self._api_name:>18}: {response_time:7.3f} [sec]"151                f" : {len(speech_text):3d} : {speech_text}"152            ),153            file=sys.stderr,154        )155156    def _set_content(157        self, response: Response, speaker_id: int, tag: str = "dump", **kwargs158    ) -> Any:159        if self._dump_dir is not None:160            if self._dumper is None:161                self._dumper = TaggedDumper(162                    self._dump_dir / self._api_name, "wav", is_indexed=True163                )164            self._dumper.dump(response.content, tag + f"_s{speaker_id:02d}")165166        return response.content167168169class TextToAccentPhrasesAPI(TextToAudioQueryAPI):170    def _request(self, client, text, speaker_id, is_kana=False, **kwargs) -> Response:171        # OpenAPI boolean should be lowercase keyword172        flag_kana = "true" if is_kana else "false"173174        return client.post(175            f"/{self._api_name}",176            params={"text": text, "speaker": speaker_id, "is_kana": flag_kana},177        )178179180class AccentPhraseEditAPI(EndPoint):181    def _request(182        self, client, accent_phrases: List[Dict[str, Any]], speaker_id: int, **kwargs183    ) -> Response:184        return client.post(185            f"/{self._api_name}",186            json=accent_phrases,187            params={"speaker": speaker_id},188            headers={"Content-Type": "application/json"},189        )190191    def _set_content(192        self, response: Response, speaker_id: int, tag: str = "dump", **kwargs193    ) -> Any:194        if self._dump_dir is not None:195            if self._dumper is None:196                self._dumper = TaggedDumper(197                    self._dump_dir / self._api_name, "json", is_indexed=True198                )199            self._dumper.dump(response.text, tag + f"_s{speaker_id:02d}")200201        try:202            json_response = response.json()203        except JSONDecodeError:204            json_response = {}205        return json_response206207208class BatchSynthesisAPI(EndPoint):209    def _request(210        self, client, audio_queries: List[Dict[str, Any]], speaker_id: int, **kwargs211    ) -> Response:212        return client.post(213            f"/{self._api_name}",214            json=audio_queries,215            params={"speaker": speaker_id},216            headers={"Content-Type": "application/json"},217        )218219    def _set_content(220        self, response: Response, speaker_id: int, tag: str = "dump", **kwargs221    ) -> Any:222        if self._dump_dir is not None:223            if self._dumper is None:224                self._dumper = TaggedDumper(225                    self._dump_dir / self._api_name, "zip", is_indexed=True226                )227            self._dumper.dump(response.content, tag + f"_s{speaker_id:02d}")228229        return response.content230231232class ConcatWavesAPI(EndPoint):233    def _request(self, client, base64_waves: List[str], **kwargs) -> Response:234        return client.post(235            f"/{self._api_name}",236            json=base64_waves,237            headers={"Content-Type": "application/json"},238        )239240    def _set_content(self, response: Response, tag: str = "dump", **kwargs) -> Any:241        if self._dump_dir is not None:242            if self._dumper is None:243                self._dumper = TaggedDumper(244                    self._dump_dir / self._api_name, "wav", is_indexed=True245                )246            self._dumper.dump(response.content, tag)247248        return response.content249250251class TextToAudioQueryWithPresetAPI(EndPoint):252    def _request(self, client, text, preset_id, **kwargs) -> Response:253        return client.post(254            f"/{self._api_name}",255            params={"text": text, "preset_id": preset_id},256        )257258    def _put_log(259        self, response_time: float, response: Response, text: str, **kwargs260    ) -> None:261        print(262            (263                f"{self._api_name:>18}: {response_time:7.3f} [sec]"264                f" : {len(text):3d} : {text}"265            ),266            file=sys.stderr,267        )268269    def _set_content(270        self, response: Response, preset_id: int, tag: str = "dump", **kwargs271    ) -> Any:272        if self._dump_dir is not None:273            if self._dumper is None:274                self._dumper = TaggedDumper(275                    self._dump_dir / self._api_name, "json", is_indexed=True276                )277            self._dumper.dump(response.text, tag + f"_p{preset_id:02d}")278279        try:280            json_response = response.json()281        except JSONDecodeError:282            json_response = {}283        return json_response284285286class VveService:287    def __init__(self, client: VveClient, dump_root_dir: Optional[Path] = None) -> None:288        self.__client = client289        self.__dump_root_dir = dump_root_dir290291        self.__apis: Dict[str, EndPoint] = {}292293    def _get_api(self, endpoint_type, api_name):294        if not api_name in self.__apis:295            self.__apis[api_name] = endpoint_type(api_name, self.__dump_root_dir)296        return self.__apis[api_name]297298    def version(self) -> str:299        api = self._get_api(InformationQueryAPI, "version")300        return api.run(self.__client)301302    def speakers(self) -> Dict[str, Any]:303        api = self._get_api(InformationQueryAPI, "speakers")304        return api.run(self.__client)305306    def audio_query(307        self, text: str, speaker_id: int, tag: str = "dump"308    ) -> Dict[str, Any]:309        api = self._get_api(TextToAudioQueryAPI, "audio_query")310        return api.run(self.__client, text=text, speaker_id=speaker_id, tag=tag)311312    def synthesis(313        self, audio_query: Dict[str, Any], speaker_id: int, tag: str = "dump"314    ) -> bytes:315        api = self._get_api(SynthesisAPI, "synthesis")316        return api.run(317            self.__client, audio_query=audio_query, speaker_id=speaker_id, tag=tag318        )319320    def accent_phrases(321        self, text: str, speaker_id: int, is_kana: bool = False, tag: str = "dump"322    ) -> List[Dict[str, Any]]:323        api = self._get_api(TextToAccentPhrasesAPI, "accent_phrases")324        return api.run(325            self.__client, text=text, speaker_id=speaker_id, is_kana=is_kana, tag=tag326        )327328    def mora_data(329        self, accent_phrases: List[Dict[str, Any]], speaker_id: int, tag: str = "dump"330    ) -> List[Dict[str, Any]]:331        api = self._get_api(AccentPhraseEditAPI, "mora_data")332        return api.run(333            self.__client, accent_phrases=accent_phrases, speaker_id=speaker_id, tag=tag334        )335336    def mora_length(337        self, accent_phrases: List[Dict[str, Any]], speaker_id: int, tag: str = "dump"338    ) -> List[Dict[str, Any]]:339        api = self._get_api(AccentPhraseEditAPI, "mora_length")340        return api.run(341            self.__client, accent_phrases=accent_phrases, speaker_id=speaker_id, tag=tag342        )343344    def mora_pitch(345        self, accent_phrases: List[Dict[str, Any]], speaker_id: int, tag: str = "dump"346    ) -> List[Dict[str, Any]]:347        api = self._get_api(AccentPhraseEditAPI, "mora_pitch")348        return api.run(349            self.__client, accent_phrases=accent_phrases, speaker_id=speaker_id, tag=tag350        )351352    def multi_synthesis(353        self, audio_queries: List[Dict[str, Any]], speaker_id: int, tag: str = "dump"354    ) -> bytes:355        api = self._get_api(BatchSynthesisAPI, "multi_synthesis")356        return api.run(357            self.__client, audio_queries=audio_queries, speaker_id=speaker_id, tag=tag358        )359360    def connect_waves(self, base64_waves: List[str], tag: str = "dump") -> bytes:361        api = self._get_api(ConcatWavesAPI, "connect_waves")362        return api.run(self.__client, base64_waves=base64_waves, tag=tag)363364    def audio_query_from_preset(365        self, text: str, preset_id: int, tag: str = "dump"366    ) -> Dict[str, Any]:367        api = self._get_api(TextToAudioQueryWithPresetAPI, "audio_query_from_preset")368        return api.run(self.__client, text=text, preset_id=preset_id, tag=tag)369370    def presets(self) -> List[Dict[str, Any]]:371        api = self._get_api(InformationQueryAPI, "presets")
...

yaml.py

Source:yaml.py

1from rez.utils.data_utils import SourceCode2from rez.vendor import yaml3from rez.vendor.yaml.dumper import SafeDumper4from rez.vendor.yaml.nodes import ScalarNode, MappingNode5from rez.vendor.version.version import Version6from rez.vendor.version.requirement import Requirement7from types import FunctionType, BuiltinFunctionType8from inspect import getsourcelines9from textwrap import dedent10class _Dumper(SafeDumper):11    """Dumper which can serialise custom types such as Version, and keeps12    long strings nicely formatted in >/| block-style format.13    """14    # modified from yaml.representer.SafeRepresenter.represent_str()15    def represent_str(self, data):16        tag = None17        if '\n' in data:18            style = '|'19        elif len(data) > 80:20            style = '>'21        else:22            style = None23        try:24            data = unicode(data, 'ascii')25            tag = u'tag:yaml.org,2002:str'26        except UnicodeDecodeError:27            try:28                data = unicode(data, 'utf-8')29                tag = u'tag:yaml.org,2002:str'30            except UnicodeDecodeError:31                data = data.encode('base64')32                tag = u'tag:yaml.org,2002:binary'33                style = '|'34        return self.represent_scalar(tag, data, style=style)35    def represent_as_str(self, data):36        return self.represent_str(str(data))37    def represent_function(self, data):38        loc = getsourcelines(data)[0][1:]39        code = dedent(''.join(loc))40        return self.represent_str(code)41    def represent_builtin_function(self, data):42        return self.represent_str(str(data))43    def represent_sourcecode(self, data):44        code = data.source45        return self.represent_str(code)46_Dumper.add_representer(str, _Dumper.represent_str)47_Dumper.add_representer(Version, _Dumper.represent_as_str)48_Dumper.add_representer(Requirement, _Dumper.represent_as_str)49_Dumper.add_representer(FunctionType, _Dumper.represent_function)50_Dumper.add_representer(BuiltinFunctionType, _Dumper.represent_builtin_function)51_Dumper.add_representer(SourceCode, _Dumper.represent_sourcecode)52def dump_yaml(data, Dumper=_Dumper, default_flow_style=False):53    """Returns data as yaml-formatted string."""54    content = yaml.dump(data,55                        default_flow_style=default_flow_style,56                        Dumper=Dumper)57    return content.strip()58def load_yaml(filepath):59    """Convenience function for loading yaml-encoded data from disk."""60    with open(filepath) as f:61        txt = f.read()62    return yaml.load(txt)63# Copyright 2013-2016 Allan Johns.64#65# This library is free software: you can redistribute it and/or66# modify it under the terms of the GNU Lesser General Public67# License as published by the Free Software Foundation, either68# version 3 of the License, or (at your option) any later version.69#70# This library is distributed in the hope that it will be useful,71# but WITHOUT ANY WARRANTY; without even the implied warranty of72# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU73# Lesser General Public License for more details.74#75# You should have received a copy of the GNU Lesser General Public...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.