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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
from telnetlib import Telnet
import sys
import socket
import re
from string import split, join
import time
__all__ = ["FlightGear"]
CRLF = '\r\n'
class FGTelnet(Telnet):
def __init__(self,host,port):
Telnet.__init__(self,host,port)
self.prompt = []
self.prompt.append( re.compile('/[^>]*> ') )
self.timeout = 5
#Telnet.set_debuglevel(self,2)
def help(self):
return
def ls(self,dir=None):
"""
Returns a list of properties.
"""
if dir == None:
self._putcmd('ls')
else:
self._putcmd('ls %s' % dir )
return self._getresp()
def dump(self):
"""Dump current state as XML."""
self._putcmd('dump')
return self._getresp()
def cd(self, dir):
"""Change directory."""
self._putcmd('cd ' + dir)
self._getresp()
return
def pwd(self):
"""Display current path."""
self._putcmd('pwd')
return self._getresp()
def get(self,var):
"""Retrieve the value of a parameter."""
self._putcmd('get %s' % var )
return self._getresp()
def set(self,var,value):
"""Set variable to a new value"""
self._putcmd('set %s %s' % (var,value))
self._getresp() # Discard response
def quit(self):
"""Terminate connection"""
self._putcmd('quit')
self.close()
return
# Internal: send one command to FlightGear
def _putcmd(self,cmd):
cmd = cmd + CRLF;
Telnet.write(self, cmd)
return
# Internal: get a response from FlightGear
def _getresp(self):
(i,match,resp) = Telnet.expect(self, self.prompt, self.timeout)
# Remove the terminating prompt.
# Everything preceding it is the response.
return split(resp, '\n')[:-1]
class FlightGear:
"""FlightGear interface class.
An instance of this class represents a connection to a FlightGear telnet
server.
Properties are accessed using a dictionary style interface:
For example:
# Connect to flightgear telnet server.
fg = FlightGear('myhost', 5500)
# parking brake on
fg['/controls/gear/brake-parking'] = 1
# Get current heading
heading = fg['/orientation/heading-deg']
Other non-property related methods
"""
def __init__( self, host = 'localhost', port = 5500 ):
try:
self.telnet = FGTelnet(host,port)
except socket.error, msg:
self.telnet = None
raise socket.error, msg
def __del__(self):
# Ensure telnet connection is closed cleanly.
self.quit();
def __getitem__(self,key):
"""Get a FlightGear property value.
Where possible the value is converted to the equivalent Python type.
"""
s = self.telnet.get(key)[0]
match = re.compile( '[^=]*=\s*\'([^\']*)\'\s*([^\r]*)\r').match( s )
if not match:
return None
value,type = match.groups()
#value = match.group(1)
#type = match.group(2)
if value == '':
return None
if type == '(double)':
return float(value)
elif type == '(int)':
return int(value)
elif type == '(bool)':
if value == 'true':
return 1
else:
return 0
else:
return value
def __setitem__(self, key, value):
"""Set a FlightGear property value."""
self.telnet.set( key, value )
def quit(self):
"""Close the telnet connection to FlightGear."""
if self.telnet:
self.telnet.quit()
self.telnet = None
|
