← Back to branch summary

~usb-creator-hackers/usb-creator/trunk

« back to all changes in this revision

Viewing changes to usbcreator/install.py

  • Committer: Benjamin Drung
  • Date: 2022-11-02 13:01:26 UTC
  • Revision ID: benjamin.drung@canonical.com-20221102130126-4z0xyivy5f37dp13
Move to https://code.launchpad.net/~usb-creator-hackers/usb-creator/+git/main

Show diffs side-by-side

added added

removed removed

Lines of Context:
usbcreator/install.py
1
 
# Copyright (C) 2008, 2009 Canonical Ltd.
2
 
 
3
 
# This program is free software: you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License version 3,
5
 
# as published by the Free Software Foundation.
6
 
#
7
 
# This program is distributed in the hope that it will be useful,
8
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
9
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
10
 
# GNU General Public License for more details.
11
 
#
12
 
# You should have received a copy of the GNU General Public License
13
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
14
 
 
15
 
import os
16
 
import stat
17
 
import sys
18
 
import shutil
19
 
from usbcreator.misc import (
20
 
    USBCreatorProcessException,
21
 
    callable,
22
 
    popen,
23
 
    )
24
 
from threading import Thread, Event
25
 
import logging
26
 
from hashlib import md5
27
 
from usbcreator.misc import MAX_DBUS_TIMEOUT
28
 
 
29
 
 
30
 
class install(Thread):
31
 
    def __init__(self, source, target, device=None,
32
 
                 allow_system_internal=False):
33
 
        Thread.__init__(self)
34
 
        self.source = source
35
 
        self.target = target
36
 
        self.device = device
37
 
        self.allow_system_internal = allow_system_internal
38
 
        self._stopevent = Event()
39
 
        logging.debug('install thread source: %s' % source)
40
 
        logging.debug('install thread target: %s' % target)
41
 
 
42
 
    # Signals.
43
 
 
44
 
    def success(self):
45
 
        pass
46
 
    
47
 
    def _success(self):
48
 
        if callable(self.success):
49
 
            self.success()
50
 
 
51
 
    def failure(self, message=None):
52
 
        pass
53
 
 
54
 
    def _failure(self, message=None):
55
 
        logging.critical(message)
56
 
        if callable(self.failure):
57
 
            self.failure(message)
58
 
        sys.exit(1)
59
 
 
60
 
    def progress(self, complete):
61
 
        '''Emitted with an integer percentage of progress completed, time
62
 
        remaining, and speed.'''
63
 
        pass
64
 
 
65
 
    def progress_message(self, message):
66
 
        '''Emitted with a translated string like "Installing the
67
 
        bootloader..."
68
 
        '''
69
 
        pass
70
 
 
71
 
    def retry(self, message):
72
 
        '''Will be called when we need to know if the user wants to try a
73
 
        failed operation again.  Must return a boolean value.'''
74
 
        pass
75
 
    
76
 
    def join(self, timeout=None):
77
 
        self._stopevent.set()
78
 
        Thread.join(self, timeout)
79
 
 
80
 
    def check(self):
81
 
        if self._stopevent.isSet():
82
 
            logging.debug('Asked by the controlling thread to shut down.')
83
 
            sys.exit(0)
84
 
    
85
 
    # Exception catching wrapper.
86
 
 
87
 
    def run(self):
88
 
        try:
89
 
            if os.path.isfile(self.source):
90
 
                ext = os.path.splitext(self.source)[1].lower()
91
 
                if ext not in ['.iso', '.img']:
92
 
                    self._failure(_('The extension "%s" is not supported.') %
93
 
                                    ext)
94
 
                self.diskimage_install()
95
 
            else:
96
 
                self.diskimage_install()
97
 
            self._success()
98
 
        except Exception as e:
99
 
            # TODO evand 2009-07-25: Bring up our own apport-like utility.
100
 
            logging.exception('Exception raised:')
101
 
            self._failure(_('An uncaught exception was raised:\n%s') % str(e))
102
 
 
103
 
    # Helpers for core routines.
104
 
    def diskimage_install(self):
105
 
        self.progress_message(_('Writing disk image...'))
106
 
        failure_msg = _('Could not write the disk image (%(source)s) to the device'
107
 
                        ' (%(device)s).') % {'source': self.source,
108
 
                                             'device': self.device}
109
 
        
110
 
        import dbus
111
 
        try:
112
 
            bus = dbus.SystemBus()
113
 
            obj = bus.get_object('com.ubuntu.USBCreator',
114
 
                                 '/com/ubuntu/USBCreator')
115
 
            obj.Image(self.source, self.device, self.allow_system_internal,
116
 
                      dbus_interface='com.ubuntu.USBCreator',
117
 
                      timeout=MAX_DBUS_TIMEOUT)
118
 
        except dbus.DBusException:
119
 
            self._failure(failure_msg)
120