__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
# 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_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]
def delete_books(self, paths, end_session=True):
'''
Delete books at paths on device.
'''
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 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()
Change language