starlingx/ptp-notification-armada-app
Code Issues Proposed changes

Merge "Fix log filename where nmea is searched"

Browse Source
This commit is contained in:
Zuul 2022-11-04 18:45:28 +00:00 committed by Gerrit Code Review
commit 1d0a102859
3 changed files with 37 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
notificationservice-base/docker/ptptrackingfunction/trackingfunctionsdk/common/helpers/cgu_handler.py
View File

@ -6,7 +6,6 @@
import logging import logging
import os import os
import re import re
import sys
from trackingfunctionsdk.common.helpers import constants from trackingfunctionsdk.common.helpers import constants
from trackingfunctionsdk.common.helpers import log_helper from trackingfunctionsdk.common.helpers import log_helper
@ -16,7 +15,8 @@ log_helper.config_logger(LOG)
class CguHandler: class CguHandler:
def __init__(self, config_file, nmea_serialport=None, pci_addr=None, cgu_path=None): def __init__(self, config_file, nmea_serialport=None, pci_addr=None,
cgu_path=None):
self.config_file = config_file self.config_file = config_file
self.nmea_serialport = nmea_serialport self.nmea_serialport = nmea_serialport
self.pci_addr = pci_addr self.pci_addr = pci_addr
@ -39,33 +39,38 @@ class CguHandler:
LOG.error(err) LOG.error(err)
raise raise
def convert_nmea_serialport_to_pci_addr(self, dmesg_path="/logs/dmesg"): def convert_nmea_serialport_to_pci_addr(self, log_file="/logs/kern.log"):
# Parse the nmea_serialport value into a PCI address so that we can later find the cgu # Parse the nmea_serialport value into a PCI address so that we can
# later find the cgu
# Returns the address or None # Returns the address or None
pci_addr = None pci_addr = None
# Get only the ttyGNSS_1800_0 portion of the path # Get only the ttyGNSS_1800_0 portion of the path
nmea_serialport = self.nmea_serialport.split('/')[2] nmea_serialport = self.nmea_serialport.split('/')[2]
LOG.debug("Looking for nmea_serialport value: %s" % nmea_serialport) LOG.debug("Looking for nmea_serialport value: %s" % nmea_serialport)
with open(dmesg_path, 'r') as dmesg: with open(log_file, 'r') as file:
for line in dmesg: for line in file:
if nmea_serialport in line: if nmea_serialport in line:
# Regex split to make any number of spaces the delimiter # Regex split to make any number of spaces the delimiter
# Eg. [ 4.834255] ice 0000:18:00.0: ttyGNSS_1800_0 registered # Eg.: ... ice 0000:18:00.0: ttyGNSS_1800_0 registered
# Becomes: 0000:18:00.0 # Becomes: 0000:18:00.0
pci_addr = re.split(' +', line)[3].strip(':') pci_addr = re.split(' +', line)[7].strip(':')
LOG.debug("Found with PCI addr: %s" % pci_addr)
break
self.pci_addr = pci_addr self.pci_addr = pci_addr
return
def get_cgu_path_from_pci_addr(self): def get_cgu_path_from_pci_addr(self):
# Search for a cgu file using the given pci address # Search for a cgu file using the given pci address
cgu_path = "/ice/" + self.pci_addr + "/cgu" cgu_path = "/ice/" + self.pci_addr + "/cgu"
if os.path.exists(cgu_path): if os.path.exists(cgu_path):
LOG.debug("PCI address %s has cgu path %s" % (self.pci_addr, cgu_path)) LOG.debug("PCI address %s has cgu path %s" %
(self.pci_addr, cgu_path))
self.cgu_path = cgu_path self.cgu_path = cgu_path
return return
else: else:
LOG.error("Could not find cgu path for PCI address %s" % self.pci_addr) LOG.error("Could not find cgu path for PCI address %s" %
self.pci_addr)
raise FileNotFoundError raise FileNotFoundError
def read_cgu(self): def read_cgu(self):
@ -75,7 +80,6 @@ class CguHandler:
with open(self.cgu_path, 'r') as infile: with open(self.cgu_path, 'r') as infile:
cgu_output = infile.read() cgu_output = infile.read()
self.cgu_output_raw = cgu_output self.cgu_output_raw = cgu_output
return
def cgu_output_to_dict(self): def cgu_output_to_dict(self):
# Take raw cgu output and parse it into a dict # Take raw cgu output and parse it into a dict
@ -94,19 +98,25 @@ class CguHandler:
for line in cgu_output[7:14]: for line in cgu_output[7:14]:
# Build a dict out of the 7 line table # Build a dict out of the 7 line table
dict_to_insert = {re.split(' +', line)[1]: {'state': re.split(' +', line)[4], dict_to_insert = {
'priority': {'EEC': re.split(' +', line)[6], re.split(' +', line)[1]: {
'PPS': re.split(' +', line)[8]} 'state': re.split(' +', line)[4],
} 'priority': {
} 'EEC': re.split(' +', line)[6],
'PPS': re.split(' +', line)[8]
}
}
}
cgu_dict['input'].update(dict_to_insert) cgu_dict['input'].update(dict_to_insert)
# Add the DPLL data below the table # Add the DPLL data below the table
cgu_dict['EEC DPLL']['Current reference'] = re.split('[ \t]+', cgu_output[16])[3] cgu_dict['EEC DPLL']['Current reference'] = \
re.split('[ \t]+', cgu_output[16])[3]
cgu_dict['EEC DPLL']['Status'] = re.split('[ \t]+', cgu_output[17])[2] cgu_dict['EEC DPLL']['Status'] = re.split('[ \t]+', cgu_output[17])[2]
cgu_dict['PPS DPLL']['Current reference'] = re.split('[ \t]+', cgu_output[20])[3] cgu_dict['PPS DPLL']['Current reference'] = \
re.split('[ \t]+', cgu_output[20])[3]
cgu_dict['PPS DPLL']['Status'] = re.split('[ \t]+', cgu_output[21])[2] cgu_dict['PPS DPLL']['Status'] = re.split('[ \t]+', cgu_output[21])[2]
cgu_dict['PPS DPLL']['Phase offset'] = re.split('[ \t]+', cgu_output[22])[3] cgu_dict['PPS DPLL']['Phase offset'] = \
re.split('[ \t]+', cgu_output[22])[3]
self.cgu_output_parsed = cgu_dict self.cgu_output_parsed = cgu_dict
return

12
notificationservice-base/docker/ptptrackingfunction/trackingfunctionsdk/tests/test_cgu_handler.py
View File

@ -33,14 +33,14 @@ class CguHandlerTests(unittest.TestCase):
def test_convert_nmea_serialport_to_pci_addr(self): def test_convert_nmea_serialport_to_pci_addr(self):
# Test success path # Test success path
self.testCguHandler.get_gnss_nmea_serialport_from_ts2phc_config() self.testCguHandler.get_gnss_nmea_serialport_from_ts2phc_config()
self.testCguHandler.convert_nmea_serialport_to_pci_addr(testpath + self.testCguHandler.convert_nmea_serialport_to_pci_addr(
"test_input_files/mock_dmesg") testpath + "test_input_files/mock_kern.log")
self.assertEqual(self.testCguHandler.pci_addr, "0000:18:00.0") self.assertEqual(self.testCguHandler.pci_addr, "0000:18:00.0")
# Test pci address not found # Test pci address not found
self.testCguHandler.nmea_serialport = "/dev/ttyGNSS_not_present" self.testCguHandler.nmea_serialport = "/dev/ttyGNSS_not_present"
self.testCguHandler.convert_nmea_serialport_to_pci_addr(testpath + self.testCguHandler.convert_nmea_serialport_to_pci_addr(
"test_input_files/mock_dmesg") testpath + "test_input_files/mock_kern.log")
self.assertEqual(self.testCguHandler.pci_addr, None) self.assertEqual(self.testCguHandler.pci_addr, None)
@mock.patch('trackingfunctionsdk.common.helpers.cgu_handler.os.path') @mock.patch('trackingfunctionsdk.common.helpers.cgu_handler.os.path')
@ -48,8 +48,8 @@ class CguHandlerTests(unittest.TestCase):
# Setup mock # Setup mock
mock_path.exists.return_value = True mock_path.exists.return_value = True
self.testCguHandler.get_gnss_nmea_serialport_from_ts2phc_config() self.testCguHandler.get_gnss_nmea_serialport_from_ts2phc_config()
self.testCguHandler.convert_nmea_serialport_to_pci_addr(testpath + self.testCguHandler.convert_nmea_serialport_to_pci_addr(
"test_input_files/mock_dmesg") testpath + "test_input_files/mock_kern.log")
self.testCguHandler.get_cgu_path_from_pci_addr() self.testCguHandler.get_cgu_path_from_pci_addr()
self.assertEqual(self.testCguHandler.cgu_path, "/ice/0000:18:00.0/cgu") self.assertEqual(self.testCguHandler.cgu_path, "/ice/0000:18:00.0/cgu")

6
notificationservice-base/docker/ptptrackingfunction/trackingfunctionsdk/tests/test_input_files/mock_dmesg
View File

@ -1,6 +0,0 @@
[ 4.834255] ice 0000:18:00.0: ttyGNSS_1800_0 registered
[ 4.834258] ice 0000:18:00.0: GNSS TTY init successful
[ 4.846647] ice 0000:18:00.0: <DPLL0> state changed to: locked_ho_ack, pin GNSS-1PPS
[ 4.859247] ice 0000:18:00.0: <DPLL1> state changed to: locked_ho_ack, pin GNSS-1PPS
[ 8.535739] ice 0000:1a:00.0: ttyGNSS_1a00_0 registered
[ 8.535741] ice 0000:1a:00.0: GNSS TTY init successful