cengal.text_processing.text_translator.versions.v_0.text_translator

View Source

  1#!/usr/bin/env python
  2# coding=utf-8
  3
  4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>
  5# 
  6# Licensed under the Apache License, Version 2.0 (the "License");
  7# you may not use this file except in compliance with the License.
  8# You may obtain a copy of the License at
  9# 
 10#     http://www.apache.org/licenses/LICENSE-2.0
 11# 
 12# Unless required by applicable law or agreed to in writing, software
 13# distributed under the License is distributed on an "AS IS" BASIS,
 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 15# See the License for the specific language governing permissions and
 16# limitations under the License.
 17
 18
 19__all__ = ['TextTranslationDictionary', 'TextEntityId', 'TranslatableText', 'TranslationLanguageId', 'TextTranslatorError', 'TextTranslator', 'TranslationLanguageMapper', 'TranslationLanguageChooser', 'TextTranslationReapplier', 'CoroPriority', 'tt', 'TranslationWorker']
 20
 21
 22from collections.abc import Mapping
 23from typing import Hashable, Dict, Union, Optional, Callable
 24from cengal.data_manipulation.serialization import *
 25from cengal.code_flow_control.call_history_reapplier import *
 26from cengal.parallel_execution.coroutines.coro_scheduler import *
 27from cengal.parallel_execution.coroutines.coro_standard_services.async_event_bus import *
 28from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import *
 29from cengal.parallel_execution.coroutines.coro_standard_services.sleep import *
 30from uuid import uuid4
 31import sys
 32
 33
 34"""
 35Module Docstring
 36Docstrings: http://www.python.org/dev/peps/pep-0257/
 37"""
 38
 39__author__ = "ButenkoMS <gtalk@butenkoms.space>"
 40__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
 41__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
 42__license__ = "Apache License, Version 2.0"
 43__version__ = "4.4.1"
 44__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
 45__email__ = "gtalk@butenkoms.space"
 46# __status__ = "Prototype"
 47__status__ = "Development"
 48# __status__ = "Production"
 49
 50
 51TranslationLanguageId = Hashable
 52TextEntityId = Hashable
 53TranslatableText = str
 54TextTranslationDictionary = Dict
 55'''
 56In Python:
 57text_translation_dictionary = {
 58    '{TranslatableText}': {
 59        'default': {  // Can be empty. "Variants" field must not be empty in this case.
 60            '{TranslationLanguageId}': '{TranslatableText}',
 61            '{TranslationLanguageId}': '{TranslatableText}',
 62            ...
 63        },
 64        'variants': {  // Can be empty. "Default" field must not be empty in this case.
 65            '{TextEntityId}': {
 66                '{TranslationLanguageId}': '{TranslatableText}',
 67                '{TranslationLanguageId}': '{TranslatableText}',
 68                ...
 69            },
 70            '{TextEntityId}': {
 71                '{TranslationLanguageId}': '{TranslatableText}',
 72                '{TranslationLanguageId}': '{TranslatableText}',
 73                ...
 74            },
 75            ...
 76        }
 77    }
 78}
 79
 80In JSON:
 81{
 82    'type': 'Cengal.TextTranslationDictionary',
 83    'version': '1.0.0'
 84    'text_translation_list': [
 85        {
 86            'text': '{TranslatableText}',
 87            'translations': {
 88                'default': {  // Can be empty. "Variants" field must not be empty in this case.
 89                    '{TranslationLanguageId}': '{TranslatableText}',
 90                    '{TranslationLanguageId}': '{TranslatableText}',
 91                    ...
 92                },
 93                'variants': {  // Can be empty. "Default" field must not be empty in this case.
 94                    '{TextEntityId}': {
 95                        '{TranslationLanguageId}': '{TranslatableText}',
 96                        '{TranslationLanguageId}': '{TranslatableText}',
 97                        ...
 98                    },
 99                    '{TextEntityId}': {
100                        '{TranslationLanguageId}': '{TranslatableText}',
101                        '{TranslationLanguageId}': '{TranslatableText}',
102                        ...
103                    },
104                    ...
105                }
106            }
107        }
108    ]
109}
110'''
111TranslationLangToLangMap = Dict[TranslationLanguageId, TranslationLanguageId]
112TranslationWorker = Callable[['TranslatableText'], None]
113
114
115class TextTranslatorError(Exception):
116    pass
117
118class TextTranslator:
119    @classmethod
120    def from_json(cls, json_data: Union[bytes, str], encoding: Optional[str]=None):
121        serializer = best_serializer_for_standard_data((DataFormats.json,
122                                 Tags.decode_str_as_str,
123                                 Tags.decode_list_as_list,
124                                 Tags.superficial,
125                                 Tags.current_platform),
126                                TestDataType.deep_large,
127                                0.1)
128        encoding = encoding or 'utf-8'
129        if isinstance(json_data, bytes):
130            json_data = json_data.decode(encoding=encoding)
131        decoded_data: Dict = serializer.loads(json_data)
132        if not isinstance(decoded_data, Mapping):
133            raise TextTranslatorError('Wrong json data: root must be a dict')
134        if 'Cengal.TextTranslationDictionary' != decoded_data.get('type'):
135            raise TextTranslatorError('Wrong json data: lack of "type" field or a "type" field value mismatch')
136        try:
137            text_translation_dictionary: TextTranslationDictionary = dict()
138            text_translation_list = decoded_data['text_translation_list']
139            for item in text_translation_list:
140                text_translation_dictionary[item['text']] = item['translations']
141            return cls(text_translation_dictionary, decoded_data)
142        except (KeyError, TypeError):
143            raise TextTranslatorError('Wrong json data or other parsing error')
144    
145    def __init__(self, dictionary: TextTranslationDictionary, decoded_data: Optional[Dict]=None):
146        self.dictionary = dictionary
147        self.decoded_data = decoded_data
148    
149    def __call__(self, language: TranslationLanguageId, text: TranslatableText, entity_id: Optional[TextEntityId]=None) -> TranslatableText:
150        try:
151            translations = self.dictionary[text]
152            if entity_id is None:
153                variant = translations['default']
154            else:
155                variant = translations['variants'][entity_id]
156            return variant[language]
157        except KeyError:
158            return text
159
160
161class TranslationLanguageMapper:
162    def __init__(self, lang_2_lang: TranslationLangToLangMap, default_lang: TranslationLanguageId):
163        self.lang_2_lang: TranslationLangToLangMap = lang_2_lang
164        self.default_lang: TranslationLanguageId = default_lang
165    
166    def __call__(self, lang: TranslationLanguageId):
167        return self.lang_2_lang.get(lang, None) or self.default_lang
168
169
170class TranslationLanguageChooser:
171    def __init__(self, text_translator: TextTranslator, 
172                 translation_language_mapper: TranslationLanguageMapper,
173                 coro_scheduler: Optional[CoroSchedulerType]=None):
174        self._end_lang: Optional[TranslationLanguageId] = None
175        self._lang: Optional[TranslationLanguageId] = None
176        self.text_translator: TextTranslator = text_translator
177        self.translation_language_mapper: TranslationLanguageMapper = translation_language_mapper
178        self.coro_scheduler: CoroSchedulerType = coro_scheduler or CoroScheduler.current_loop()
179        self.translation_language_changed_event: str = str(uuid4())
180    
181    @property
182    def lang(self) -> TranslationLanguageId:
183        return self._lang
184    
185    @lang.setter
186    def lang(self, language: TranslationLanguageId):
187        self._lang = language
188        self._end_lang = self.translation_language_mapper(self._lang)
189        
190        # def raise_translation_language_changed_event(
191        #     interface: Interface,
192        #     translation_language_changed_event: str,
193        #     lang: TranslationLanguageId,
194        #     end_lang: TranslationLanguageId
195        #     ):
196        #     with log_uncatched_exception():
197        #         print('raise_translation_language_changed_event - raising event...')
198        #         print(interface, translation_language_changed_event, lang, end_lang)
199        #         interface(Sleep, 1.0)
200        #         interface(AsyncEventBus, AsyncEventBusRequest().send_event(translation_language_changed_event, (lang, end_lang), CoroPriority.low))
201        #         interface(Sleep, 1.0)
202        #         print('raise_translation_language_changed_event - done')
203        
204        # try_put_coro_to(get_interface_and_loop_with_explicit_loop(self.coro_scheduler), raise_translation_language_changed_event, 
205        #         self.translation_language_changed_event, self._lang, self._end_lang)
206        try_send_async_event(self.coro_scheduler, self.translation_language_changed_event, (self._lang, self._end_lang), CoroPriority.high)
207    
208    # @staticmethod
209    # def raise_translation_language_changed_event(
210    #     interface: Interface,
211    #     translation_language_changed_event: str,
212    #     lang: TranslationLanguageId,
213    #     end_lang: TranslationLanguageId
214    #     ):
215    #     print('raise_translation_language_changed_event - raising event...')
216    #     print(interface, translation_language_changed_event, lang, end_lang)
217    #     interface(AsyncEventBus, AsyncEventBusRequest().send_event(translation_language_changed_event, (lang, end_lang), CoroPriority.low))
218    #     print('raise_translation_language_changed_event - done')
219    
220    def set_lang(self, language: TranslationLanguageId) -> 'TranslationLanguageChooser':
221        '''
222        For usage with ArgsManager like
223                am = ArgsManager(
224                    EArgs(text_translator=TranslationLanguageChooser(
225                        TextTranslator.from_json(TEXT_DICTIONARY), 
226                        TranslationLanguageMapper(TRANSLATION_LANGUAGE_MAP, 'en')).set_lang('ru'))
227                )
228
229        '''
230        self.lang = language
231        return self
232    
233    @property
234    def end_lang(self) -> TranslationLanguageId:
235        return self._end_lang
236    
237    @end_lang.setter
238    def end_lang(self, language: TranslationLanguageId):
239        pass
240    
241    def __call__(self, text: TranslatableText, entity_id: Optional[TextEntityId]=None) -> TranslatableText:
242        return self.text_translator(self._end_lang, text, entity_id)
243
244
245class TextTranslationReapplier(CallHistoryReapplier):
246    class ToBeTranslated:
247        def __init__(self):
248            self.text: TranslatableText = None
249        
250        def __call__(self, text: TranslatableText) -> 'TextTranslationReapplier.ToBeTranslated':
251            self.text = text
252            return self
253        
254    def __init__(self, text_translator: TranslationLanguageChooser, priority: CoroPriority=CoroPriority.low):
255        self.text_translator = text_translator
256        super().__init__(priority)
257    
258    def _translate_needed(self, value: Any, entity_id: Optional[TextEntityId]) -> Any:
259        if isinstance(value, TextTranslationReapplier.ToBeTranslated):
260            return self.text_translator(value.text, entity_id)
261        else:
262            return value
263    
264    def call_impl(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: TranslationWorker, text_template: Optional[str], *args, **kwargs):
265        new_args = list()
266        for arg in args:
267            new_args.append(self._translate_needed(arg, entity_id))
268        
269        new_kwargs = dict()
270        for key, value in kwargs.items():
271            new_kwargs[key] = self._translate_needed(value, entity_id)
272        
273        if text_template is None:
274            if new_kwargs:
275                raise RuntimeError('There are tt items in kwargs, however text_template is None')
276
277            translated_text = ' '.join(new_args)
278        else:
279            translated_text = text_template.format(*new_args, **new_kwargs)
280        
281        translation_worker(translated_text)
282    
283    def args_to_key_value(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: Callable, text_template: Optional[str], *args, **kwargs) -> Tuple[Hashable, Any]:
284        return ((entity_id, id(obj), field), (translation_worker, text_template, args, kwargs))
285    
286    def key_value_to_args(self, key: Hashable, value: Any) -> Tuple[Tuple, Dict]:
287        entity_id, obj, field = key
288        translation_worker, text_template, args, kwargs = value
289        new_args = tuple([entity_id, obj, field, translation_worker, text_template] + list(args))
290        return new_args, kwargs
291
292
293def tt(text: TranslatableText) -> 'TextTranslationReapplier.ToBeTranslated':
294    return TextTranslationReapplier.ToBeTranslated()(text)

TextTranslationDictionary = typing.Dict

In Python: text_translation_dictionary = { '{TranslatableText}': { 'default': { // Can be empty. "Variants" field must not be empty in this case. '{TranslationLanguageId}': '{TranslatableText}', '{TranslationLanguageId}': '{TranslatableText}', ... }, 'variants': { // Can be empty. "Default" field must not be empty in this case. '{TextEntityId}': { '{TranslationLanguageId}': '{TranslatableText}', '{TranslationLanguageId}': '{TranslatableText}', ... }, '{TextEntityId}': { '{TranslationLanguageId}': '{TranslatableText}', '{TranslationLanguageId}': '{TranslatableText}', ... }, ... } } }

In JSON: { 'type': 'Cengal.TextTranslationDictionary', 'version': '1.0.0' 'text_translation_list': [ { 'text': '{TranslatableText}', 'translations': { 'default': { // Can be empty. "Variants" field must not be empty in this case. '{TranslationLanguageId}': '{TranslatableText}', '{TranslationLanguageId}': '{TranslatableText}', ... }, 'variants': { // Can be empty. "Default" field must not be empty in this case. '{TextEntityId}': { '{TranslationLanguageId}': '{TranslatableText}', '{TranslationLanguageId}': '{TranslatableText}', ... }, '{TextEntityId}': { '{TranslationLanguageId}': '{TranslatableText}', '{TranslationLanguageId}': '{TranslatableText}', ... }, ... } } } ] }

TextEntityId = typing.Hashable

TranslatableText = <class 'str'>

TranslationLanguageId = typing.Hashable

class TextTranslatorError(builtins.Exception): View Source

116class TextTranslatorError(Exception):
117    pass

Common base class for all non-exit exceptions.

Inherited Members

builtins.Exception

Exception

builtins.BaseException

with_traceback

args

class TextTranslator: View Source

119class TextTranslator:
120    @classmethod
121    def from_json(cls, json_data: Union[bytes, str], encoding: Optional[str]=None):
122        serializer = best_serializer_for_standard_data((DataFormats.json,
123                                 Tags.decode_str_as_str,
124                                 Tags.decode_list_as_list,
125                                 Tags.superficial,
126                                 Tags.current_platform),
127                                TestDataType.deep_large,
128                                0.1)
129        encoding = encoding or 'utf-8'
130        if isinstance(json_data, bytes):
131            json_data = json_data.decode(encoding=encoding)
132        decoded_data: Dict = serializer.loads(json_data)
133        if not isinstance(decoded_data, Mapping):
134            raise TextTranslatorError('Wrong json data: root must be a dict')
135        if 'Cengal.TextTranslationDictionary' != decoded_data.get('type'):
136            raise TextTranslatorError('Wrong json data: lack of "type" field or a "type" field value mismatch')
137        try:
138            text_translation_dictionary: TextTranslationDictionary = dict()
139            text_translation_list = decoded_data['text_translation_list']
140            for item in text_translation_list:
141                text_translation_dictionary[item['text']] = item['translations']
142            return cls(text_translation_dictionary, decoded_data)
143        except (KeyError, TypeError):
144            raise TextTranslatorError('Wrong json data or other parsing error')
145    
146    def __init__(self, dictionary: TextTranslationDictionary, decoded_data: Optional[Dict]=None):
147        self.dictionary = dictionary
148        self.decoded_data = decoded_data
149    
150    def __call__(self, language: TranslationLanguageId, text: TranslatableText, entity_id: Optional[TextEntityId]=None) -> TranslatableText:
151        try:
152            translations = self.dictionary[text]
153            if entity_id is None:
154                variant = translations['default']
155            else:
156                variant = translations['variants'][entity_id]
157            return variant[language]
158        except KeyError:
159            return text

TextTranslator( dictionary: typing.Dict, decoded_data: typing.Union[typing.Dict, NoneType] = None) View Source

146    def __init__(self, dictionary: TextTranslationDictionary, decoded_data: Optional[Dict]=None):
147        self.dictionary = dictionary
148        self.decoded_data = decoded_data

@classmethod

def from_json( cls, json_data: typing.Union[bytes, str], encoding: typing.Union[str, NoneType] = None): View Source

120    @classmethod
121    def from_json(cls, json_data: Union[bytes, str], encoding: Optional[str]=None):
122        serializer = best_serializer_for_standard_data((DataFormats.json,
123                                 Tags.decode_str_as_str,
124                                 Tags.decode_list_as_list,
125                                 Tags.superficial,
126                                 Tags.current_platform),
127                                TestDataType.deep_large,
128                                0.1)
129        encoding = encoding or 'utf-8'
130        if isinstance(json_data, bytes):
131            json_data = json_data.decode(encoding=encoding)
132        decoded_data: Dict = serializer.loads(json_data)
133        if not isinstance(decoded_data, Mapping):
134            raise TextTranslatorError('Wrong json data: root must be a dict')
135        if 'Cengal.TextTranslationDictionary' != decoded_data.get('type'):
136            raise TextTranslatorError('Wrong json data: lack of "type" field or a "type" field value mismatch')
137        try:
138            text_translation_dictionary: TextTranslationDictionary = dict()
139            text_translation_list = decoded_data['text_translation_list']
140            for item in text_translation_list:
141                text_translation_dictionary[item['text']] = item['translations']
142            return cls(text_translation_dictionary, decoded_data)
143        except (KeyError, TypeError):
144            raise TextTranslatorError('Wrong json data or other parsing error')

dictionary

decoded_data

class TranslationLanguageMapper: View Source

162class TranslationLanguageMapper:
163    def __init__(self, lang_2_lang: TranslationLangToLangMap, default_lang: TranslationLanguageId):
164        self.lang_2_lang: TranslationLangToLangMap = lang_2_lang
165        self.default_lang: TranslationLanguageId = default_lang
166    
167    def __call__(self, lang: TranslationLanguageId):
168        return self.lang_2_lang.get(lang, None) or self.default_lang

TranslationLanguageMapper( lang_2_lang: typing.Dict[typing.Hashable, typing.Hashable], default_lang: typing.Hashable) View Source

163    def __init__(self, lang_2_lang: TranslationLangToLangMap, default_lang: TranslationLanguageId):
164        self.lang_2_lang: TranslationLangToLangMap = lang_2_lang
165        self.default_lang: TranslationLanguageId = default_lang

lang_2_lang: Dict[Hashable, Hashable]

default_lang: Hashable

class TranslationLanguageChooser: View Source

171class TranslationLanguageChooser:
172    def __init__(self, text_translator: TextTranslator, 
173                 translation_language_mapper: TranslationLanguageMapper,
174                 coro_scheduler: Optional[CoroSchedulerType]=None):
175        self._end_lang: Optional[TranslationLanguageId] = None
176        self._lang: Optional[TranslationLanguageId] = None
177        self.text_translator: TextTranslator = text_translator
178        self.translation_language_mapper: TranslationLanguageMapper = translation_language_mapper
179        self.coro_scheduler: CoroSchedulerType = coro_scheduler or CoroScheduler.current_loop()
180        self.translation_language_changed_event: str = str(uuid4())
181    
182    @property
183    def lang(self) -> TranslationLanguageId:
184        return self._lang
185    
186    @lang.setter
187    def lang(self, language: TranslationLanguageId):
188        self._lang = language
189        self._end_lang = self.translation_language_mapper(self._lang)
190        
191        # def raise_translation_language_changed_event(
192        #     interface: Interface,
193        #     translation_language_changed_event: str,
194        #     lang: TranslationLanguageId,
195        #     end_lang: TranslationLanguageId
196        #     ):
197        #     with log_uncatched_exception():
198        #         print('raise_translation_language_changed_event - raising event...')
199        #         print(interface, translation_language_changed_event, lang, end_lang)
200        #         interface(Sleep, 1.0)
201        #         interface(AsyncEventBus, AsyncEventBusRequest().send_event(translation_language_changed_event, (lang, end_lang), CoroPriority.low))
202        #         interface(Sleep, 1.0)
203        #         print('raise_translation_language_changed_event - done')
204        
205        # try_put_coro_to(get_interface_and_loop_with_explicit_loop(self.coro_scheduler), raise_translation_language_changed_event, 
206        #         self.translation_language_changed_event, self._lang, self._end_lang)
207        try_send_async_event(self.coro_scheduler, self.translation_language_changed_event, (self._lang, self._end_lang), CoroPriority.high)
208    
209    # @staticmethod
210    # def raise_translation_language_changed_event(
211    #     interface: Interface,
212    #     translation_language_changed_event: str,
213    #     lang: TranslationLanguageId,
214    #     end_lang: TranslationLanguageId
215    #     ):
216    #     print('raise_translation_language_changed_event - raising event...')
217    #     print(interface, translation_language_changed_event, lang, end_lang)
218    #     interface(AsyncEventBus, AsyncEventBusRequest().send_event(translation_language_changed_event, (lang, end_lang), CoroPriority.low))
219    #     print('raise_translation_language_changed_event - done')
220    
221    def set_lang(self, language: TranslationLanguageId) -> 'TranslationLanguageChooser':
222        '''
223        For usage with ArgsManager like
224                am = ArgsManager(
225                    EArgs(text_translator=TranslationLanguageChooser(
226                        TextTranslator.from_json(TEXT_DICTIONARY), 
227                        TranslationLanguageMapper(TRANSLATION_LANGUAGE_MAP, 'en')).set_lang('ru'))
228                )
229
230        '''
231        self.lang = language
232        return self
233    
234    @property
235    def end_lang(self) -> TranslationLanguageId:
236        return self._end_lang
237    
238    @end_lang.setter
239    def end_lang(self, language: TranslationLanguageId):
240        pass
241    
242    def __call__(self, text: TranslatableText, entity_id: Optional[TextEntityId]=None) -> TranslatableText:
243        return self.text_translator(self._end_lang, text, entity_id)

TranslationLanguageChooser( text_translator: TextTranslator, translation_language_mapper: TranslationLanguageMapper, coro_scheduler: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType] = None) View Source

172    def __init__(self, text_translator: TextTranslator, 
173                 translation_language_mapper: TranslationLanguageMapper,
174                 coro_scheduler: Optional[CoroSchedulerType]=None):
175        self._end_lang: Optional[TranslationLanguageId] = None
176        self._lang: Optional[TranslationLanguageId] = None
177        self.text_translator: TextTranslator = text_translator
178        self.translation_language_mapper: TranslationLanguageMapper = translation_language_mapper
179        self.coro_scheduler: CoroSchedulerType = coro_scheduler or CoroScheduler.current_loop()
180        self.translation_language_changed_event: str = str(uuid4())

text_translator: TextTranslator

translation_language_mapper: TranslationLanguageMapper

coro_scheduler: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable]

translation_language_changed_event: str

lang: Hashable View Source

182    @property
183    def lang(self) -> TranslationLanguageId:
184        return self._lang

def set_lang( self, language: typing.Hashable) -> TranslationLanguageChooser: View Source

221    def set_lang(self, language: TranslationLanguageId) -> 'TranslationLanguageChooser':
222        '''
223        For usage with ArgsManager like
224                am = ArgsManager(
225                    EArgs(text_translator=TranslationLanguageChooser(
226                        TextTranslator.from_json(TEXT_DICTIONARY), 
227                        TranslationLanguageMapper(TRANSLATION_LANGUAGE_MAP, 'en')).set_lang('ru'))
228                )
229
230        '''
231        self.lang = language
232        return self

For usage with ArgsManager like am = ArgsManager( EArgs(text_translator=TranslationLanguageChooser( TextTranslator.from_json(TEXT_DICTIONARY), TranslationLanguageMapper(TRANSLATION_LANGUAGE_MAP, 'en')).set_lang('ru')) )

end_lang: Hashable View Source

234    @property
235    def end_lang(self) -> TranslationLanguageId:
236        return self._end_lang

class TextTranslationReapplier(cengal.code_flow_control.call_history_reapplier.versions.v_0.call_history_reapplier.CallHistoryReapplier): View Source

246class TextTranslationReapplier(CallHistoryReapplier):
247    class ToBeTranslated:
248        def __init__(self):
249            self.text: TranslatableText = None
250        
251        def __call__(self, text: TranslatableText) -> 'TextTranslationReapplier.ToBeTranslated':
252            self.text = text
253            return self
254        
255    def __init__(self, text_translator: TranslationLanguageChooser, priority: CoroPriority=CoroPriority.low):
256        self.text_translator = text_translator
257        super().__init__(priority)
258    
259    def _translate_needed(self, value: Any, entity_id: Optional[TextEntityId]) -> Any:
260        if isinstance(value, TextTranslationReapplier.ToBeTranslated):
261            return self.text_translator(value.text, entity_id)
262        else:
263            return value
264    
265    def call_impl(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: TranslationWorker, text_template: Optional[str], *args, **kwargs):
266        new_args = list()
267        for arg in args:
268            new_args.append(self._translate_needed(arg, entity_id))
269        
270        new_kwargs = dict()
271        for key, value in kwargs.items():
272            new_kwargs[key] = self._translate_needed(value, entity_id)
273        
274        if text_template is None:
275            if new_kwargs:
276                raise RuntimeError('There are tt items in kwargs, however text_template is None')
277
278            translated_text = ' '.join(new_args)
279        else:
280            translated_text = text_template.format(*new_args, **new_kwargs)
281        
282        translation_worker(translated_text)
283    
284    def args_to_key_value(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: Callable, text_template: Optional[str], *args, **kwargs) -> Tuple[Hashable, Any]:
285        return ((entity_id, id(obj), field), (translation_worker, text_template, args, kwargs))
286    
287    def key_value_to_args(self, key: Hashable, value: Any) -> Tuple[Tuple, Dict]:
288        entity_id, obj, field = key
289        translation_worker, text_template, args, kwargs = value
290        new_args = tuple([entity_id, obj, field, translation_worker, text_template] + list(args))
291        return new_args, kwargs

TextTranslationReapplier( text_translator: TranslationLanguageChooser, priority: CoroPriority = <CoroPriority.low: 2>) View Source

255    def __init__(self, text_translator: TranslationLanguageChooser, priority: CoroPriority=CoroPriority.low):
256        self.text_translator = text_translator
257        super().__init__(priority)

text_translator

def call_impl( self, entity_id: typing.Union[typing.Hashable, NoneType], obj: typing.Any, field: typing.Hashable, translation_worker: collections.abc.Callable[str, NoneType], text_template: typing.Union[str, NoneType], *args, **kwargs): View Source

265    def call_impl(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: TranslationWorker, text_template: Optional[str], *args, **kwargs):
266        new_args = list()
267        for arg in args:
268            new_args.append(self._translate_needed(arg, entity_id))
269        
270        new_kwargs = dict()
271        for key, value in kwargs.items():
272            new_kwargs[key] = self._translate_needed(value, entity_id)
273        
274        if text_template is None:
275            if new_kwargs:
276                raise RuntimeError('There are tt items in kwargs, however text_template is None')
277
278            translated_text = ' '.join(new_args)
279        else:
280            translated_text = text_template.format(*new_args, **new_kwargs)
281        
282        translation_worker(translated_text)

def args_to_key_value( self, entity_id: typing.Union[typing.Hashable, NoneType], obj: typing.Any, field: typing.Hashable, translation_worker: typing.Callable, text_template: typing.Union[str, NoneType], *args, **kwargs) -> Tuple[Hashable, Any]: View Source

284    def args_to_key_value(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: Callable, text_template: Optional[str], *args, **kwargs) -> Tuple[Hashable, Any]:
285        return ((entity_id, id(obj), field), (translation_worker, text_template, args, kwargs))

def key_value_to_args(self, key: typing.Hashable, value: typing.Any) -> Tuple[Tuple, Dict]: View Source

287    def key_value_to_args(self, key: Hashable, value: Any) -> Tuple[Tuple, Dict]:
288        entity_id, obj, field = key
289        translation_worker, text_template, args, kwargs = value
290        new_args = tuple([entity_id, obj, field, translation_worker, text_template] + list(args))
291        return new_args, kwargs

Inherited Members

cengal.code_flow_control.call_history_reapplier.versions.v_0.call_history_reapplier.CallHistoryReapplier

history

priority

reapply

destroy

class TextTranslationReapplier.ToBeTranslated: View Source

247    class ToBeTranslated:
248        def __init__(self):
249            self.text: TranslatableText = None
250        
251        def __call__(self, text: TranslatableText) -> 'TextTranslationReapplier.ToBeTranslated':
252            self.text = text
253            return self

text: str

class CoroPriority(enum.Enum): View Source

59class CoroPriority(Enum):
60    high = 0
61    normal = 1
62    low = 2

An enumeration.

high = <CoroPriority.high: 0>

normal = <CoroPriority.normal: 1>

low = <CoroPriority.low: 2>

Inherited Members

enum.Enum

name

value

def tt( text: str) -> TextTranslationReapplier.ToBeTranslated: View Source

294def tt(text: TranslatableText) -> 'TextTranslationReapplier.ToBeTranslated':
295    return TextTranslationReapplier.ToBeTranslated()(text)

TranslationWorker = typing.Callable[[ForwardRef('TranslatableText')], NoneType]