cengal.text_processing.text_translator.versions.v_1.text_translator
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', 'TranslationLanguageId', 20 'TextTranslatorError', 'TextTranslator', 'TranslationLanguageMapper', 'TranslationLanguageChooser', 21 'TextTranslationReapplier', 'CoroPriority', 'TranslationWorker', 'TranslatableText', 'tt', 22 'TranslateMe', 'TMe', 'tme', 'TranslatableTextElement', 'TTE'] 23 24 25from collections.abc import Mapping 26from typing import Hashable, Dict, Union, Optional, Callable, List, Any, TypeVar, Generic 27from cengal.data_manipulation.serialization import * 28from cengal.code_flow_control.call_history_reapplier import * 29from cengal.parallel_execution.coroutines.coro_scheduler import * 30from cengal.parallel_execution.coroutines.coro_standard_services.async_event_bus import * 31from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import * 32from cengal.parallel_execution.coroutines.coro_standard_services.sleep import * 33from cengal.parallel_execution.coroutines.coro_standard_services.wait_coro import WaitCoroRequest, CoroutineNotFoundError 34from cengal.introspection.inspect import is_callable 35from uuid import uuid4 36import sys 37 38 39""" 40Module Docstring 41Docstrings: http://www.python.org/dev/peps/pep-0257/ 42""" 43 44__author__ = "ButenkoMS <gtalk@butenkoms.space>" 45__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 46__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 47__license__ = "Apache License, Version 2.0" 48__version__ = "4.4.1" 49__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 50__email__ = "gtalk@butenkoms.space" 51# __status__ = "Prototype" 52__status__ = "Development" 53# __status__ = "Production" 54 55 56TranslationLanguageId = Hashable 57TextEntityId = Hashable 58TextTranslationDictionary = Dict 59''' 60In Python: 61text_translation_dictionary = { 62 '{str}': { 63 'default': { // Can be empty. "Variants" field must not be empty in this case. 64 '{TranslationLanguageId}': '{str}', 65 '{TranslationLanguageId}': '{str}', 66 ... 67 }, 68 'variants': { // Can be empty. "Default" field must not be empty in this case. 69 '{TextEntityId}': { 70 '{TranslationLanguageId}': '{str}', 71 '{TranslationLanguageId}': '{str}', 72 ... 73 }, 74 '{TextEntityId}': { 75 '{TranslationLanguageId}': '{str}', 76 '{TranslationLanguageId}': '{str}', 77 ... 78 }, 79 ... 80 } 81 } 82} 83 84In JSON: 85{ 86 'type': 'Cengal.TextTranslationDictionary', 87 'version': '1.0.0' 88 'text_translation_list': [ 89 { 90 'text': '{str}', 91 'translations': { 92 'default': { // Can be empty. "Variants" field must not be empty in this case. 93 '{TranslationLanguageId}': '{str}', 94 '{TranslationLanguageId}': '{str}', 95 ... 96 }, 97 'variants': { // Can be empty. "Default" field must not be empty in this case. 98 '{TextEntityId}': { 99 '{TranslationLanguageId}': '{str}', 100 '{TranslationLanguageId}': '{str}', 101 ... 102 }, 103 '{TextEntityId}': { 104 '{TranslationLanguageId}': '{str}', 105 '{TranslationLanguageId}': '{str}', 106 ... 107 }, 108 ... 109 } 110 } 111 } 112 ] 113} 114''' 115TranslationLangToLangMap = Dict[TranslationLanguageId, TranslationLanguageId] 116TranslationWorker = Callable[[str], None] 117 118 119class TextTranslatorError(Exception): 120 pass 121 122class TextTranslator: 123 @classmethod 124 def from_json(cls, json_data: Union[bytes, str], encoding: Optional[str]=None): 125 serializer = best_serializer_for_standard_data((DataFormats.json, 126 Tags.decode_str_as_str, 127 Tags.decode_list_as_list, 128 Tags.superficial, 129 Tags.current_platform), 130 TestDataType.deep_large, 131 0.1) 132 encoding = encoding or 'utf-8' 133 if isinstance(json_data, bytes): 134 json_data = json_data.decode(encoding=encoding) 135 decoded_data: Dict = serializer.loads(json_data) 136 if not isinstance(decoded_data, Mapping): 137 raise TextTranslatorError('Wrong json data: root must be a dict') 138 if 'Cengal.TextTranslationDictionary' != decoded_data.get('type'): 139 raise TextTranslatorError('Wrong json data: lack of "type" field or a "type" field value mismatch') 140 try: 141 text_translation_dictionary: TextTranslationDictionary = dict() 142 text_translation_list = decoded_data['text_translation_list'] 143 for item in text_translation_list: 144 text_translation_dictionary[item['text']] = item['translations'] 145 return cls(text_translation_dictionary, decoded_data) 146 except (KeyError, TypeError): 147 raise TextTranslatorError('Wrong json data or other parsing error') 148 149 def __init__(self, dictionary: TextTranslationDictionary, decoded_data: Optional[Dict]=None): 150 self.dictionary = dictionary 151 self.decoded_data = decoded_data 152 153 def __call__(self, language: TranslationLanguageId, text: str, entity_id: Optional[TextEntityId]=None) -> str: 154 try: 155 translations = self.dictionary[text] 156 if entity_id is None: 157 variant = translations['default'] 158 else: 159 variant = translations['variants'][entity_id] 160 return variant[language] 161 except KeyError: 162 return text 163 164 165class TranslationLanguageMapper: 166 def __init__(self, lang_2_lang: TranslationLangToLangMap, default_lang: TranslationLanguageId): 167 self.lang_2_lang: TranslationLangToLangMap = lang_2_lang 168 self.default_lang: TranslationLanguageId = default_lang 169 170 def __call__(self, lang: TranslationLanguageId): 171 return self.lang_2_lang.get(lang, None) or self.default_lang 172 173 174class TranslationLanguageChooser: 175 def __init__(self, text_translator: TextTranslator, 176 translation_language_mapper: TranslationLanguageMapper, 177 coro_scheduler: Optional[CoroSchedulerType]=None): 178 self._end_lang: Optional[TranslationLanguageId] = None 179 self._lang: Optional[TranslationLanguageId] = None 180 self.text_translator: TextTranslator = text_translator 181 self.translation_language_mapper: TranslationLanguageMapper = translation_language_mapper 182 self.coro_scheduler: CoroSchedulerType = coro_scheduler or CoroScheduler.current_loop() 183 self.translation_language_changed_event: str = str(uuid4()) 184 185 @property 186 def lang(self) -> TranslationLanguageId: 187 return self._lang 188 189 @lang.setter 190 def lang(self, language: TranslationLanguageId): 191 self._lang = language 192 self._end_lang = self.translation_language_mapper(self._lang) 193 194 # def raise_translation_language_changed_event( 195 # interface: Interface, 196 # translation_language_changed_event: str, 197 # lang: TranslationLanguageId, 198 # end_lang: TranslationLanguageId 199 # ): 200 # with log_uncatched_exception(): 201 # print('raise_translation_language_changed_event - raising event...') 202 # print(interface, translation_language_changed_event, lang, end_lang) 203 # interface(Sleep, 1.0) 204 # interface(AsyncEventBus, AsyncEventBusRequest().send_event(translation_language_changed_event, (lang, end_lang), CoroPriority.low)) 205 # interface(Sleep, 1.0) 206 # print('raise_translation_language_changed_event - done') 207 208 # try_put_coro_to(get_interface_and_loop_with_explicit_loop(self.coro_scheduler), raise_translation_language_changed_event, 209 # self.translation_language_changed_event, self._lang, self._end_lang) 210 try_send_async_event(self.coro_scheduler, self.translation_language_changed_event, (self._lang, self._end_lang), CoroPriority.high) 211 212 # @staticmethod 213 # def raise_translation_language_changed_event( 214 # interface: Interface, 215 # translation_language_changed_event: str, 216 # lang: TranslationLanguageId, 217 # end_lang: TranslationLanguageId 218 # ): 219 # print('raise_translation_language_changed_event - raising event...') 220 # print(interface, translation_language_changed_event, lang, end_lang) 221 # interface(AsyncEventBus, AsyncEventBusRequest().send_event(translation_language_changed_event, (lang, end_lang), CoroPriority.low)) 222 # print('raise_translation_language_changed_event - done') 223 224 def set_lang(self, language: TranslationLanguageId) -> 'TranslationLanguageChooser': 225 ''' 226 For usage with ArgsManager like 227 am = ArgsManager( 228 EArgs(text_translator=TranslationLanguageChooser( 229 TextTranslator.from_json(TEXT_DICTIONARY), 230 TranslationLanguageMapper(TRANSLATION_LANGUAGE_MAP, 'en')).set_lang('ru')) 231 ) 232 233 ''' 234 self.lang = language 235 return self 236 237 @property 238 def end_lang(self) -> TranslationLanguageId: 239 return self._end_lang 240 241 @end_lang.setter 242 def end_lang(self, language: TranslationLanguageId): 243 pass 244 245 def __call__(self, text: str, entity_id: Optional[TextEntityId]=None) -> str: 246 return self.text_translator(self._end_lang, text, entity_id) 247 248 249class TextTranslationReapplier(CallHistoryReapplier): 250 def __init__(self, text_translator: TranslationLanguageChooser, priority: CoroPriority=CoroPriority.low): 251 self.text_translator = text_translator 252 super().__init__(priority) 253 254 def _translate_needed(self, value: Any, entity_id: Optional[TextEntityId]) -> Any: 255 if isinstance(value, TranslatableText): 256 return self.text_translator(value.text, entity_id) 257 else: 258 return value 259 260 def call_impl(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: TranslationWorker, text_template: Optional[str], *args, **kwargs): 261 new_args = list() 262 for arg in args: 263 new_args.append(self._translate_needed(arg, entity_id)) 264 265 new_kwargs = dict() 266 for key, value in kwargs.items(): 267 new_kwargs[key] = self._translate_needed(value, entity_id) 268 269 if text_template is None: 270 if new_kwargs: 271 raise RuntimeError('There are tt items in kwargs, however text_template is None') 272 273 translated_text = ' '.join(new_args) 274 else: 275 translated_text = text_template.format(*new_args, **new_kwargs) 276 277 translation_worker(translated_text) 278 279 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]: 280 return ((entity_id, id(obj), field), (translation_worker, text_template, args, kwargs)) 281 282 def key_value_to_args(self, key: Hashable, value: Any) -> Tuple[Tuple, Dict]: 283 entity_id, obj, field = key 284 translation_worker, text_template, args, kwargs = value 285 new_args = tuple([entity_id, obj, field, translation_worker, text_template] + list(args)) 286 return new_args, kwargs 287 288 289class TranslatableText: 290 def __init__(self, text: Union[str, Callable], entity_id: Optional[TextEntityId] = None, formatter: Optional[Callable] = None) -> None: 291 self.text: str = text 292 self.entity_id: Optional[TextEntityId] = entity_id 293 self.formatter: Optional[Callable] = formatter 294 self.is_awaitable: bool = False 295 296 def __call__(self) -> Any: 297 if is_callable(self.text): 298 return self.text() 299 else: 300 return self.text 301 302 def format(self, text: str) -> str: 303 if self.formatter is None: 304 return text 305 else: 306 return self.formatter(text) 307 308 def __str__(self) -> str: 309 return self() 310 311 312tt = TranslatableText 313 314 315class TranslateMe: 316 def __init__(self, *args) -> None: 317 self.args: Tuple[Union[str, TranslatableText]] = args 318 self._contains_translatable_text: bool = any(isinstance(arg, TranslatableText) for arg in self.args) 319 self._tte: 'TranslatableTextElement' = None 320 321 def __bool__(self) -> bool: 322 return self._contains_translatable_text 323 324 def to_str(self, text_translator: TextTranslator, language: TranslationLanguageId) -> str: 325 return ''.join([arg.format(text_translator(language, arg(), arg.entity_id)) if isinstance(arg, TranslatableText) else arg for arg in self.args]) 326 327 def tte(self, tte: 'TranslatableTextElement') -> 'TranslateMe': 328 self._tte = tte 329 return self 330 331 def __str__(self) -> str: 332 return self._tte.translate_me(self) 333 334 335TMe = TranslateMe 336tme = TranslateMe 337 338 339T = TypeVar('T') 340 341 342class TranslatableTextElement(Generic[T]): 343 def __init__(self, text_translation_language_chooser: TranslationLanguageChooser) -> None: 344 self.text_translation_language_chooser: TranslationLanguageChooser = text_translation_language_chooser 345 self.text_translator: TextTranslator = text_translation_language_chooser.text_translator 346 self.elements_and_their_translatable_text: Dict[T, TranslatableText] = dict() 347 self.current_translation_reapplier_coros_ids: Set[CoroID] = set() 348 self.lang_changing_queue: List[Tuple[str, str]] = list() 349 350 def __call__(self, text_element: T) -> Any: 351 raise NotImplementedError() 352 353 def set_text(self, text_element: T, text: Union[TranslateMe, str]) -> T: 354 raise NotImplementedError() 355 356 def translate_me(self, translate_me: TranslateMe) -> str: 357 return translate_me.to_str(self.text_translator, self.text_translation_language_chooser.end_lang) 358 359 def start_translation_reapplier(self, i: Interface): 360 coro_id: CoroID = i(PutCoro, self.translation_reapplier) 361 self.current_translation_reapplier_coros_ids.add(coro_id) 362 363 async def translation_reapplier(self, i: Interface): 364 waiting_set: Set[CoroID] = self.current_translation_reapplier_coros_ids - {i.coro_id} 365 # await i(WaitCoroRequest().list(waiting_set)) # TODO: Not Implemented currently 366 for coro_id in waiting_set: 367 try: 368 await i(WaitCoroRequest().single(coro_id)) 369 except CoroutineNotFoundError: 370 pass 371 372 self.current_translation_reapplier_coros_ids = self.current_translation_reapplier_coros_ids - waiting_set 373 374 if not self.lang_changing_queue: 375 return 376 377 lang_changing_queue = self.lang_changing_queue 378 self.lang_changing_queue = type(self.lang_changing_queue)() 379 for lang, end_lang in lang_changing_queue: 380 for text_element, translate_me in self.elements_and_their_translatable_text.items(): 381 text_element.text = translate_me.to_str(self.text_translator, end_lang) 382 383 def register_on_lang_changed_handler(self): 384 i: Interface = current_interface() 385 i(AsyncEventBusRequest().register_handler(self.text_translation_language_chooser.translation_language_changed_event, self.on_lang_changed)) 386 387 async def aregister_on_lang_changed_handler(self): 388 i: Interface = current_interface() 389 await i(AsyncEventBusRequest().register_handler(self.text_translation_language_chooser.translation_language_changed_event, self.on_lang_changed)) 390 391 def remove_on_lang_changed_handler(self): 392 i: Interface = current_interface() 393 i(AsyncEventBusRequest().remove_handler(self.text_translation_language_chooser.translation_language_changed_event, self.on_lang_changed)) 394 395 async def aremove_on_lang_changed_handler(self): 396 i: Interface = current_interface() 397 await i(AsyncEventBusRequest().remove_handler(self.text_translation_language_chooser.translation_language_changed_event, self.on_lang_changed)) 398 399 def on_lang_changed(self, event: Hashable, data: Tuple[str, str]): 400 self.lang_changing_queue.append(data) 401 put_coro(self.start_translation_reapplier) 402 403 404TTE = TranslatableTextElement
TextTranslationDictionary =
typing.Dict
In Python: text_translation_dictionary = { '{str}': { 'default': { // Can be empty. "Variants" field must not be empty in this case. '{TranslationLanguageId}': '{str}', '{TranslationLanguageId}': '{str}', ... }, 'variants': { // Can be empty. "Default" field must not be empty in this case. '{TextEntityId}': { '{TranslationLanguageId}': '{str}', '{TranslationLanguageId}': '{str}', ... }, '{TextEntityId}': { '{TranslationLanguageId}': '{str}', '{TranslationLanguageId}': '{str}', ... }, ... } } }
In JSON: { 'type': 'Cengal.TextTranslationDictionary', 'version': '1.0.0' 'text_translation_list': [ { 'text': '{str}', 'translations': { 'default': { // Can be empty. "Variants" field must not be empty in this case. '{TranslationLanguageId}': '{str}', '{TranslationLanguageId}': '{str}', ... }, 'variants': { // Can be empty. "Default" field must not be empty in this case. '{TextEntityId}': { '{TranslationLanguageId}': '{str}', '{TranslationLanguageId}': '{str}', ... }, '{TextEntityId}': { '{TranslationLanguageId}': '{str}', '{TranslationLanguageId}': '{str}', ... }, ... } } } ] }
TextEntityId =
typing.Hashable
TranslationLanguageId =
typing.Hashable
class
TextTranslatorError(builtins.Exception):
Common base class for all non-exit exceptions.
Inherited Members
- builtins.Exception
- Exception
- builtins.BaseException
- with_traceback
- args
class
TextTranslator:
123class TextTranslator: 124 @classmethod 125 def from_json(cls, json_data: Union[bytes, str], encoding: Optional[str]=None): 126 serializer = best_serializer_for_standard_data((DataFormats.json, 127 Tags.decode_str_as_str, 128 Tags.decode_list_as_list, 129 Tags.superficial, 130 Tags.current_platform), 131 TestDataType.deep_large, 132 0.1) 133 encoding = encoding or 'utf-8' 134 if isinstance(json_data, bytes): 135 json_data = json_data.decode(encoding=encoding) 136 decoded_data: Dict = serializer.loads(json_data) 137 if not isinstance(decoded_data, Mapping): 138 raise TextTranslatorError('Wrong json data: root must be a dict') 139 if 'Cengal.TextTranslationDictionary' != decoded_data.get('type'): 140 raise TextTranslatorError('Wrong json data: lack of "type" field or a "type" field value mismatch') 141 try: 142 text_translation_dictionary: TextTranslationDictionary = dict() 143 text_translation_list = decoded_data['text_translation_list'] 144 for item in text_translation_list: 145 text_translation_dictionary[item['text']] = item['translations'] 146 return cls(text_translation_dictionary, decoded_data) 147 except (KeyError, TypeError): 148 raise TextTranslatorError('Wrong json data or other parsing error') 149 150 def __init__(self, dictionary: TextTranslationDictionary, decoded_data: Optional[Dict]=None): 151 self.dictionary = dictionary 152 self.decoded_data = decoded_data 153 154 def __call__(self, language: TranslationLanguageId, text: str, entity_id: Optional[TextEntityId]=None) -> str: 155 try: 156 translations = self.dictionary[text] 157 if entity_id is None: 158 variant = translations['default'] 159 else: 160 variant = translations['variants'][entity_id] 161 return variant[language] 162 except KeyError: 163 return text
@classmethod
def
from_json( cls, json_data: typing.Union[bytes, str], encoding: typing.Union[str, NoneType] = None):
124 @classmethod 125 def from_json(cls, json_data: Union[bytes, str], encoding: Optional[str]=None): 126 serializer = best_serializer_for_standard_data((DataFormats.json, 127 Tags.decode_str_as_str, 128 Tags.decode_list_as_list, 129 Tags.superficial, 130 Tags.current_platform), 131 TestDataType.deep_large, 132 0.1) 133 encoding = encoding or 'utf-8' 134 if isinstance(json_data, bytes): 135 json_data = json_data.decode(encoding=encoding) 136 decoded_data: Dict = serializer.loads(json_data) 137 if not isinstance(decoded_data, Mapping): 138 raise TextTranslatorError('Wrong json data: root must be a dict') 139 if 'Cengal.TextTranslationDictionary' != decoded_data.get('type'): 140 raise TextTranslatorError('Wrong json data: lack of "type" field or a "type" field value mismatch') 141 try: 142 text_translation_dictionary: TextTranslationDictionary = dict() 143 text_translation_list = decoded_data['text_translation_list'] 144 for item in text_translation_list: 145 text_translation_dictionary[item['text']] = item['translations'] 146 return cls(text_translation_dictionary, decoded_data) 147 except (KeyError, TypeError): 148 raise TextTranslatorError('Wrong json data or other parsing error')
class
TranslationLanguageMapper:
166class TranslationLanguageMapper: 167 def __init__(self, lang_2_lang: TranslationLangToLangMap, default_lang: TranslationLanguageId): 168 self.lang_2_lang: TranslationLangToLangMap = lang_2_lang 169 self.default_lang: TranslationLanguageId = default_lang 170 171 def __call__(self, lang: TranslationLanguageId): 172 return self.lang_2_lang.get(lang, None) or self.default_lang
class
TranslationLanguageChooser:
175class TranslationLanguageChooser: 176 def __init__(self, text_translator: TextTranslator, 177 translation_language_mapper: TranslationLanguageMapper, 178 coro_scheduler: Optional[CoroSchedulerType]=None): 179 self._end_lang: Optional[TranslationLanguageId] = None 180 self._lang: Optional[TranslationLanguageId] = None 181 self.text_translator: TextTranslator = text_translator 182 self.translation_language_mapper: TranslationLanguageMapper = translation_language_mapper 183 self.coro_scheduler: CoroSchedulerType = coro_scheduler or CoroScheduler.current_loop() 184 self.translation_language_changed_event: str = str(uuid4()) 185 186 @property 187 def lang(self) -> TranslationLanguageId: 188 return self._lang 189 190 @lang.setter 191 def lang(self, language: TranslationLanguageId): 192 self._lang = language 193 self._end_lang = self.translation_language_mapper(self._lang) 194 195 # def raise_translation_language_changed_event( 196 # interface: Interface, 197 # translation_language_changed_event: str, 198 # lang: TranslationLanguageId, 199 # end_lang: TranslationLanguageId 200 # ): 201 # with log_uncatched_exception(): 202 # print('raise_translation_language_changed_event - raising event...') 203 # print(interface, translation_language_changed_event, lang, end_lang) 204 # interface(Sleep, 1.0) 205 # interface(AsyncEventBus, AsyncEventBusRequest().send_event(translation_language_changed_event, (lang, end_lang), CoroPriority.low)) 206 # interface(Sleep, 1.0) 207 # print('raise_translation_language_changed_event - done') 208 209 # try_put_coro_to(get_interface_and_loop_with_explicit_loop(self.coro_scheduler), raise_translation_language_changed_event, 210 # self.translation_language_changed_event, self._lang, self._end_lang) 211 try_send_async_event(self.coro_scheduler, self.translation_language_changed_event, (self._lang, self._end_lang), CoroPriority.high) 212 213 # @staticmethod 214 # def raise_translation_language_changed_event( 215 # interface: Interface, 216 # translation_language_changed_event: str, 217 # lang: TranslationLanguageId, 218 # end_lang: TranslationLanguageId 219 # ): 220 # print('raise_translation_language_changed_event - raising event...') 221 # print(interface, translation_language_changed_event, lang, end_lang) 222 # interface(AsyncEventBus, AsyncEventBusRequest().send_event(translation_language_changed_event, (lang, end_lang), CoroPriority.low)) 223 # print('raise_translation_language_changed_event - done') 224 225 def set_lang(self, language: TranslationLanguageId) -> 'TranslationLanguageChooser': 226 ''' 227 For usage with ArgsManager like 228 am = ArgsManager( 229 EArgs(text_translator=TranslationLanguageChooser( 230 TextTranslator.from_json(TEXT_DICTIONARY), 231 TranslationLanguageMapper(TRANSLATION_LANGUAGE_MAP, 'en')).set_lang('ru')) 232 ) 233 234 ''' 235 self.lang = language 236 return self 237 238 @property 239 def end_lang(self) -> TranslationLanguageId: 240 return self._end_lang 241 242 @end_lang.setter 243 def end_lang(self, language: TranslationLanguageId): 244 pass 245 246 def __call__(self, text: str, entity_id: Optional[TextEntityId]=None) -> str: 247 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)
176 def __init__(self, text_translator: TextTranslator, 177 translation_language_mapper: TranslationLanguageMapper, 178 coro_scheduler: Optional[CoroSchedulerType]=None): 179 self._end_lang: Optional[TranslationLanguageId] = None 180 self._lang: Optional[TranslationLanguageId] = None 181 self.text_translator: TextTranslator = text_translator 182 self.translation_language_mapper: TranslationLanguageMapper = translation_language_mapper 183 self.coro_scheduler: CoroSchedulerType = coro_scheduler or CoroScheduler.current_loop() 184 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]
225 def set_lang(self, language: TranslationLanguageId) -> 'TranslationLanguageChooser': 226 ''' 227 For usage with ArgsManager like 228 am = ArgsManager( 229 EArgs(text_translator=TranslationLanguageChooser( 230 TextTranslator.from_json(TEXT_DICTIONARY), 231 TranslationLanguageMapper(TRANSLATION_LANGUAGE_MAP, 'en')).set_lang('ru')) 232 ) 233 234 ''' 235 self.lang = language 236 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')) )
class
TextTranslationReapplier(cengal.code_flow_control.call_history_reapplier.versions.v_0.call_history_reapplier.CallHistoryReapplier):
250class TextTranslationReapplier(CallHistoryReapplier): 251 def __init__(self, text_translator: TranslationLanguageChooser, priority: CoroPriority=CoroPriority.low): 252 self.text_translator = text_translator 253 super().__init__(priority) 254 255 def _translate_needed(self, value: Any, entity_id: Optional[TextEntityId]) -> Any: 256 if isinstance(value, TranslatableText): 257 return self.text_translator(value.text, entity_id) 258 else: 259 return value 260 261 def call_impl(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: TranslationWorker, text_template: Optional[str], *args, **kwargs): 262 new_args = list() 263 for arg in args: 264 new_args.append(self._translate_needed(arg, entity_id)) 265 266 new_kwargs = dict() 267 for key, value in kwargs.items(): 268 new_kwargs[key] = self._translate_needed(value, entity_id) 269 270 if text_template is None: 271 if new_kwargs: 272 raise RuntimeError('There are tt items in kwargs, however text_template is None') 273 274 translated_text = ' '.join(new_args) 275 else: 276 translated_text = text_template.format(*new_args, **new_kwargs) 277 278 translation_worker(translated_text) 279 280 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]: 281 return ((entity_id, id(obj), field), (translation_worker, text_template, args, kwargs)) 282 283 def key_value_to_args(self, key: Hashable, value: Any) -> Tuple[Tuple, Dict]: 284 entity_id, obj, field = key 285 translation_worker, text_template, args, kwargs = value 286 new_args = tuple([entity_id, obj, field, translation_worker, text_template] + list(args)) 287 return new_args, kwargs
TextTranslationReapplier( text_translator: TranslationLanguageChooser, priority: CoroPriority = <CoroPriority.low: 2>)
def
call_impl( self, entity_id: typing.Union[typing.Hashable, NoneType], obj: typing.Any, field: typing.Hashable, translation_worker: typing.Callable[[str], NoneType], text_template: typing.Union[str, NoneType], *args, **kwargs):
261 def call_impl(self, entity_id: Optional[TextEntityId], obj: Any, field: Hashable, translation_worker: TranslationWorker, text_template: Optional[str], *args, **kwargs): 262 new_args = list() 263 for arg in args: 264 new_args.append(self._translate_needed(arg, entity_id)) 265 266 new_kwargs = dict() 267 for key, value in kwargs.items(): 268 new_kwargs[key] = self._translate_needed(value, entity_id) 269 270 if text_template is None: 271 if new_kwargs: 272 raise RuntimeError('There are tt items in kwargs, however text_template is None') 273 274 translated_text = ' '.join(new_args) 275 else: 276 translated_text = text_template.format(*new_args, **new_kwargs) 277 278 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]:
Inherited Members
- cengal.code_flow_control.call_history_reapplier.versions.v_0.call_history_reapplier.CallHistoryReapplier
- history
- priority
- reapply
- destroy
class
CoroPriority(enum.Enum):
An enumeration.
high =
<CoroPriority.high: 0>
normal =
<CoroPriority.normal: 1>
low =
<CoroPriority.low: 2>
Inherited Members
- enum.Enum
- name
- value
TranslationWorker =
typing.Callable[[str], NoneType]
class
TranslatableText:
290class TranslatableText: 291 def __init__(self, text: Union[str, Callable], entity_id: Optional[TextEntityId] = None, formatter: Optional[Callable] = None) -> None: 292 self.text: str = text 293 self.entity_id: Optional[TextEntityId] = entity_id 294 self.formatter: Optional[Callable] = formatter 295 self.is_awaitable: bool = False 296 297 def __call__(self) -> Any: 298 if is_callable(self.text): 299 return self.text() 300 else: 301 return self.text 302 303 def format(self, text: str) -> str: 304 if self.formatter is None: 305 return text 306 else: 307 return self.formatter(text) 308 309 def __str__(self) -> str: 310 return self()
TranslatableText( text: typing.Union[str, typing.Callable], entity_id: typing.Union[typing.Hashable, NoneType] = None, formatter: typing.Union[typing.Callable, NoneType] = None)
291 def __init__(self, text: Union[str, Callable], entity_id: Optional[TextEntityId] = None, formatter: Optional[Callable] = None) -> None: 292 self.text: str = text 293 self.entity_id: Optional[TextEntityId] = entity_id 294 self.formatter: Optional[Callable] = formatter 295 self.is_awaitable: bool = False
tt =
<class 'TranslatableText'>
class
TranslateMe:
316class TranslateMe: 317 def __init__(self, *args) -> None: 318 self.args: Tuple[Union[str, TranslatableText]] = args 319 self._contains_translatable_text: bool = any(isinstance(arg, TranslatableText) for arg in self.args) 320 self._tte: 'TranslatableTextElement' = None 321 322 def __bool__(self) -> bool: 323 return self._contains_translatable_text 324 325 def to_str(self, text_translator: TextTranslator, language: TranslationLanguageId) -> str: 326 return ''.join([arg.format(text_translator(language, arg(), arg.entity_id)) if isinstance(arg, TranslatableText) else arg for arg in self.args]) 327 328 def tte(self, tte: 'TranslatableTextElement') -> 'TranslateMe': 329 self._tte = tte 330 return self 331 332 def __str__(self) -> str: 333 return self._tte.translate_me(self)
args: Tuple[Union[str, TranslatableText]]
TMe =
<class 'TranslateMe'>
tme =
<class 'TranslateMe'>
class
TranslatableTextElement(typing.Generic[~T]):
343class TranslatableTextElement(Generic[T]): 344 def __init__(self, text_translation_language_chooser: TranslationLanguageChooser) -> None: 345 self.text_translation_language_chooser: TranslationLanguageChooser = text_translation_language_chooser 346 self.text_translator: TextTranslator = text_translation_language_chooser.text_translator 347 self.elements_and_their_translatable_text: Dict[T, TranslatableText] = dict() 348 self.current_translation_reapplier_coros_ids: Set[CoroID] = set() 349 self.lang_changing_queue: List[Tuple[str, str]] = list() 350 351 def __call__(self, text_element: T) -> Any: 352 raise NotImplementedError() 353 354 def set_text(self, text_element: T, text: Union[TranslateMe, str]) -> T: 355 raise NotImplementedError() 356 357 def translate_me(self, translate_me: TranslateMe) -> str: 358 return translate_me.to_str(self.text_translator, self.text_translation_language_chooser.end_lang) 359 360 def start_translation_reapplier(self, i: Interface): 361 coro_id: CoroID = i(PutCoro, self.translation_reapplier) 362 self.current_translation_reapplier_coros_ids.add(coro_id) 363 364 async def translation_reapplier(self, i: Interface): 365 waiting_set: Set[CoroID] = self.current_translation_reapplier_coros_ids - {i.coro_id} 366 # await i(WaitCoroRequest().list(waiting_set)) # TODO: Not Implemented currently 367 for coro_id in waiting_set: 368 try: 369 await i(WaitCoroRequest().single(coro_id)) 370 except CoroutineNotFoundError: 371 pass 372 373 self.current_translation_reapplier_coros_ids = self.current_translation_reapplier_coros_ids - waiting_set 374 375 if not self.lang_changing_queue: 376 return 377 378 lang_changing_queue = self.lang_changing_queue 379 self.lang_changing_queue = type(self.lang_changing_queue)() 380 for lang, end_lang in lang_changing_queue: 381 for text_element, translate_me in self.elements_and_their_translatable_text.items(): 382 text_element.text = translate_me.to_str(self.text_translator, end_lang) 383 384 def register_on_lang_changed_handler(self): 385 i: Interface = current_interface() 386 i(AsyncEventBusRequest().register_handler(self.text_translation_language_chooser.translation_language_changed_event, self.on_lang_changed)) 387 388 async def aregister_on_lang_changed_handler(self): 389 i: Interface = current_interface() 390 await i(AsyncEventBusRequest().register_handler(self.text_translation_language_chooser.translation_language_changed_event, self.on_lang_changed)) 391 392 def remove_on_lang_changed_handler(self): 393 i: Interface = current_interface() 394 i(AsyncEventBusRequest().remove_handler(self.text_translation_language_chooser.translation_language_changed_event, self.on_lang_changed)) 395 396 async def aremove_on_lang_changed_handler(self): 397 i: Interface = current_interface() 398 await i(AsyncEventBusRequest().remove_handler(self.text_translation_language_chooser.translation_language_changed_event, self.on_lang_changed)) 399 400 def on_lang_changed(self, event: Hashable, data: Tuple[str, str]): 401 self.lang_changing_queue.append(data) 402 put_coro(self.start_translation_reapplier)
Abstract base class for generic types.
A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::
class Mapping(Generic[KT, VT]): def __getitem__(self, key: KT) -> VT: ... # Etc.
This class can then be used as follows::
def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default
TranslatableTextElement( text_translation_language_chooser: TranslationLanguageChooser)
344 def __init__(self, text_translation_language_chooser: TranslationLanguageChooser) -> None: 345 self.text_translation_language_chooser: TranslationLanguageChooser = text_translation_language_chooser 346 self.text_translator: TextTranslator = text_translation_language_chooser.text_translator 347 self.elements_and_their_translatable_text: Dict[T, TranslatableText] = dict() 348 self.current_translation_reapplier_coros_ids: Set[CoroID] = set() 349 self.lang_changing_queue: List[Tuple[str, str]] = list()
text_translation_language_chooser: TranslationLanguageChooser
text_translator: TextTranslator
elements_and_their_translatable_text: Dict[~T, TranslatableText]
def
start_translation_reapplier( self, i: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface):
async def
translation_reapplier( self, i: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface):
364 async def translation_reapplier(self, i: Interface): 365 waiting_set: Set[CoroID] = self.current_translation_reapplier_coros_ids - {i.coro_id} 366 # await i(WaitCoroRequest().list(waiting_set)) # TODO: Not Implemented currently 367 for coro_id in waiting_set: 368 try: 369 await i(WaitCoroRequest().single(coro_id)) 370 except CoroutineNotFoundError: 371 pass 372 373 self.current_translation_reapplier_coros_ids = self.current_translation_reapplier_coros_ids - waiting_set 374 375 if not self.lang_changing_queue: 376 return 377 378 lang_changing_queue = self.lang_changing_queue 379 self.lang_changing_queue = type(self.lang_changing_queue)() 380 for lang, end_lang in lang_changing_queue: 381 for text_element, translate_me in self.elements_and_their_translatable_text.items(): 382 text_element.text = translate_me.to_str(self.text_translator, end_lang)
TTE =
<class 'TranslatableTextElement'>