How to use _dumper method in Airtest

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")
...

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...