FlightSim-Streaming/XPlaneConnectCSP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed several bugs found by unit tests.

Browse Source
This commit is contained in:
Jason Watkins
2015-04-28 11:54:48 -07:00
parent 31095b716b
commit 685d121edd
3 changed files with 89 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -6,6 +6,7 @@
*.exe
*.o
*.so
*.pyc
# Logs and databases #
######################

29
Python/src/xplaneConnect.py
View File

@@ -30,13 +30,14 @@ class XPlaneConnect:
raise ValueError("timeout must be non-negative.")
# Setup XPlane IP and port
self.xpDst = (socket.inet_pton(socket.AF_INET, xpIP), xpPort)
self.xpDst = (xpIP, xpPort)
# Create and bind socket
clientAddr = (socket.INADDR_ANY, port)
clientAddr = ("0.0.0.0", port)
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
self.socket.bind(clientAddr)
self.socket.settimeout(timeout / 1000)
timeout /= 1000.0
self.socket.settimeout(timeout)
def __del__(self):
self.close()
@@ -212,11 +213,11 @@ class XPlaneConnect:
if hasattr(values, "__len__"):
if len(values) > 255:
raise ValueError("values must have less than 256 items.")
fmt = "<4sxB" + len(dref) + "sB" + len(values) + "f"
buffer = struct.pack(fmt, ("DREF", len(dref), dref, len(values), values))
fmt = "<4sxB{0:d}sB{1:d}f".format(len(dref), len(values))
buffer = struct.pack(fmt, "DREF", len(dref), dref, len(values), values)
else:
fmt = "<4sxB" + len(dref) + "sBf"
buffer = struct.pack(fmt, ("DREF", len(dref), dref, 1, values))
fmt = "<4sxB{0:d}sBf".format(len(dref))
buffer = struct.pack(fmt, "DREF", len(dref), dref, 1, values)
# Send
self.sendUDP(buffer)
@@ -229,7 +230,7 @@ class XPlaneConnect:
Returns: A sequence of data representing the values of the requested dataref.
'''
return getDREFS([dref])[0]
return self.getDREFs([dref])[0]
def getDREFs(self, drefs):
'''Gets the value of one or more X-Plane datarefs.
@@ -241,20 +242,22 @@ class XPlaneConnect:
datarefs.
'''
# Send request
buffer = struct.pack("<4sxB" ("GETD", len(drefs)))
buffer = struct.pack("<4sxB", "GETD", len(drefs))
for dref in drefs:
struct.pack_into("<B" + len(dref) + "s", buffer, len(buffer), dref)
fmt = "<B{0:d}s".format(len(dref))
buffer += struct.pack(fmt, len(dref), dref)
self.sendUDP(buffer)
# Read and parse response
buffer = self.readUDP()
resultCount = struct.unpack_from("B", buffer, 5)
resultCount = struct.unpack_from("B", buffer, 5)[0]
offset = 6
result = []
for i in range(resultCount):
rowLen = struct.unpack_from("B", buffer, offset)
rowLen = struct.unpack_from("B", buffer, offset)[0]
offset += 1
row = struct.unpack_from("<" + rowLen + "f", buffer, offset)
fmt = "<{0:d}f".format(rowLen)
row = struct.unpack_from(fmt, buffer, offset)
result.append(row)
offset += rowLen * 4
return result

83
TestScripts/Python Tests/Tests.py
View File

@@ -8,15 +8,50 @@ class XPCTests(unittest.TestCase):
"""Tests the functionality of the XPlaneConnect class."""
def test_init(self):
xpc = xplaneConnect.XPlaneConnect()
self.assertTrue(True)
try:
xpc = xplaneConnect.XPlaneConnect()
except:
self.fail("Default constructor failed.")
try:
xpc = xplaneConnect.XPlaneConnect("I'm not a real host")
self.fail("Failed to catch invalid XP host.")
except ValueError:
pass
try:
xpc = xplaneConnect.XPlaneConnect("127.0.0.1", 90001)
self.fail("Failed to catch invalid XP port.")
except ValueError:
pass
try:
xpc = xplaneConnect.XPlaneConnect("127.0.0.1", -1)
self.fail("Failed to catch invalid XP port.")
except ValueError:
pass
try:
xpc = xplaneConnect.XPlaneConnect("127.0.0.1", 49009, 90001)
self.fail("Failed to catch invalid local port.")
except ValueError:
pass
try:
xpc = xplaneConnect.XPlaneConnect("127.0.0.1", 49009, -1)
self.fail("Failed to catch invalid XP port.")
except ValueError:
pass
try:
xpc = xplaneConnect.XPlaneConnect("127.0.0.1", 49009, 0, -1)
self.fail("Failed to catch invalid timeout.")
except ValueError:
pass
def test_close(self):
xpc = xplaneConnect.XPlaneConnect("127.0.0.1", 49009, 49063)
xpc.close()
self.assertIsNone(xpc.socket)
xpc = xplaneConnect.XPlaneConnect("127.0.0.1", 49009, 49063)
self.assertTrue(True)
def test_send_read(self):
# Init
@@ -38,10 +73,11 @@ class XPCTests(unittest.TestCase):
for a, b in zip(test, buf):
self.assertEqual(a, b)
def test_request_dref(self):
def test_getDREFs(self):
# Setup
xpc = xplaneConnect.XPlaneConnect()
drefs = ["sim/cockpit/switches/gear_handle_status", "cockpit2/switches/panel_brightness_ratio"]
drefs = ["sim/cockpit/switches/gear_handle_status",\
"sim/cockpit2/switches/panel_brightness_ratio"]
# Execution
result = xpc.getDREFs(drefs)
@@ -54,22 +90,47 @@ class XPCTests(unittest.TestCase):
self.assertEqual(1, len(result[0]))
self.assertEqual(4, len(result[1]))
def test_send_dref(self):
def test_sendDREF(self):
# Setup
xpc = xplaneConnect.XPlaneConnect()
dref = "sim/cockpit/switches/gear_handle_status"
# Execution
sender.sendDREF(dref_array)
result = receiver.getDREF(dref_array)
xpc.sendDREF(dref, 0.0)
result = xpc.getDREF(dref)
# Cleanup
xpc.close();
xpc.close()
# Tests
self.assertEqual(1, len(result))
self.assertEqual(1, len(result[0]))
self.assertEqual(0.0, result[0][0])
self.assertEqual(0.0, result[0])
def test_pauseSim(self):
# Setup
xpc = xplaneConnect.XPlaneConnect()
dref = "sim/operation/override/override_planepath"
# Execution 1
xpc.pauseSim(True)
result = xpc.getDREF(dref)
# Cleanup 1
xpc.close()
# Test 1
self.assertAlmostEqual(1.0, result[0])
# Execution 2
xpc = xplaneConnect.XPlaneConnect()
xpc.pauseSim(False)
result = xpc.getDREF(dref)
# Cleanup 2
xpc.close()
# Tests
self.assertEqual(0.0, result[0])
if __name__ == '__main__':
unittest.main()