Source code for cosas.ui

# Copyright 2012 Canonical Ltd.
# This file is part of u1db.
# u1db is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License version 3
# as published by the Free Software Foundation.
# u1db is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public License
# along with u1db.  If not, see <>.

"""User interface for the cosas example application."""

from collections import defaultdict
from datetime import datetime
import os
import sys
from PyQt4 import QtGui, QtCore, uic

from cosas import TodoStore, get_database, extract_tags
from u1db.errors import DatabaseDoesNotExist
from u1db.remote.http_database import HTTPDatabase
from ubuntuone.platform.credentials import CredentialsManagementTool

FOREGROUND = QtGui.QColor('#1d1f21')
DONE = QtGui.QColor('#969896')
CONFLICT_COLOR = QtGui.QColor('#A54242')
    '#8C9440', '#de935f', '#5F819D', '#85678F',
    '#5E8D87', '#cc6666', '#b5bd68', '#f0c674',
    '#81a2be', '#b294bb']
U1_URL = ''
TIMEOUT = 1000 * 0.5 * 60 * 60  # 30 minutes

[docs]class UITask(QtGui.QTreeWidgetItem): """Task list item.""" def __init__(self, task, parent, store, font, main_window): super(UITask, self).__init__(parent) self.task = task # If the task is done, check off the list item. = store self._bg_color = BACKGROUND self._font = font self.main_window = main_window def set_color(self, color): self._bg_color = color def setData(self, column, role, value): if column == 1: if role == QtCore.Qt.CheckStateRole: if value == QtCore.Qt.Checked: self.task.done = True else: self.task.done = False if role == QtCore.Qt.EditRole: text = unicode(value.toString(), 'utf-8') if not text: # There was no text in the edit field so do nothing. return self.update_task_text(text) super(UITask, self).setData(column, role, value) def data(self, column, role): if role == QtCore.Qt.BackgroundRole and column == 0: return self._bg_color if role == QtCore.Qt.ForegroundRole: if self.task.has_conflicts: return CONFLICT_COLOR return DONE if self.task.done else FOREGROUND if column == 1: if role == QtCore.Qt.FontRole: font = self._font font.setStrikeOut(self.task.done) return font if role == QtCore.Qt.EditRole: return self.task.title if role == QtCore.Qt.DisplayRole: return self.task.title if role == QtCore.Qt.CheckStateRole: return ( QtCore.Qt.Checked if self.task.done else QtCore.Qt.Unchecked) elif column == 2: if role == QtCore.Qt.DisplayRole: return '!' if self.task.has_conflicts else '' return super(UITask, self).data(column, role)
[docs] def update_task_text(self, text): """Edit an existing todo item.""" # Change the task's title to the text in the edit field. self.task.title = text # Record the current tags. old_tags = set(self.task.tags) if self.task.tags else set([]) # Extract the new tags from the new text. new_tags = set(extract_tags(text)) # Check if the tag filter buttons need updating. self.main_window.update_tags(self, old_tags, new_tags) # Set the tags on the task. self.task.tags = list(new_tags) # Save the changed task to the database.
class Conflicts(QtGui.QDialog): def __init__(self, other, conflicts): super(Conflicts, self).__init__() self.selected_doc = None uifile = os.path.join( os.path.abspath(os.path.dirname(__file__)), 'conflicts.ui') uic.loadUi(uifile, self) self.other = other self.revs = [] for conflict in conflicts: self.revs.append(conflict.rev) # XXX: this does not deserve any prizes, but it was the quickest # way I could figure out to use loop variables in a 'closure' and # not just get the last value everywhere. def toggled(value, doc=conflict): if value: self.selected_doc = doc radio = QtGui.QRadioButton(conflict.title) self.conflicts.layout().addWidget(radio) if conflict.done: font = radio.font() font.setStrikeOut(True) radio.setFont(font) radio.toggled.connect(toggled) def accept(self): if self.selected_doc: self.other.resolve(self.selected_doc, self.revs) super(Conflicts, self).accept() class Sync(QtGui.QDialog): def __init__(self, other): super(Sync, self).__init__() uifile = os.path.join( os.path.abspath(os.path.dirname(__file__)), 'sync.ui') uic.loadUi(uifile, self) self.other = other if other.auto_sync: self.auto_sync.setChecked(True) if other.sync_target == U1_URL: self.u1_radio.setChecked(True) self.url_radio.setChecked(False) else: self.url_radio.setChecked(True) self.u1_radio.setChecked(False) self.url_edit.setText(other.sync_target) self.connect_events() def connect_events(self): """Hook up all the signal handlers.""" self.sync_button.clicked.connect(self.synchronize) self.u1_radio.toggled.connect(self.toggle_u1) self.url_radio.toggled.connect(self.toggle_url) self.auto_sync.toggled.connect(self.toggle_sync) self.url_edit.editingFinished.connect(self.url_changed) def enable_button(self, _=None): self.sync_button.setEnabled(True) def synchronize(self): self.sync_button.setEnabled(False) self.other.synchronize(self.enable_button) self.last_synced.setText( '<span style="color:green">%s</span>' % (,)) def toggle_u1(self, value): if value: self.other.sync_target = U1_URL else: text = unicode(self.url_edit.text(), 'utf-8') if not text: # There was no text in the edit field so do nothing. self.other.sync_target = None return self.other.sync_target = text def toggle_url(self, value): if value: text = unicode(self.url_edit.text(), 'utf-8') if not text: # There was no text in the edit field so do nothing. self.other.sync_target = None return else: self.other.sync_target = U1_URL self.other.sync_target = text def toggle_sync(self, value): self.other.auto_sync = value if value: self.other.start_auto_sync() else: self.other.stop_auto_sync() def url_changed(self): if not self.url_radio.isChecked(): return text = unicode(self.url_edit.text(), 'utf-8') if not text: # There was no text in the edit field so do nothing. self.other.sync_target = None return self.other.sync_target = text
[docs]class Main(QtGui.QMainWindow): """Main window of our application.""" def __init__(self, in_memory=False): super(Main, self).__init__() # Dynamically load the ui file generated by QtDesigner. uifile = os.path.join( os.path.abspath(os.path.dirname(__file__)), 'cosas.ui') uic.loadUi(uifile, self) self.buttons_frame.hide() # hook up the signals to the signal handlers. self.connect_events() # Load the cosas database. db = get_database() # And wrap it in a TodoStore object. = TodoStore(db) # create or update the indexes if they are not up-to-date # hook up the delegate header = self.todo_list.header() header.setResizeMode(0, 2) # first column fixed header.setResizeMode(1, 1) # stretch second column header.setResizeMode(2, 2) # third column fixed header.setDefaultSectionSize(20) header.setStretchLastSection(False) # Initialize some variables we will use to keep track of the tags. self._tag_docs = defaultdict(list) self._tag_buttons = {} self._tag_filter = [] self._tag_colors = {} # A list of colors to give differently tagged items different colors. self.colors = TAG_COLORS[:] # Get all the tasks in the database, and add them to the UI. for task in self.add_task(task) self.title_edit.clear() # Give the edit field focus. self.title_edit.setFocus() self.editing = False self.last_synced = None self.sync_target = U1_URL self.auto_sync = False self._timer = QtCore.QTimer() def update_status_bar(self, message): self.statusBar.showMessage(message) def keyPressEvent(self, event): if event.key() == QtCore.Qt.Key_Delete: self.delete() return if event.key() == QtCore.Qt.Key_Return: current = self.todo_list.currentItem() if current and current.task and current.task.has_conflicts: self.open_conflicts_window(current.task.doc_id) return if not self.editing: self.editing = True self.todo_list.openPersistentEditor(current) return else: self.todo_list.closePersistentEditor(current) self.editing = False return super(Main, self).keyPressEvent(event)
[docs] def get_tag_color(self): """Get a color number to use for a new tag.""" # Remove a color from the list of available ones and return it. if not self.colors: return BACKGROUND return self.colors.pop(0)
[docs] def connect_events(self): """Hook up all the signal handlers.""" # On enter, save the task that was being edited. self.title_edit.returnPressed.connect(self.update) self.action_synchronize.triggered.connect(self.open_sync_window) self.buttons_toggle.clicked.connect(self.show_buttons) self.todo_list.itemClicked.connect(self.maybe_open_conflicts)
def maybe_open_conflicts(self, item, column): if not item.task.has_conflicts: return self.open_conflicts_window(item.task.doc_id) def open_sync_window(self): window = Sync(self) window.exec_() def open_conflicts_window(self, doc_id): conflicts = window = Conflicts(self, conflicts) window.exec_()
[docs] def show_buttons(self): """Show the frame with the tag buttons.""" self.buttons_toggle.clicked.disconnect(self.show_buttons) self.buttons_toggle.clicked.connect(self.hide_buttons)
[docs] def hide_buttons(self): """Show the frame with the tag buttons.""" self.buttons_toggle.clicked.disconnect(self.hide_buttons) self.buttons_frame.hide() self.buttons_toggle.clicked.connect(self.show_buttons)
[docs] def refresh_filter(self): """Remove all tasks, and show only those that satisfy the new filter. """ # Remove everything from the list. while self.todo_list.topLevelItemCount(): self.todo_list.takeTopLevelItem(0) # Get the filtered tasks from the database. for task in # Add them to the UI. self.add_task(task) # Clear the current selection. self.todo_list.setCurrentItem(None) self.title_edit.clear() self.item = None
[docs] def update(self): """Either add a new task or update an existing one.""" text = unicode(self.title_edit.text(), 'utf-8') if not text: # There was no text in the edit field so do nothing. return # No task was selected, so add a new one. task =, tags=extract_tags(text)) self.add_task(task) # Clear the current selection. self.title_edit.clear()
[docs] def delete(self): """Delete a todo item.""" # Delete the item from the database. index = self.todo_list.indexFromItem(self.todo_list.currentItem()) item = self.todo_list.takeTopLevelItem(index.row()) if item is None: return # Clear the current selection. self.item = None
[docs] def add_task(self, task): """Add a new todo item.""" # Wrap the task in a UITask object. item = UITask( task, self.todo_list,, self.todo_list.font(), self) if not task.has_conflicts: item.setFlags(item.flags() | QtCore.Qt.ItemIsEditable) self.todo_list.addTopLevelItem(item) if not task.tags: return # If the task has tags, we add them as filter buttons to the UI, if # they are new. for tag in task.tags: self.add_tag(task.doc_id, tag) if task.tags: item.set_color(self._tag_colors[task.tags[0]]['qcolor']) else: item.set_color(BACKGROUND)
[docs] def add_tag(self, doc_id, tag): """Create a link between the task with id doc_id and the tag, and add a new button for tag if it was not already there. """ # Add the task id to the list of document ids associated with this tag. self._tag_docs[tag].append(doc_id) # If the list has more than one element the tag button was already # present. if len(self._tag_docs[tag]) > 1: return # Add a tag filter button for this tag to the UI. button = QtGui.QPushButton(tag) color = self.get_tag_color() qcolor = QtGui.QColor(color) self._tag_colors[tag] = { 'color_tuple': color, 'qcolor': qcolor} button.setStyleSheet('background-color: %s' % color) button._todo_tag = tag # Make the button an on/off button. button.setCheckable(True) # Store a reference to the button in a dictionary so we can find it # back more easily if we need to delete it. self._tag_buttons[tag] = button # We define a function to handle the clicked signal of the button, # since each button will need its own handler. def filter_toggle(checked): """Toggle the filter for the tag associated with this button.""" if checked: # Add the tag to the current filter. self._tag_filter.append(button._todo_tag) else: # Remove the tag from the current filter. self._tag_filter.remove(button._todo_tag) # Apply the new filter. self.refresh_filter() # Attach the handler to the button's clicked signal. button.clicked.connect(filter_toggle) # Get the position where the button needs to be inserted. (We keep them # sorted alphabetically by the text of the tag. index = sorted(self._tag_buttons.keys()).index(tag) # And add the button to the UI. self.buttons_layout.insertWidget(index, button)
[docs] def remove_tag(self, doc_id, tag): """Remove the link between the task with id doc_id and the tag, and remove the button for tag if it no longer has any tasks associated with it. """ # Remove the task id from the list of document ids associated with this # tag. self._tag_docs[tag].remove(doc_id) # If the list is not empty, we do not remove the button, because there # are still tasks that have this tag. if self._tag_docs[tag]: return # Look up the button. button = self._tag_buttons[tag] # Remove it from the ui. button.hide() self.buttons_layout.removeWidget(button) # And remove the reference. del self._tag_buttons[tag]
[docs] def update_tags(self, item, old_tags, new_tags): """Process any changed tags for this item.""" # Process all removed tags. for tag in old_tags - new_tags: self.remove_tag(item.task.doc_id, tag) # Process all tags newly added. for tag in new_tags - old_tags: self.add_tag(item.task.doc_id, tag) if new_tags: item.set_color(self._tag_colors[list(new_tags)[0]]['qcolor']) return item.set_color(BACKGROUND)
def get_ubuntuone_credentials(self): cmt = CredentialsManagementTool() return cmt.find_credentials() def synchronize(self, finalize): if self.sync_target == '': d = self.get_ubuntuone_credentials() d.addCallback(self._synchronize) d.addCallback(finalize) else: # TODO: add ui for entering creds for non u1 servers. self._synchronize() finalize() def _auto_sync(self): self._timer.stop() try: self.synchronize(lambda _: None) finally: self._timer.start(TIMEOUT) def start_auto_sync(self): self._timer.timeout.connect(self._auto_sync) self._timer.start(TIMEOUT) def stop_auto_sync(self): self._timer.stop() def _synchronize(self, creds=None): target = self.sync_target assert target.startswith('http://') or target.startswith('https://') if creds is not None: # convert into expected form creds = {'oauth': { 'token_key': creds['token'], 'token_secret': creds['token_secret'], 'consumer_key': creds['consumer_key'], 'consumer_secret': creds['consumer_secret'] }}, creds=creds) # refresh the UI to show changed or new tasks self.refresh_filter() self.update_status_bar("last synced: %s" % (,)) def resolve(self, doc, revs):, revs) # refresh the UI to show the resolved version self.refresh_filter()
if __name__ == "__main__": # TODO: Unfortunately, to be able to use ubuntuone.platform.credentials on # linux, we now depend on dbus. :( from dbus.mainloop.qt import DBusQtMainLoop main_loop = DBusQtMainLoop(set_as_default=True) app = QtGui.QApplication(sys.argv) main = Main() app.exec_()