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

Finished implementing unit tests and fixed bugs found.

Browse Source
This commit is contained in:
Jason Watkins
2015-05-01 13:50:48 -07:00
parent 685d121edd
commit 10d21cc0df
2 changed files with 229 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
Python/src/xplaneConnect.py
View File

@@ -1,8 +1,9 @@
import socket
import struct
class XPlaneConnect:
class XPlaneConnect(object):
'''XPlaneConnect (XPC) facilitates communication to and from the XPCPlugin.'''
socket = None
# Basic Functions
def __init__(self, xpHost = 'localhost', xpPort = 49009, port = 0, timeout = 100):
@@ -71,7 +72,7 @@ class XPlaneConnect:
if port < 0 or port > 65535:
raise ValueError("The specified port is not a valid port number.")
buffer = struct.pack("<4sxH", ("CONN", port))
buffer = struct.pack("<4sxH", "CONN", port)
self.sendUDP(buffer)
def pauseSim(self, pause):
@@ -84,7 +85,7 @@ class XPlaneConnect:
if pause:
pause_val = 1
buffer = struct.pack("<4sxB", ("SIMU", pause_val))
buffer = struct.pack("<4sxB", "SIMU", pause_val)
self.sendUDP(buffer)
# X-Plane UDP Data
@@ -116,11 +117,11 @@ class XPlaneConnect:
if len(data) > 134:
raise ValueError("Too many rows in data.")
buffer = struct.pack("4sx", "DATA")
buffer = struct.pack("<4sxB", "DATA", len(data))
for row in data:
if len(row) != 9:
raise ValueError("Row does not contain exactly 9 values. <" + str(row) + ">")
struct.pack_into("<I8f", buffer, len(buffer), row)
buffer += struct.pack("<I8f", *row)
self.sendUDP(buffer)
# Position
@@ -148,12 +149,12 @@ class XPlaneConnect:
raise ValueError("Aircraft number must be between 0 and 20.")
# Pack message
buffer = struct.pack("<4sxB", ("POSI", ac))
buffer = struct.pack("<4sxB", "POSI", ac)
for i in range(7):
val = -998
if i < len(values):
val = values[i]
struct.pack_into("<f", buffer, len(buffer), val)
buffer += struct.pack("<f", val)
# Send
self.sendUDP(buffer)
@@ -187,8 +188,12 @@ class XPlaneConnect:
val = -998
if i < len(values):
val = values[i]
struct.pack_into("<f", buffer, len(buffer), val)
struct.pack_into("B", buffer, len(buffer), ac)
if i == 4:
val = -1 if val == -998 else val
buffer += struct.pack("B", val)
else:
buffer += struct.pack("<f", val)
buffer += struct.pack("B", ac)
# Send
self.sendUDP(buffer)
@@ -282,7 +287,7 @@ class XPlaneConnect:
msg = ""
msgLen = len(msg)
buffer = struct.pack("<4sxiiB" + msgLen + "s", ("TEXT", x, y, msgLen, msg))
buffer = struct.pack("<4sxiiB" + str(msgLen) + "s", "TEXT", x, y, msgLen, msg)
self.sendUDP(buffer)
def sendWYPT(self, op, points):
@@ -304,5 +309,8 @@ class XPlaneConnect:
if len(points) / 3 > 255:
raise ValueError("Too many points. You can only send 255 points at a time.")
buffer = struct.pack("<4sxBB" + len(points) + "f", ("WYPT", op, len(points), points))
if op == 3:
buffer = struct.pack("<4sxBB", "WYPT", 3, 0)
else:
buffer = struct.pack("<4sxBB" + str(len(points)) + "f", "WYPT", op, len(points), *points)
self.sendUDP(buffer)

250
TestScripts/Python Tests/Tests.py
View File

@@ -1,65 +1,66 @@
import random
import unittest
import imp
import time
xplaneConnect = imp.load_source('xplaneConnect', '../../Python/src/xplaneConnect.py')
xpc = imp.load_source('xplaneConnect', '../../Python/src/xplaneConnect.py')
class XPCTests(unittest.TestCase):
"""Tests the functionality of the XPlaneConnect class."""
def test_init(self):
try:
xpc = xplaneConnect.XPlaneConnect()
client = xpc.XPlaneConnect()
except:
self.fail("Default constructor failed.")
try:
xpc = xplaneConnect.XPlaneConnect("I'm not a real host")
client = xpc.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)
client = xpc.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)
client = xpc.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)
client = xpc.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)
client = xpc.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)
client = xpc.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)
client = xpc.XPlaneConnect("127.0.0.1", 49009, 49063)
client.close()
self.assertIsNone(client.socket)
client = xpc.XPlaneConnect("127.0.0.1", 49009, 49063)
def test_send_read(self):
# Init
test = "\x00\x01\x02\x03\x05"
# Setup
sender = xplaneConnect.XPlaneConnect("127.0.0.1", 49063, 49064)
receiver = xplaneConnect.XPlaneConnect("127.0.0.1", 49009, 49063)
sender = xpc.XPlaneConnect("127.0.0.1", 49063, 49064)
receiver = xpc.XPlaneConnect("127.0.0.1", 49009, 49063)
# Execution
sender.sendUDP(test)
@@ -75,15 +76,15 @@ class XPCTests(unittest.TestCase):
def test_getDREFs(self):
# Setup
xpc = xplaneConnect.XPlaneConnect()
client = xpc.XPlaneConnect()
drefs = ["sim/cockpit/switches/gear_handle_status",\
"sim/cockpit2/switches/panel_brightness_ratio"]
# Execution
result = xpc.getDREFs(drefs)
result = client.getDREFs(drefs)
# Cleanup
xpc.close()
client.close()
# Tests
self.assertEqual(2, len(result))
@@ -91,46 +92,207 @@ class XPCTests(unittest.TestCase):
self.assertEqual(4, len(result[1]))
def test_sendDREF(self):
# Setup
xpc = xplaneConnect.XPlaneConnect()
dref = "sim/cockpit/switches/gear_handle_status"
value = None
def do_test():
# Setup
client = xpc.XPlaneConnect()
# Execution
xpc.sendDREF(dref, 0.0)
result = xpc.getDREF(dref)
# Execute
client.sendDREF(dref, value)
result = client.getDREF(dref)
# Cleanup
xpc.close()
client.close()
# Tests
self.assertEqual(1, len(result))
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()
self.assertEqual(value, result[0])
# Test 1
self.assertAlmostEqual(1.0, result[0])
value = 1
do_test()
# Execution 2
xpc = xplaneConnect.XPlaneConnect()
xpc.pauseSim(False)
result = xpc.getDREF(dref)
# Test 2
value = 0
do_test()
# Cleanup 2
xpc.close()
def test_sendDATA(self):
# Setup
dref = "sim/aircraft/parts/acf_gear_deploy"
data = [[ 14, 1, 0, -998, -998, -998, -998, -998, -998 ]]
client = xpc.XPlaneConnect()
# Execute
client.sendDATA(data)
result = client.getDREF(dref)
# Cleanup
client.close()
#Tests
self.assertEqual(result[0], data[0][1])
def test_pauseSim(self):
dref = "sim/operation/override/override_planepath"
value = None
expected = None
def do_test():
# Setup
client = xpc.XPlaneConnect()
# Execute
client.pauseSim(value)
result = client.getDREF(dref)
# Cleanup
client.close()
# Test
self.assertAlmostEqual(expected, result[0])
# Test 1
value = True
expected = 1.0
do_test()
# Test 2
value = False
expected = 0.0
do_test()
def test_sendCTRL(self):
# Setup
drefs = ["sim/cockpit2/controls/yoke_pitch_ratio",\
"sim/cockpit2/controls/yoke_roll_ratio",\
"sim/cockpit2/controls/yoke_heading_ratio",\
"sim/flightmodel/engine/ENGN_thro",\
"sim/cockpit/switches/gear_handle_status",\
"sim/flightmodel/controls/flaprqst"]
ctrl = []
def do_test():
client = xpc.XPlaneConnect()
# Execute
client.sendCTRL(ctrl)
result = client.getDREFs(drefs)
# Cleanup
client.close()
# Tests
self.assertEqual(0.0, result[0])
self.assertEqual(6, len(result))
for i in range(6):
self.assertAlmostEqual(ctrl[i], result[i][0], 4)
# Test 1
ctrl = [ -1.0, -1.0, -1.0, 0.0, 1.0, 1.0 ]
do_test()
# Test 2
ctrl = [ 1.0, 1.0, 1.0, 0.0, 1.0, 0.5 ]
do_test()
# Test 2
ctrl = [ 0.0, 0.0, 0.0, 0.8, 1.0, 0.0 ]
do_test()
def test_sendPOSI(self):
# Setup
drefs = ["sim/flightmodel/position/latitude",\
"sim/flightmodel/position/longitude",\
"sim/flightmodel/position/elevation",\
"sim/flightmodel/position/theta",\
"sim/flightmodel/position/phi",\
"sim/flightmodel/position/psi",\
"sim/cockpit/switches/gear_handle_status"]
posi = None
def do_test():
client = xpc.XPlaneConnect()
# Execute
client.pauseSim(True)
client.sendPOSI(posi)
result = client.getDREFs(drefs)
client.pauseSim(False)
# Cleanup
client.close()
# Tests
self.assertEqual(7, len(result))
for i in range(7):
self.assertAlmostEqual(posi[i], result[i][0], 4)
# Test 1
posi = [ 37.524, -122.06899, 2500, 5, 7, 11, 1 ]
do_test()
# Test 2
posi = [ 38, -121.0, 2000, -10, 0, 0, 0 ]
do_test()
def test_sendTEXT(self):
# Setup
client = xpc.XPlaneConnect()
x = 200
y = 700
msg = "Python sendTEXT test message."
# Execution
client.sendTEXT(msg, x, y)
# NOTE: Manually verify that msg appears on the screen in X-Plane
# Cleanup
client.close()
def test_sendWYPT(self):
# Setup
client = xpc.XPlaneConnect()
points = [\
37.5245, -122.06899, 2500,\
37.455397, -122.050037, 2500,\
37.469567, -122.051411, 2500,\
37.479376, -122.060509, 2300,\
37.482237, -122.076130, 2100,\
37.474881, -122.087288, 1900,\
37.467660, -122.079391, 1700,\
37.466298, -122.090549, 1500,\
37.362562, -122.039223, 1000,\
37.361448, -122.034416, 1000,\
37.361994, -122.026348, 1000,\
37.365541, -122.022572, 1000,\
37.373727, -122.024803, 1000,\
37.403869, -122.041283, 50,\
37.418544, -122.049222, 6]
# Execution
client.sendPOSI([37.5245, -122.06899, 2500])
client.sendWYPT(3, [])
client.sendWYPT(1, points)
# NOTE: Manually verify that points appear on the screen in X-Plane
# Cleanup
client.close()
def test_setCONN(self):
# Setup
dref = "sim/cockpit/switches/gear_handle_status";
client = xpc.XPlaneConnect()
# Execute
client.setCONN(49055)
result = client.getDREF(dref)
# Cleanup
client.close()
# Test
self.assertEqual(1, len(result))
if __name__ == '__main__':
unittest.main()