1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
#
# Copyright 2009 Martin Owens
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>
#
"""
Provides some basic methods and classes (which should be replaced)
"""
import os
import logging
import threading
from fnmatch import fnmatch
from lazr.uri import URI
from xdgapp import XdgApplication
from launchpadlib.launchpad import EDGE_SERVICE_ROOT
from GroundControl import __stage__, __appname__
def clean_api_url(service_root):
"""Create a reformed API URL"""
web_root_uri = URI(service_root)
web_root_uri.path = ""
web_root_uri.host = web_root_uri.host.replace("api.", "", 1)
return str(web_root_uri.ensureSlash())
if __stage__ == 'DEV':
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
datefmt='%m-%d %H:%M')
else:
# We will log as debug also if the log file exists.
LOGFILE = os.path.expanduser("~/groundcontrol.log")
if os.path.exists(LOGFILE):
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
datefmt='%m-%d %H:%M',
filename=LOGFILE,
filemode='w')
else:
logging.basicConfig(level=logging.ERROR)
PROJECT_NAME = "Launchpad Ground Control"
PROJECT_PKG = __appname__
PROJECT_XDG = XdgApplication(PROJECT_PKG)
LAUNCHPAD_XDG = XdgApplication('launchpad')
LAUNCHPAD_OBJ = None
EDGE_WEB_ROOT = clean_api_url(EDGE_SERVICE_ROOT)
def listfiles(*dirs):
"""List files in a directory given a filtership"""
fdir, pattern = os.path.split(os.path.join(*dirs))
return [os.path.join(fdir, filename)
for filename in os.listdir(os.path.abspath(fdir))
if filename[0] != '.' and fnmatch(filename, pattern)]
class Thread(threading.Thread):
"""Special thread object for catching errors and logging them"""
def run(self, *args, **kwargs):
"""The code to run when the thread us being run"""
try:
super(Thread, self).run(*args, **kwargs)
except Exception, message:
logging.exception(message)
class Events(object):
"""Simple event base class, may be able to replace with gobject"""
signals = None
def call_signal(self, name, *opts, **args):
"""All the named events"""
if self.signals and self.signals.has_key(name):
for signal in self.signals[name].values():
method = signal[0]
opts += signal[1]
args.update(signal[2])
method(*opts, **args)
def connect_signal(self, name, method, *opts, **args):
"""Connect a method to a named event"""
if not self.signals:
self.signals = {}
if not self.signals.has_key(name):
self.signals[name] = {}
# This will add any calls from different objects
# but replace calls of the same method and same object.
self.signals[name][id(method)] = [method, opts, args]
def disconnect_signal(self, name, method):
"""Disconnect a method from a named event"""
if self.signals.has_key(name):
if self.signals[name].has_key(id(method)):
del(self.signals[name][id(method)])
if not self.signals[name]:
del(self.signals[name])
|