BOSWatch/BW3-Core
1
0
Fork 0
mirror of https://github.com/BOSWatch/BW3-Core.git synced 2026-01-07 01:00:58 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

added testmode to bw_client

Browse source
This commit is contained in:
Bastian Schroll 2019-10-24 08:45:18 +02:00
parent 3dddcdcff2
commit 6233e9fbd4
1 changed files with 18 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
bw_client.py
View file

@ -60,7 +60,7 @@ parser = argparse.ArgumentParser(prog="bw_client.py",
epilog="""More options you can find in the extern client.ini
file in the folder /config""")
parser.add_argument("-c", "--config", help="Name to configuration File", required=True)
parser.add_argument("-t", "--test", help="Start Client with testdata-set")
parser.add_argument("-t", "--test", help="Start Client with testdata-set", action="store_true")
args = parser.parse_args()
bwConfig = ConfigYAML()
@ -79,9 +79,6 @@ try:
ip = broadcastClient.serverIP
port = broadcastClient.serverPort
inputQueue = queue.Queue()
inputThreadRunning = True
# ========== INPUT CODE ==========
def handleSDRInput(dataQueue, sdrConfig, decoderConfig): # todo exception handling inside
sdrProc = ProcessManager(str(sdrConfig.get("rtlPath", default="rtl_fm")))
@ -134,13 +131,25 @@ try:
logging.debug("stopping thread")
mmProc.stop()
sdrProc.stop()
# ========== INPUT CODE ==========
mmThread = threading.Thread(target=handleSDRInput, name="mmReader",
args=(inputQueue, bwConfig.get("inputSource", "sdr"), bwConfig.get("decoder")))
mmThread.daemon = True
mmThread.start()
inputQueue = queue.Queue()
if not args.test:
inputThreadRunning = True
mmThread = threading.Thread(target=handleSDRInput, name="mmReader",
args=(inputQueue, bwConfig.get("inputSource", "sdr"), bwConfig.get("decoder")))
mmThread.daemon = True
mmThread.start()
else:
logging.warning("STARTING TESTMODE!")
logging.debug("reading testdata from file")
testFile = open("test/testdata.list", "r")
for testData in testFile:
if (len(testData.rstrip(' \t\n\r')) > 1) and ("#" not in testData[0]):
logging.info("Testdata: %s", testData.rstrip(' \t\n\r'))
inputQueue.put_nowait((testData.rstrip(' \t\n\r'), time.time()))
logging.debug("finished reading testdata")
bwClient = TCPClient()
bwClient.connect(ip, port)