File: factory.py

package info (click to toggle)
ibus-typing-booster 2.30.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky
  • size: 141,124 kB
  • sloc: xml: 1,123,826; python: 46,963; sh: 5,183; makefile: 373; sed: 16
file content (135 lines) | stat: -rw-r--r-- 5,598 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
# vim:et sw=4 sts=4 sw=4
#
# ibus-typing-booster - A completion input method for IBus
#
# Copyright (c) 2011-2013 Anish Patil <apatil@redhat.com>
# Copyright (c) 2012-2018 Mike FABIAN <mfabian@redhat.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>
'''
TypingBoosterEngine Factory
'''
from typing import Dict
from typing import Optional
import re
import os
import logging
import gettext
from gi import require_version
# pylint: disable=wrong-import-position
require_version('IBus', '1.0')
from gi.repository import IBus
# pylint: enable=wrong-import-position
import hunspell_table
import tabsqlitedb
import itb_util

LOGGER = logging.getLogger('ibus-typing-booster')

DEBUG_LEVEL = int(0)

DOMAINNAME = 'ibus-typing-booster'

def _(text: str) -> str:
    '''Gettext translation function.'''
    return gettext.dgettext(DOMAINNAME, text)

def N_(text: str) -> str: # pylint: disable=invalid-name
    '''Mark string for translation without actually translating.

    Used by gettext tools to extract strings that need translation.
    '''
    return text

class EngineFactory(IBus.Factory):
    """Table IM Engine Factory"""
    def __init__(self, bus: IBus.Bus) -> None:
        global DEBUG_LEVEL # pylint: disable=global-statement
        try:
            DEBUG_LEVEL = int(
                str(os.getenv('IBUS_TYPING_BOOSTER_DEBUG_LEVEL')))
        except (TypeError, ValueError):
            DEBUG_LEVEL = int(0)
        if DEBUG_LEVEL > 1:
            LOGGER.debug('EngineFactory.__init__(bus=%s)\n', bus)
        self.database: Optional[tabsqlitedb.TabSqliteDb] = None
        self.database_dict: Dict[str, tabsqlitedb.TabSqliteDb] = {}
        self.enginedict: Dict[str, hunspell_table.TypingBoosterEngine] = {}
        self.bus = bus
        #engine.Engine.CONFIG_RELOADED(bus)
        super().__init__(
            connection=bus.get_connection(), object_path=IBus.PATH_FACTORY)
        self.engine_id = 0

    def do_create_engine( # pylint: disable=arguments-differ
            self, engine_name: str) -> hunspell_table.TypingBoosterEngine:
        if DEBUG_LEVEL > 1:
            LOGGER.debug(
                'EngineFactory.do_create_engine(engine_name=%s)\n',
                engine_name)
        engine_path = ('/com/redhat/IBus/engines/typing_booster/'
                       f"{re.sub(r'[^a-zA-Z0-9_/]', '_', engine_name)}"
                       '/engine/')
        try:
            if engine_name in self.database_dict:
                self.database = self.database_dict[engine_name]
            else:
                user_db_file = 'user.db'
                if engine_name != 'typing-booster':
                    match = itb_util.M17N_ENGINE_NAME_PATTERN.search(
                            engine_name)
                    if not match:
                        raise ValueError('Invalid engine name.')
                    m17n_ime_lang = match.group('lang')
                    m17n_ime_name = match.group('name')
                    user_db_file = f'user-{m17n_ime_lang}-{m17n_ime_name}.db'
                self.database = tabsqlitedb.TabSqliteDb(
                    user_db_file=user_db_file)
                self.database_dict[engine_name] = self.database
            if engine_name in self.enginedict:
                engine = self.enginedict[engine_name]
            else:
                engine = hunspell_table.TypingBoosterEngine(
                    self.bus,
                    engine_path + str(self.engine_id),
                    self.database,
                    engine_name=engine_name)
                self.enginedict[engine_name] = engine
                self.engine_id += 1
            return engine
        except Exception as error: # pylint: disable=broad-except
            LOGGER.exception(
                'Failed to create engine %s: %s: %s',
                engine_name, error.__class__.__name__, error)
            raise Exception from error # pylint: disable=broad-exception-raised

    def do_destroy(self) -> None:  # pylint: disable=arguments-differ
        '''Destructor, which finish some task for IME'''
        # It is necessary to call the do_destroy() function here to
        # get finish stuff properly, for example do_focus_out() needs
        # to be called because it calls
        # _revert_autosettings(). Without that, applied autosettings
        # survive a “ibus restart” if the autosettings are written go
        # gsettings when they are applied. I an not sure yet whether
        # writing applied autosettings to gsettings is a good idea,
        # but calling do_destroy() here is a good idea anyway to make
        # sure everything is properly finished.
        for key, engine in self.enginedict.items():
            LOGGER.info('Destroying engine %s %s', key, engine)
            engine.do_destroy()
        LOGGER.info('Syncing user database(s)')
        for key, database in self.database_dict.items():
            LOGGER.info('Syncing %s %s', key, database)
            database.sync_usrdb()
        super().destroy()