src/lamp_core_base.py

   1 #!/usr/bin/env python
   2 #-*- coding: utf-8 -*-
   3
   4 #=========================
   5 # Filename:  lamp_core_base.py
   6 # Author: Rodrigo J. S. Peixoto
   7 # Date: 2008-08-21
   8 #--------------------------------------------------
   9 # Description:
  10 # It's responsible for the application management.
  11 #--------------------------------------------------
  12 # License:
  13 #
  14 #==================================================
  15 from lamp_application_manager import LampApplicationManager
  16 from lamp_application_manager import LampExecutionQueueIsEmptyError
  17 from lamp_application_manager import LampExecutionQueueIsFullError
  18 from lamp_hardware_manager import LampHardwareManager
  19 from lamp_hardware_manager import VolumeIsNotMountedError
  20 from lamp_hardware_manager import VolumeAlreadyMountedError
  21 from lamp_bluetooth_manager import LampBluetoothManager
  22
  23 from lamp_utils.lamp_built_in_functions import check_types
  24 from lamp_utils.lamp_built_in_functions import get_parent_path
  25 from lamp_utils.lamp_recents_manager import LampRecentsManager
  26 from lamp_utils.lamp_recents_manager import DirectoryNotFoundError
  27 from lamp_utils.lamp_recents_manager import RecentsCannotBeCreatedError
  28 import os
  29 import re
  30 import sys
  31 import time
  32 import subprocess
  33
  34
  35 global LAMPEJO_CORE_PATH
  36 LAMPEJO_CORE_PATH=os.getenv("LAMPEJO_CORE_PATH")
  37 if LAMPEJO_CORE_PATH:
  38     sys.path.append(LAMPEJO_CORE_PATH)
  39 else:
  40     raise SystemExit("Error, the environment variable LAMPEJO_CORE_PATH is not defined.")
  41
  42 PATH_TO_CFG_FILE=os.path.join(get_parent_path(LAMPEJO_CORE_PATH), 'config', 'lampejo.cfg')
  43 if not os.path.exists(PATH_TO_CFG_FILE):
  44     raise SystemExit("Error, the lampejo.cfg file is not found in the path: %s" % PATH_TO_CFG_FILE)
  45 IMAGE_APP=u"eog"
  46 AUDIO_VIDEO=u"kaffeine"
  47 DOCUMENTS_PDF_PS=u"kpdf"
  48 DUCOMENTS=u"soffice"
  49
  50 class InvalidStatementError(Exception): pass
  51
  52 class LampCoreBase:
  53     '''
  54     Tests:
  55     >>> test = LampCoreBase()
  56     '''
  57     def __init__(self):
  58         self.recents_manager = LampRecentsManager()
  59         self.app_manager = LampApplicationManager(PATH_TO_CFG_FILE)
  60         self.hardware_manager = LampHardwareManager()
  61         self.bluetooth_manager = LampBluetoothManager()
  62         self.bluetooth_manager.start_service()
  63         self.app_manager.set__update_control_image_callback(self.bluetooth_manager.change_control_image)
  64         self.__error_callback = None
  65         self.__hardware_insertion_notify_callback = None
  66         #INFO: Para utilizar no lampejo deve-se remover o comentário da linha abaixo
  67         #self.mouse_config()
  68
  69     def mouse_config(self):
  70         print "Configure mouse buttons..."
  71         try:
  72             subprocess.call(["xmodmap", "-e","pointer = 1 116 115"])
  73         except:
  74             print "DEBUG: Error: mouse cannot be configured by core."
  75
  76     def shutdown(self):
  77         self.app_manager.shutdown()
  78         self.bluetooth_manager.shutdown_service()
  79         #self.hardware_manager.s
  80
  81         #self.recents_manager.s
  82         pass
  83
  84     def decode_statement(self, data):
  85         '''
  86         Tests:
  87         >>> import time
  88         >>> test = LampCoreBase()
  89         >>> test.decode_statement("#exec firefox#")
  90         >>> test.decode_statement("#exec firefox##exec firefox#")
  91         Traceback (most recent call last):
  92             ...
  93         InvalidStatementError: The statement is no valid!
  94         >>> test.decode_statement("#exec firefox##exec firefox##exec fir")
  95         Traceback (most recent call last):
  96             ...
  97         InvalidStatementError: The statement is no valid!
  98         >>> test.decode_statement("#open ../media_test/presentation.odp#")
  99         >>> time.sleep(2)
 100         >>> test.switch_application()
 101         >>> time.sleep(2)
 102         >>> test.switch_application()
 103         >>> time.sleep(2)
 104         >>> test.switch_application()
 105         >>> time.sleep(2)
 106         >>> test.switch_application()
 107         >>> test.close_application()
 108         >>> test.close_application()
 109         '''
 110         print "DEBUG: from flash -> ", data
 111         if data == "#started#":
 112             for dev in self.hardware_manager.devices:
 113                 time.sleep(0.5)
 114                 self.hardware_manager.hardware_modify_notification(added=True, device=dev)
 115         else:
 116             #FUnciona => re.match(r"^#(exec|open) ([a-zA-Z0-9-_.]+|[a-zA-Z0-9. _/-]+)#$", data) and \
 117             data = data and \
 118                re.match(r"^#(exec|open|save|delete|close) .*#$", data) and \
 119                data[1:-1] or None
 120             if data:
 121                 #test = re.compile(r"^((?P<e>exec)|(?P<o>open)) (?(e)[a-zA-Z0-9-_ .]+)(?(o)[a-zA-Z0-9. _/-]+)$")
 122                 #command_content = data and test.match(data) and re.sub("(exec |open )", "", data) or None
 123                 command_content = re.sub("(exec |open |save |delete |close )", "", data) or None
 124                 do_action = {"exec": self.execute_application,
 125                              "open": self.open_file,
 126                              "save": self.save_dir,
 127                              "delete":self.delete_dir,
 128                              "close":self.close_window_by_flash}
 129                 action =  data[:data.index(" ")]
 130                 if type(command_content) == type(u""):
 131                     do_action[action](command_content)
 132                 else:
 133                     do_action[action](u'"%s"' % unicode(command_content, "UTF-8"))
 134             else:
 135                 print "DEBUG: Warning the statement (%s) is not valid!" % data
 136                 #raise InvalidStatementError("The statement is no valid!")
 137
 138
 139     def execute_application(self, command_content):
 140         try:
 141             command_content and self.app_manager.run_application(command_content)
 142         except LampExecutionQueueIsFullError, e:
 143             self.show_error(e)
 144
 145     def open_file(self, command_content):
 146         try:
 147             app_alias = command_content and self.check_extension(command_content)
 148             command_content and app_alias and self.app_manager.run_application(app_alias, command_content)
 149         except LampExecutionQueueIsFullError, e:
 150             self.show_error(e)
 151
 152     def save_dir(self, command_content):
 153         pendrive = self.hardware_manager.get_storage_device()
 154         if pendrive:
 155             mp = pendrive.mount_point()
 156             print "Save dir: %s to %s!" % (command_content, mp)
 157             #TODO: Basta descomentar que está pronto!
 158             self.recents_manager.save_dir(command_content, mp)
 159         else:
 160             print "DEBUG: Saving error -> pendrive not found!"
 161
 162     def delete_dir(self, command_content):
 163         print "Delete dir: %s!" % command_content
 164         #TODO: Basta descomentar que está pronto!
 165         self.recents_manager.remove_contents_of_dir(command_content)
 166
 167     def close_window_by_flash(self, command_content):
 168         print "Close window: %s" % command_content
 169         #TODO: Basta descomentar que está pronto!
 170         self.close_application()
 171
 172     def set_error_callback(self, callback):
 173         if callback and check_types({callback: type(lambda:None)}):
 174             self.__error_callback = callback
 175
 176     def show_error(self, message):
 177         '''This is temporary!!! It depends the zenity application.
 178         Tests:
 179         >>> test = LampCoreBase()
 180         >>> def tmp(msg, call): print msg
 181         >>> test.show_error("Test1")
 182         >>> test.show_error("Test2")
 183         >>> test.show_error("Test3")
 184         '''
 185         if self.__error_callback:
 186             self.__error_callback(message)
 187         else:
 188             print "DEBUG: Error -> __error_callback not valid!"
 189             print "DEBUG: message -> %s" % message
 190
 191     def check_extension(self, file):
 192         '''
 193         Tests:
 194         >>> test = LampCoreBase()
 195         >>> test.check_extension("/home/lamp/test/image.png")
 196         u'eog'
 197         >>> test.check_extension("/home/lamp/test/music.ogg")
 198         u'totem'
 199         >>> test.check_extension("/home/lamp/test/document.odt")
 200         u'ooffice'
 201         >>> test.check_extension("/home/lamp/test/document.pdf")
 202         u'evince'
 203         >>> test.check_extension("/home/lamp/test/document.pddsss")
 204         '''
 205         ret = None
 206         apps = {DOCUMENTS_PDF_PS: [".pdf", ".ps"],
 207                 DUCOMENTS: [".odt", ".ods", ".odp", ".doc", ".xls", ".ppt", ".pps",".docx", ".xlsx", ".pptx"],
 208                 IMAGE_APP: [".jpg", ".png", ".bmp", ".gif"],
 209                 AUDIO_VIDEO: [".mp3",".ogg", ".dvd", ".avi", ".mpeg",".mpg", ".mov", ".wmv",".wav", ".wma"],
 210                 u"firefox": ["swf", "gif"]}
 211         file_name, file_ext = os.path.splitext(file)
 212         for app, exts in apps.items():
 213             if file_ext in exts:
 214                 ret = app
 215                 break
 216         if not ret:
 217             print "DEBUG: Warning extension %s not supported yet!" % file_ext
 218         return ret
 219
 220     def show_home(self):
 221         self.app_manager.show_home()
 222
 223     def switch_application(self):
 224         self.app_manager.switch_application()
 225
 226     def close_application(self):
 227         try:
 228             self.app_manager.kill_current_application()
 229         except LampExecutionQueueIsEmptyError:
 230             message = "Info: Trying to close an application; There is not application running now."
 231             self.show_error(message)
 232
 233     def mount_pendrive(self):
 234         '''
 235
 236         Tests:
 237         >>> test = LampCoreBase()
 238         >>> test.mount_pendrive()
 239         '''
 240         for dev in self.hardware_manager.devices:
 241             try:
 242                 if dev.who_am_i() == "storage":
 243                     return dev.mount()
 244                 else:
 245                     self.show_error("Warning: There is not a pendrive!")
 246             except VolumeAlreadyMountedError:
 247                 self.show_error("Warning: The pendrive is already mounted!")
 248
 249
 250     def unmount_pendrive(self):
 251         '''
 252
 253         Tests:
 254         >>> test = LampCoreBase()
 255         >>> test.unmount_pendrive()
 256         '''
 257         for dev in self.hardware_manager.devices:
 258             try:
 259                 if dev.who_am_i() == "storage":
 260                     return dev.eject()
 261                     #self.hardware_manager.hardware_modify_notification(added=False, device=dev)
 262                 else:
 263                     self.show_error("Warning: There is not a pendrive!")
 264             except VolumeIsNotMountedError:
 265                 self.show_error("Warning: The pendrive is already unmounted!")
 266
 267
 268 def _test():
 269     import doctest
 270     doctest.testmod()
 271
 272
 273 if __name__ == '__main__':
 274     _test()