Source code for calibre.devices.interface

__license__   = 'GPL v3'
__copyright__ = '2008, Kovid Goyal <kovid at kovidgoyal.net>'
import os
from collections import namedtuple

from calibre import prints
from calibre.constants import iswindows
from calibre.customize import Plugin


class OpenPopupMessage:

    def __init__(self, title='', message='', level='info', skip_dialog_skip_precheck=True):
        self.title = title
        self.message = message
        self.level = level
        self.skip_dialog_skip_precheck = skip_dialog_skip_precheck


[docs] class DevicePlugin(Plugin): """ Defines the interface that should be implemented by backends that communicate with an e-book reader. """ type = _('Device interface') #: Ordered list of supported formats FORMATS = ["lrf", "rtf", "pdf", "txt"] # If True, the config dialog will not show the formats box HIDE_FORMATS_CONFIG_BOX = False #: VENDOR_ID can be either an integer, a list of integers or a dictionary #: If it is a dictionary, it must be a dictionary of dictionaries, #: of the form:: #: #: { #: integer_vendor_id : { product_id : [list of BCDs], ... }, #: ... #: } #: VENDOR_ID = 0x0000 #: An integer or a list of integers PRODUCT_ID = 0x0000 #: BCD can be either None to not distinguish between devices based on BCD, or #: it can be a list of the BCD numbers of all devices supported by this driver. BCD = None #: Height for thumbnails on the device THUMBNAIL_HEIGHT = 68 #: Compression quality for thumbnails. Set this closer to 100 to have better #: quality thumbnails with fewer compression artifacts. Of course, the #: thumbnails get larger as well. THUMBNAIL_COMPRESSION_QUALITY = 75 #: Set this to True if the device supports updating cover thumbnails during #: sync_booklists. Setting it to true will ask device.py to refresh the #: cover thumbnails during book matching WANTS_UPDATED_THUMBNAILS = False #: Whether the metadata on books can be set via the GUI. CAN_SET_METADATA = ['title', 'authors', 'collections'] #: Whether the device can handle device_db metadata plugboards CAN_DO_DEVICE_DB_PLUGBOARD = False # Set this to None if the books on the device are files that the GUI can # access in order to add the books from the device to the library BACKLOADING_ERROR_MESSAGE = _('Cannot get files from this device') #: Path separator for paths to books on device path_sep = os.sep #: Icon for this device icon = 'reader.png' # Encapsulates an annotation fetched from the device UserAnnotation = namedtuple('Annotation','type, value') #: GUI displays this as a message if not None in the status bar. Useful if opening can take a #: long time OPEN_FEEDBACK_MESSAGE = None #: Set of extensions that are "virtual books" on the device #: and therefore cannot be viewed/saved/added to library. #: For example: ``frozenset(['kobo'])`` VIRTUAL_BOOK_EXTENSIONS = frozenset() #: Message to display to user for virtual book extensions. VIRTUAL_BOOK_EXTENSION_MESSAGE = None #: Whether to nuke comments in the copy of the book sent to the device. If #: not None this should be short string that the comments will be replaced #: by. NUKE_COMMENTS = None #: If True indicates that this driver completely manages device detection, #: ejecting and so forth. If you set this to True, you *must* implement the #: detect_managed_devices and debug_managed_device_detection methods. #: A driver with this set to true is responsible for detection of devices, #: managing a blacklist of devices, a list of ejected devices and so forth. #: calibre will periodically call the detect_managed_devices() method and #: if it returns a detected device, calibre will call open(). open() will #: be called every time a device is returned even if previous calls to open() #: failed, therefore the driver must maintain its own blacklist of failed #: devices. Similarly, when ejecting, calibre will call eject() and then #: assuming the next call to detect_managed_devices() returns None, it will #: call post_yank_cleanup(). MANAGES_DEVICE_PRESENCE = False #: If set the True, calibre will call the :meth:`get_driveinfo()` method #: after the books lists have been loaded to get the driveinfo. SLOW_DRIVEINFO = False #: If set to True, calibre will ask the user if they want to manage the #: device with calibre, the first time it is detected. If you set this to #: True you must implement :meth:`get_device_uid()` and #: :meth:`ignore_connected_device()` and #: :meth:`get_user_blacklisted_devices` and #: :meth:`set_user_blacklisted_devices` ASK_TO_ALLOW_CONNECT = False #: Set this to a dictionary of the form {'title':title, 'msg':msg, 'det_msg':detailed_msg} to have calibre popup #: a message to the user after some callbacks are run (currently only upload_books). #: Be careful to not spam the user with too many messages. This variable is checked after *every* callback, #: so only set it when you really need to. user_feedback_after_callback = None @classmethod def get_gui_name(cls): if hasattr(cls, 'gui_name'): return cls.gui_name if hasattr(cls, '__name__'): return cls.__name__ return cls.name
[docs] @classmethod def get_open_popup_message(self): ' GUI displays this as a non-modal popup. Should be an instance of OpenPopupMessage ' return None
# Device detection {{{ def test_bcd(self, bcdDevice, bcd): if bcd is None or len(bcd) == 0: return True for c in bcd: if c == bcdDevice: return True return False
[docs] def is_usb_connected(self, devices_on_system, debug=False, only_presence=False): ''' Return True, device_info if a device handled by this plugin is currently connected. :param devices_on_system: List of devices currently connected ''' vendors_on_system = {x[0] for x in devices_on_system} vendors = set(self.VENDOR_ID) if hasattr(self.VENDOR_ID, '__len__') else {self.VENDOR_ID} if hasattr(self.VENDOR_ID, 'keys'): products = [] for ven in self.VENDOR_ID: products.extend(self.VENDOR_ID[ven].keys()) else: products = self.PRODUCT_ID if hasattr(self.PRODUCT_ID, '__len__') else [self.PRODUCT_ID] ch = self.can_handle_windows if iswindows else self.can_handle for vid in vendors_on_system.intersection(vendors): for dev in devices_on_system: cvid, pid, bcd = dev[:3] if cvid == vid: if pid in products: if hasattr(self.VENDOR_ID, 'keys'): try: cbcd = self.VENDOR_ID[vid][pid] except KeyError: # Vendor vid does not have product pid, pid # exists for some other vendor in this # device continue else: cbcd = self.BCD if self.test_bcd(bcd, cbcd): if debug: prints(dev) if ch(dev, debug=debug): return True, dev return False, None
[docs] def detect_managed_devices(self, devices_on_system, force_refresh=False): ''' Called only if MANAGES_DEVICE_PRESENCE is True. Scan for devices that this driver can handle. Should return a device object if a device is found. This object will be passed to the open() method as the connected_device. If no device is found, return None. The returned object can be anything, calibre does not use it, it is only passed to open(). This method is called periodically by the GUI, so make sure it is not too resource intensive. Use a cache to avoid repeatedly scanning the system. :param devices_on_system: Set of USB devices found on the system. :param force_refresh: If True and the driver uses a cache to prevent repeated scanning, the cache must be flushed. ''' raise NotImplementedError()
[docs] def debug_managed_device_detection(self, devices_on_system, output): ''' Called only if MANAGES_DEVICE_PRESENCE is True. Should write information about the devices detected on the system to output, which is a file like object. Should return True if a device was detected and successfully opened, otherwise False. ''' raise NotImplementedError()
# }}}
[docs] def reset(self, key='-1', log_packets=False, report_progress=None, detected_device=None): """ :param key: The key to unlock the device :param log_packets: If true the packet stream to/from the device is logged :param report_progress: Function that is called with a % progress (number between 0 and 100) for various tasks. If it is called with -1 that means that the task does not have any progress information :param detected_device: Device information from the device scanner """ raise NotImplementedError()
[docs] def can_handle_windows(self, usbdevice, debug=False): ''' Optional method to perform further checks on a device to see if this driver is capable of handling it. If it is not it should return False. This method is only called after the vendor, product ids and the bcd have matched, so it can do some relatively time intensive checks. The default implementation returns True. This method is called only on Windows. See also :meth:`can_handle`. Note that for devices based on USBMS this method by default delegates to :meth:`can_handle`. So you only need to override :meth:`can_handle` in your subclass of USBMS. :param usbdevice: A usbdevice as returned by :func:`calibre.devices.winusb.scan_usb_devices` ''' return True
[docs] def can_handle(self, device_info, debug=False): ''' Unix version of :meth:`can_handle_windows`. :param device_info: Is a tuple of (vid, pid, bcd, manufacturer, product, serial number) ''' return True
can_handle.is_base_class_implementation = True
[docs] def open(self, connected_device, library_uuid): ''' Perform any device specific initialization. Called after the device is detected but before any other functions that communicate with the device. For example: For devices that present themselves as USB Mass storage devices, this method would be responsible for mounting the device or if the device has been automounted, for finding out where it has been mounted. The method :meth:`calibre.devices.usbms.device.Device.open` has an implementation of this function that should serve as a good example for USB Mass storage devices. This method can raise an OpenFeedback exception to display a message to the user. :param connected_device: The device that we are trying to open. It is a tuple of (vendor id, product id, bcd, manufacturer name, product name, device serial number). However, some devices have no serial number and on Windows only the first three fields are present, the rest are None. :param library_uuid: The UUID of the current calibre library. Can be None if there is no library (for example when used from the command line). ''' raise NotImplementedError()
[docs] def eject(self): ''' Un-mount / eject the device from the OS. This does not check if there are pending GUI jobs that need to communicate with the device. NOTE: That this method may not be called on the same thread as the rest of the device methods. ''' raise NotImplementedError()
[docs] def post_yank_cleanup(self): ''' Called if the user yanks the device without ejecting it first. ''' raise NotImplementedError()
[docs] def set_progress_reporter(self, report_progress): ''' Set a function to report progress information. :param report_progress: Function that is called with a % progress (number between 0 and 100) for various tasks. If it is called with -1 that means that the task does not have any progress information ''' raise NotImplementedError()
[docs] def get_device_information(self, end_session=True): """ Ask device for device information. See L{DeviceInfoQuery}. :return: (device name, device version, software version on device, MIME type) The tuple can optionally have a fifth element, which is a drive information dictionary. See usbms.driver for an example. """ raise NotImplementedError()
[docs] def get_driveinfo(self): ''' Return the driveinfo dictionary. Usually called from get_device_information(), but if loading the driveinfo is slow for this driver, then it should set SLOW_DRIVEINFO. In this case, this method will be called by calibre after the book lists have been loaded. Note that it is not called on the device thread, so the driver should cache the drive info in the books() method and this function should return the cached data. ''' return {}
[docs] def card_prefix(self, end_session=True): ''' Return a 2 element list of the prefix to paths on the cards. If no card is present None is set for the card's prefix. E.G. ('/place', '/place2') (None, 'place2') ('place', None) (None, None) ''' raise NotImplementedError()
[docs] def total_space(self, end_session=True): """ Get total space available on the mountpoints: 1. Main memory 2. Memory Card A 3. Memory Card B :return: A 3 element list with total space in bytes of (1, 2, 3). If a particular device doesn't have any of these locations it should return 0. """ raise NotImplementedError()
[docs] def free_space(self, end_session=True): """ Get free space available on the mountpoints: 1. Main memory 2. Card A 3. Card B :return: A 3 element list with free space in bytes of (1, 2, 3). If a particular device doesn't have any of these locations it should return -1. """ raise NotImplementedError()
[docs] def books(self, oncard=None, end_session=True): """ Return a list of e-books on the device. :param oncard: If 'carda' or 'cardb' return a list of e-books on the specific storage card, otherwise return list of e-books in main memory of device. If a card is specified and no books are on the card return empty list. :return: A BookList. """ raise NotImplementedError()
[docs] def upload_books(self, files, names, on_card=None, end_session=True, metadata=None): ''' Upload a list of books to the device. If a file already exists on the device, it should be replaced. This method should raise a :class:`FreeSpaceError` if there is not enough free space on the device. The text of the FreeSpaceError must contain the word "card" if ``on_card`` is not None otherwise it must contain the word "memory". :param files: A list of paths :param names: A list of file names that the books should have once uploaded to the device. len(names) == len(files) :param metadata: If not None, it is a list of :class:`Metadata` objects. The idea is to use the metadata to determine where on the device to put the book. len(metadata) == len(files). Apart from the regular cover (path to cover), there may also be a thumbnail attribute, which should be used in preference. The thumbnail attribute is of the form (width, height, cover_data as jpeg). :return: A list of 3-element tuples. The list is meant to be passed to :meth:`add_books_to_metadata`. ''' raise NotImplementedError()
[docs] @classmethod def add_books_to_metadata(cls, locations, metadata, booklists): ''' Add locations to the booklists. This function must not communicate with the device. :param locations: Result of a call to L{upload_books} :param metadata: List of :class:`Metadata` objects, same as for :meth:`upload_books`. :param booklists: A tuple containing the result of calls to (:meth:`books(oncard=None)`, :meth:`books(oncard='carda')`, :meth`books(oncard='cardb')`). ''' raise NotImplementedError()
[docs] def delete_books(self, paths, end_session=True): ''' Delete books at paths on device. ''' raise NotImplementedError()
[docs] @classmethod def remove_books_from_metadata(cls, paths, booklists): ''' Remove books from the metadata list. This function must not communicate with the device. :param paths: paths to books on the device. :param booklists: A tuple containing the result of calls to (:meth:`books(oncard=None)`, :meth:`books(oncard='carda')`, :meth`books(oncard='cardb')`). ''' raise NotImplementedError()
[docs] def sync_booklists(self, booklists, end_session=True): ''' Update metadata on device. :param booklists: A tuple containing the result of calls to (:meth:`books(oncard=None)`, :meth:`books(oncard='carda')`, :meth`books(oncard='cardb')`). ''' raise NotImplementedError()
[docs] def get_file(self, path, outfile, end_session=True): ''' Read the file at ``path`` on the device and write it to outfile. :param outfile: file object like ``sys.stdout`` or the result of an :func:`open` call. ''' raise NotImplementedError()
[docs] @classmethod def config_widget(cls): ''' Should return a QWidget. The QWidget contains the settings for the device interface ''' raise NotImplementedError()
[docs] @classmethod def save_settings(cls, settings_widget): ''' Should save settings to disk. Takes the widget created in :meth:`config_widget` and saves all settings to disk. ''' raise NotImplementedError()
[docs] @classmethod def settings(cls): ''' Should return an opts object. The opts object should have at least one attribute `format_map` which is an ordered list of formats for the device. ''' raise NotImplementedError()
[docs] def set_plugboards(self, plugboards, pb_func): ''' provide the driver the current set of plugboards and a function to select a specific plugboard. This method is called immediately before add_books and sync_booklists. pb_func is a callable with the following signature:: def pb_func(device_name, format, plugboards) You give it the current device name (either the class name or DEVICE_PLUGBOARD_NAME), the format you are interested in (a 'real' format or 'device_db'), and the plugboards (you were given those by set_plugboards, the same place you got this method). :return: None or a single plugboard instance. ''' pass
[docs] def set_driveinfo_name(self, location_code, name): ''' Set the device name in the driveinfo file to 'name'. This setting will persist until the file is re-created or the name is changed again. Non-disk devices should implement this method based on the location codes returned by the get_device_information() method. ''' pass
[docs] def prepare_addable_books(self, paths): ''' Given a list of paths, returns another list of paths. These paths point to addable versions of the books. If there is an error preparing a book, then instead of a path, the position in the returned list for that book should be a three tuple: (original_path, the exception instance, traceback) ''' return paths
[docs] def startup(self): ''' Called when calibre is starting the device. Do any initialization required. Note that multiple instances of the class can be instantiated, and thus __init__ can be called multiple times, but only one instance will have this method called. This method is called on the device thread, not the GUI thread. ''' pass
[docs] def shutdown(self): ''' Called when calibre is shutting down, either for good or in preparation to restart. Do any cleanup required. This method is called on the device thread, not the GUI thread. ''' pass
[docs] def get_device_uid(self): ''' Must return a unique id for the currently connected device (this is called immediately after a successful call to open()). You must implement this method if you set ASK_TO_ALLOW_CONNECT = True ''' raise NotImplementedError()
[docs] def ignore_connected_device(self, uid): ''' Should ignore the device identified by uid (the result of a call to get_device_uid()) in the future. You must implement this method if you set ASK_TO_ALLOW_CONNECT = True. Note that this function is called immediately after open(), so if open() caches some state, the driver should reset that state. ''' raise NotImplementedError()
[docs] def get_user_blacklisted_devices(self): ''' Return map of device uid to friendly name for all devices that the user has asked to be ignored. ''' return {}
[docs] def set_user_blacklisted_devices(self, devices): ''' Set the list of device uids that should be ignored by this driver. ''' pass
[docs] def specialize_global_preferences(self, device_prefs): ''' Implement this method if your device wants to override a particular preference. You must ensure that all call sites that want a preference that can be overridden use device_prefs['something'] instead of prefs['something']. Your method should call device_prefs.set_overrides(pref=val, pref=val, ...). Currently used for: metadata management (prefs['manage_device_metadata']) ''' device_prefs.set_overrides()
[docs] def set_library_info(self, library_name, library_uuid, field_metadata): ''' Implement this method if you want information about the current calibre library. This method is called at startup and when the calibre library changes while connected. ''' pass
# Dynamic control interface. # The following methods are probably called on the GUI thread. Any driver # that implements these methods must take pains to be thread safe, because # the device_manager might be using the driver at the same time that one of # these methods is called.
[docs] def is_dynamically_controllable(self): ''' Called by the device manager when starting plugins. If this method returns a string, then a) it supports the device manager's dynamic control interface, and b) that name is to be used when talking to the plugin. This method can be called on the GUI thread. A driver that implements this method must be thread safe. ''' return None
[docs] def start_plugin(self): ''' This method is called to start the plugin. The plugin should begin to accept device connections however it does that. If the plugin is already accepting connections, then do nothing. This method can be called on the GUI thread. A driver that implements this method must be thread safe. ''' pass
[docs] def stop_plugin(self): ''' This method is called to stop the plugin. The plugin should no longer accept connections, and should cleanup behind itself. It is likely that this method should call shutdown. If the plugin is already not accepting connections, then do nothing. This method can be called on the GUI thread. A driver that implements this method must be thread safe. ''' pass
[docs] def get_option(self, opt_string, default=None): ''' Return the value of the option indicated by opt_string. This method can be called when the plugin is not started. Return None if the option does not exist. This method can be called on the GUI thread. A driver that implements this method must be thread safe. ''' return default
[docs] def set_option(self, opt_string, opt_value): ''' Set the value of the option indicated by opt_string. This method can be called when the plugin is not started. This method can be called on the GUI thread. A driver that implements this method must be thread safe. ''' pass
[docs] def is_running(self): ''' Return True if the plugin is started, otherwise false This method can be called on the GUI thread. A driver that implements this method must be thread safe. ''' return False
[docs] def synchronize_with_db(self, db, book_id, book_metadata, first_call): ''' Called during book matching when a book on the device is matched with a book in calibre's db. The method is responsible for synchronizing data from the device to calibre's db (if needed). The method must return a two-value tuple. The first value is a set of calibre book ids changed if calibre's database was changed or None if the database was not changed. If the first value is an empty set then the metadata for the book on the device is updated with calibre's metadata and given back to the device, but no GUI refresh of that book is done. This is useful when the calibre data is correct but must be sent to the device. The second value is itself a 2-value tuple. The first value in the tuple specifies whether a book format should be sent to the device. The intent is to permit verifying that the book on the device is the same as the book in calibre. This value must be None if no book is to be sent, otherwise return the base file name on the device (a string like foobar.epub). Be sure to include the extension in the name. The device subsystem will construct a send_books job for all books with not- None returned values. Note: other than to later retrieve the extension, the name is ignored in cases where the device uses a template to generate the file name, which most do. The second value in the returned tuple indicated whether the format is future-dated. Return True if it is, otherwise return False. calibre will display a dialog to the user listing all future dated books. Extremely important: this method is called on the GUI thread. It must be threadsafe with respect to the device manager's thread. book_id: the calibre id for the book in the database. book_metadata: the Metadata object for the book coming from the device. first_call: True if this is the first call during a sync, False otherwise ''' return (None, (None, False))
[docs] class BookList(list): ''' A list of books. Each Book object must have the fields #. title #. authors #. size (file size of the book) #. datetime (a UTC time tuple) #. path (path on the device to the book) #. thumbnail (can be None) thumbnail is either a str/bytes object with the image data or it should have an attribute image_path that stores an absolute (platform native) path to the image #. tags (a list of strings, can be empty). ''' __getslice__ = None __setslice__ = None def __init__(self, oncard, prefix, settings): pass
[docs] def supports_collections(self): ''' Return True if the device supports collections for this book list. ''' raise NotImplementedError()
[docs] def add_book(self, book, replace_metadata): ''' Add the book to the booklist. Intent is to maintain any device-internal metadata. Return True if booklists must be sync'ed ''' raise NotImplementedError()
[docs] def remove_book(self, book): ''' Remove a book from the booklist. Correct any device metadata at the same time ''' raise NotImplementedError()
[docs] def get_collections(self, collection_attributes): ''' Return a dictionary of collections created from collection_attributes. Each entry in the dictionary is of the form collection name:[list of books] The list of books is sorted by book title, except for collections created from series, in which case series_index is used. :param collection_attributes: A list of attributes of the Book object ''' raise NotImplementedError()
class CurrentlyConnectedDevice: def __init__(self): self._device = None @property def device(self): return self._device # A device driver can check if a device is currently connected to calibre using # the following code:: # from calibre.device.interface import currently_connected_device # if currently_connected_device.device is None: # # no device connected # The device attribute will be either None or the device driver object # (DevicePlugin instance) for the currently connected device. currently_connected_device = CurrentlyConnectedDevice()