starlingx/test
Code Issues Proposed changes

Adjusting PTP Config for Connectors

Browse Source 
Change-Id: I923cf6b278e29fbeb99130fdee8923371818cc16
Signed-off-by: croy <Christian.Roy@windriver.com>
This commit is contained in:
croy 2025-03-19 14:41:06 -04:00
parent 6f75a949a3
commit de29aad472
11 changed files with 639 additions and 179 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@ -64,7 +64,7 @@ repos:
entry: pydoclint
language: python
types: [python]
args: ["--config=pyproject.toml"]
args: [--config=pyproject.toml]
- id: interrogate
name: interrogate - ensure all functions and classes have docstrings

161
config/ptp/files/default.json5
View File

@ -1,89 +1,110 @@
{
controller_0 : {
nic1 : {
// GNSS server information if you want to turn on/off GNSS ports.
gnss_server_host: "gnss_server_hostname_or_ip",
gnss_server_username: "gnss_server_username",
gnss_server_password: "gnss_server_password",
gnss_switch_port: "./gnss_switch_port_file.exp 1",
// Information about the NIC cards and SMA connections.
hosts: {
// Parameters can be : sma1 input/output, sma2 input/output, u.fl1 output, u.fl2 input, synce_rclka enabled, synce_rclkb enabled
sma1_to_nic2: "output", // CLOCK OUTPUT INTERFACE Controller-0 NIC1
sma2_to_nic2: "",
controller_0 : {
// This port for TS2PHC and PHC2SYS.
// Can be assigned to any interface of the NIC connected to GNSS and with any NIC on the same host.
// By default, we use the first port.
base_port: "enp81s0f0",
conn_to_ctrl1_nic1: "enp81s0f1",
sma_connectors: {
// Optional Spirent config.
conn_to_spirent: "",
spirent_port: "",
nic_1_2: {
input: { nic: "nic1", sma: "sma1" },
output:{ nic: "nic2", sma: "sma2" },
},
nic_2_3: {
input: { nic: "nic2", sma: "sma1" },
output:{ nic: "nic3", sma: "sma1" },
},
},
ufl_connectors: {
nic_1_3: {
input: { nic: "nic1", ufl: "ufl1" },
output:{ nic: "nic2", ufl: "ufl1" },
},
},
nics: {
nic1 : {
// This port for TS2PHC and PHC2SYS.
// Can be assigned to any interface of the NIC connected to GNSS and with any NIC on the same host.
// By default, we use the first port.
base_port: "enp81s0f0",
gpio_switch_port: "30",
pci_slot: "00:007.01",
nic_connection: {
to_host: "controller_1",
to_nic: "nic1",
interface: "enp81s0f1"
},
// Optional Spirent config.
conn_to_spirent: "conn_spirent_placeholder",
spirent_port: "1234",
},
nic2 : {
base_port: "",
gpio_switch_port: "",
conn_to_spirent: "",
spirent_port: "",
},
nic3 : {
base_port: "",
gpio_switch_port: "",
conn_to_spirent: "",
spirent_port: "",
}
},
},
nic2 : {
controller_1 : {
gnss_switch_port: "",
sma_connectors: {
// Parameters can be : sma1 input/output, sma2 input/output, u.fl1 output, u.fl2 input, synce_rclka enabled, synce_rclkb enabled
sma1_to_nic1: "input", // CLOCK OUTPUT INTERFACE Controller-0 NIC2
sma2_to_nic1: "",
nic_1_2: {
input: { nic: "nic1", sma: "sma1" },
output:{ nic: "nic2", sma: "sma2" },
},
// This port for TS2PHC and PHC2SYS.
// Can be assigned to any interface of the NIC connected to GNSS and with any NIC on the same host.
// By default, we use the first port.
base_port: "",
conn_to_ctrl1_nic2: "",
},
// Optional Spirent config.
conn_to_spirent: "",
spirent_port: "",
}
},
controller_1 : {
nic1 : {
gnss_switch_port: "",
// Parameters can be : sma1 input/output, sma2 input/output, u.fl1 output, u.fl2 input, synce_rclka enabled, synce_rclkb enabled
sma1_to_nic2: "output", // CLOCK OUTPUT INTERFACE Controller-1 NIC1
sma2_to_nic2: "",
// This port for TS2PHC and PHC2SYS.
// Can be assigned to any interface of the NIC connected to GNSS and with any NIC on the same host.
// By default, we use the first port.
base_port: "",
conn_to_ctrl0_nic1: "",
// Optional Spirent config.
conn_to_spirent: "",
spirent_port: "",
},
nic2 : {
gnss_switch_port: "",
// Parameters can be : sma1 input/output, sma2 input/output, u.fl1 output, u.fl2 input, synce_rclka enabled, synce_rclkb enabled
sma1_to_nic1: "input", // CLOCK OUTPUT INTERFACE Controller-1 NIC2
sma2_to_nic1: "",
// This port for TS2PHC and PHC2SYS.
// Can be assigned to any interface of the NIC connected to GNSS and with any NIC on the same host.
// By default, we use the first port.
base_port: "enp82s0f0",
conn_to_ctrl0_nic2: "",
// Optional Spirent config.
conn_to_spirent: "",
spirent_port: "",
nics: {
nic1: {
base_port: "",
gpio_switch_port: "",
conn_to_spirent: "",
spirent_port: "",
},
nic2: {
base_port: "enp82s0f0",
gpio_switch_port: "",
conn_to_spirent: "",
spirent_port: "",
},
},
},
},
}

33
config/ptp/objects/ptp_config.py
View File

@ -19,10 +19,35 @@ class PTPConfig:
raise
ptp_dict = json5.load(json_data)
self.ptp_hosts = []
for ptp_host in ptp_dict:
host = PTPHost(ptp_host, ptp_dict[ptp_host])
self.ptp_hosts.append(host)
# GNSS server information
self.gnss_server_host = ptp_dict["gnss_server_host"]
self.gnss_server_username = ptp_dict["gnss_server_username"]
self.gnss_server_password = ptp_dict["gnss_server_password"]
# Extract the NIC Connections and Host information from the dictionary
self.ptp_hosts = self._extract_ptp_hosts(ptp_dict)
def _extract_ptp_hosts(self, ptp_dict: Dict[str, str]) -> List[PTPHost]:
"""
Build the PTPHost objects from the dictionary
Args:
ptp_dict (Dict[str, str]): JSON representation of the PTP config
Returns:
List[PTPHost]: List of PTPHost
"""
ptp_hosts = []
ptp_hosts_dict = {}
if "hosts" in ptp_dict:
ptp_hosts_dict = ptp_dict["hosts"]
for ptp_host_name in ptp_hosts_dict.keys():
host = PTPHost(ptp_host_name, ptp_hosts_dict[ptp_host_name])
ptp_hosts.append(host)
return ptp_hosts
def __str__(self):
"""

132
config/ptp/objects/ptp_host.py
View File

@ -1,6 +1,8 @@
from typing import Dict, List
from config.ptp.objects.ptp_nic import PTPNic
from config.ptp.objects.sma_connector import SMAConnector
from config.ptp.objects.ufl_connector import UFLConnector
class PTPHost:
@ -18,10 +20,128 @@ class PTPHost:
"""
self.name = host_name
self.nics = []
for nic_name in host_dict:
nic = PTPNic(nic_name, host_dict[nic_name])
self.nics.append(nic)
self.sma_connectors = self._extract_sma_connectors(host_dict)
self.ufl_connectors = self._extract_ufl_connectors(host_dict)
self.nics = self._extract_nics(host_dict)
# Set the SMA and UFL values for every NIC.
self._add_sma_connectors_to_nics()
self._add_ufl_connectors_to_nics()
def _extract_sma_connectors(self, host_dict: Dict[str, Dict]) -> List[SMAConnector]:
"""
Build the SMAConnector objects from the dictionary
Args:
host_dict (Dict[str, Dict]): JSON representation of the Host config
Returns:
List[SMAConnector]: List of SMAConnector
"""
sma_connectors = []
if "sma_connectors" in host_dict:
sma_connectors_dict = host_dict["sma_connectors"]
for sma_connector_key in sma_connectors_dict.keys():
sma_connector = SMAConnector(sma_connector_key, sma_connectors_dict[sma_connector_key])
sma_connectors.append(sma_connector)
return sma_connectors
def _extract_ufl_connectors(self, host_dict: Dict[str, Dict]) -> List[UFLConnector]:
"""
Build the UFLConnector objects from the dictionary
Args:
host_dict (Dict[str, Dict]): JSON representation of the Host config
Returns:
List[UFLConnector]: List of UFLConnector
"""
ufl_connectors = []
if "ufl_connectors" in host_dict:
ufl_connectors_dict = host_dict["ufl_connectors"]
for ufl_connector_key in ufl_connectors_dict.keys():
ufl_connector = UFLConnector(ufl_connector_key, ufl_connectors_dict[ufl_connector_key])
ufl_connectors.append(ufl_connector)
return ufl_connectors
def _extract_nics(self, host_dict: Dict[str, Dict]) -> List[PTPNic]:
"""
Build the PTPNic objects from the dictionary
Args:
host_dict (Dict[str, Dict]): JSON representation of the Host config
Returns:
List[PTPNic]: List of PTPNic
"""
nics = []
if "nics" not in host_dict:
raise Exception(f"You must provide 'nics' information for host {self.name}")
nics_dict = host_dict["nics"]
for nic_name in nics_dict.keys():
nic = PTPNic(nic_name, nics_dict[nic_name])
nics.append(nic)
return nics
def _add_sma_connectors_to_nics(self) -> None:
"""
This function will adjust the PTPNic objects to be linked to the sma_connectors
"""
for sma_connector in self.sma_connectors:
input_nic_name = sma_connector.get_input_nic()
input_nic = self.get_nic(input_nic_name)
if sma_connector.get_input_sma() == "sma1":
input_nic.set_sma1(sma_connector)
elif sma_connector.get_input_sma() == "sma2":
input_nic.set_sma2(sma_connector)
else:
raise Exception(f"sma_connectors can only be associated with sma1 or sma2. Invalid: {sma_connector.get_input_sma()}")
output_nic_name = sma_connector.get_output_nic()
output_nic = self.get_nic(output_nic_name)
if sma_connector.get_output_sma() == "sma1":
output_nic.set_sma1(sma_connector)
elif sma_connector.get_output_sma() == "sma2":
output_nic.set_sma2(sma_connector)
else:
raise Exception(f"sma_connectors can only be associated with sma1 or sma2. Invalid: {sma_connector.get_output_sma()}")
def _add_ufl_connectors_to_nics(self) -> None:
"""
This function will adjust the PTPNic objects to be linked to the ufl_connectors
"""
for ufl_connector in self.ufl_connectors:
input_nic_name = ufl_connector.get_input_nic()
input_nic = self.get_nic(input_nic_name)
if ufl_connector.get_input_ufl() == "ufl1":
input_nic.set_ufl1(ufl_connector)
elif ufl_connector.get_input_ufl() == "ufl2":
input_nic.set_ufl2(ufl_connector)
else:
raise Exception(f"ufl_connectors can only be associated with ufl1 or ufl2. Invalid: {ufl_connector.get_input_ufl()}")
output_nic_name = ufl_connector.get_output_nic()
output_nic = self.get_nic(output_nic_name)
if ufl_connector.get_output_ufl() == "ufl1":
output_nic.set_ufl1(ufl_connector)
elif ufl_connector.get_output_ufl() == "ufl2":
output_nic.set_ufl2(ufl_connector)
else:
raise Exception(f"ufl_connectors can only be associated with ufl1 or ufl2. Invalid: {ufl_connector.get_output_ufl()}")
def __str__(self):
"""
@ -30,7 +150,7 @@ class PTPHost:
Returns (str): String representation of this object.
"""
return f"PTPHost - {self.name}"
return self.name
def get_all_nics_dictionary(self) -> Dict[str, Dict]:
"""
@ -62,7 +182,7 @@ class PTPHost:
Getter for the NIC information associated with this host.
Returns
List[PTPNic]: All the nics
(List[PTPNic]): All the nics
"""
return self.nics

187
config/ptp/objects/ptp_nic.py
View File

@ -1,5 +1,9 @@
from typing import Dict
from config.ptp.objects.ptp_nic_connection import PTPNicConnection
from config.ptp.objects.sma_connector import SMAConnector
from config.ptp.objects.ufl_connector import UFLConnector
class PTPNic:
"""
@ -16,50 +20,31 @@ class PTPNic:
"""
self.name = nic_name
self.gnss_switch_port = None
self.sma1_to_nic1 = None
self.sma2_to_nic1 = None
self.sma1_to_nic2 = None
self.sma2_to_nic2 = None
self.gpio_switch_port = None
self.pci_slot = None
self.base_port = None
self.conn_to_ctrl0_nic1 = None
self.conn_to_ctrl0_nic2 = None
self.conn_to_ctrl1_nic1 = None
self.conn_to_ctrl1_nic2 = None
self.sma1 = None
self.sma2 = None
self.ufl1 = None
self.ufl2 = None
self.nic_connection = None
self.conn_to_spirent = None
self.spirent_port = None
# Store the raw dictionary for JINJA templating.
self.nic_dictionary = nic_dict
if "gnss_switch_port" in nic_dict and nic_dict["gnss_switch_port"]:
self.gnss_switch_port = nic_dict["gnss_switch_port"]
if "gpio_switch_port" in nic_dict and nic_dict["gpio_switch_port"]:
self.gpio_switch_port = nic_dict["gpio_switch_port"]
if "sma1_to_nic1" in nic_dict and nic_dict["sma1_to_nic1"]:
self.sma1_to_nic1 = nic_dict["sma1_to_nic1"]
if "sma2_to_nic1" in nic_dict and nic_dict["sma2_to_nic1"]:
self.sma2_to_nic1 = nic_dict["sma2_to_nic1"]
if "sma1_to_nic2" in nic_dict and nic_dict["sma1_to_nic2"]:
self.sma1_to_nic2 = nic_dict["sma1_to_nic2"]
if "sma2_to_nic2" in nic_dict and nic_dict["sma2_to_nic2"]:
self.sma2_to_nic2 = nic_dict["sma2_to_nic2"]
if "pci_slot" in nic_dict and nic_dict["pci_slot"]:
self.pci_slot = nic_dict["pci_slot"]
if "base_port" in nic_dict and nic_dict["base_port"]:
self.base_port = nic_dict["base_port"]
if "conn_to_ctrl0_nic1" in nic_dict and nic_dict["conn_to_ctrl0_nic1"]:
self.conn_to_ctrl0_nic1 = nic_dict["conn_to_ctrl0_nic1"]
if "conn_to_ctrl0_nic2" in nic_dict and nic_dict["conn_to_ctrl0_nic2"]:
self.conn_to_ctrl0_nic2 = nic_dict["conn_to_ctrl0_nic2"]
if "conn_to_ctrl1_nic1" in nic_dict and nic_dict["conn_to_ctrl1_nic1"]:
self.conn_to_ctrl1_nic1 = nic_dict["conn_to_ctrl1_nic1"]
if "conn_to_ctrl1_nic2" in nic_dict and nic_dict["conn_to_ctrl1_nic2"]:
self.conn_to_ctrl1_nic2 = nic_dict["conn_to_ctrl1_nic2"]
if "nic_connection" in nic_dict and nic_dict["nic_connection"]:
self.nic_connection = PTPNicConnection(self.name, nic_dict["nic_connection"])
if "conn_to_spirent" in nic_dict and nic_dict["conn_to_spirent"]:
self.conn_to_spirent = nic_dict["conn_to_spirent"]
@ -74,7 +59,7 @@ class PTPNic:
Returns (str): String representation of this object.
"""
return f"PTPNic - {self.name}"
return self.name
def to_dictionary(self) -> Dict[str, str]:
"""
@ -98,51 +83,25 @@ class PTPNic:
"""
return self.name
def get_gnss_switch_port(self) -> str:
def get_gpio_switch_port(self) -> str:
"""
Gets the GNSS switch port.
Gets the GPIO switch port.
Returns (str):
The GNSS switch port.
The GPIO switch port.
"""
return self.gnss_switch_port
return self.gpio_switch_port
def get_sma1_to_nic1(self) -> str:
def get_pci_slot(self) -> str:
"""
Gets the SMA1 to NIC1 connection.
Gets the pci slot
Returns (str):
The SMA1 to NIC1 connection.
"""
return self.sma1_to_nic1
The pci slot
def get_sma2_to_nic1(self) -> str:
"""
Gets the SMA2 to NIC1 connection.
Returns (str):
The SMA2 to NIC1 connection.
"""
return self.sma2_to_nic1
def get_sma1_to_nic2(self) -> str:
"""
Gets the SMA1 to NIC2 connection.
Returns (str):
The SMA1 to NIC2 connection.
"""
return self.sma1_to_nic2
def get_sma2_to_nic2(self) -> str:
"""
Gets the SMA2 to NIC2 connection.
Returns (str):
The SMA2 to NIC2 connection.
"""
return self.sma2_to_nic2
return self.pci_slot
def get_base_port(self) -> str:
"""
@ -153,41 +112,89 @@ class PTPNic:
"""
return self.base_port
def get_conn_to_ctrl0_nic1(self) -> str:
def get_sma1(self) -> SMAConnector:
"""
Gets the connection to controller 0 NIC1.
Gets the SMAConnector associated with sma1
Returns (str):
The connection to controller 0 NIC1.
Returns (SMAConnector):
The SMAConnector
"""
return self.conn_to_ctrl0_nic1
return self.sma1
def get_conn_to_ctrl0_nic2(self) -> str:
def set_sma1(self, sma_connector: SMAConnector) -> None:
"""
Gets the connection to controller 0 NIC2.
Sets the SMAConnector associated with sma1
Returns (str):
The connection to controller 0 NIC2.
"""
return self.conn_to_ctrl0_nic2
Args:
sma_connector (SMAConnector): the SMAConnector
def get_conn_to_ctrl1_nic1(self) -> str:
"""
Gets the connection to controller 1 NIC1.
Returns: None
Returns (str):
The connection to controller 1 NIC1.
"""
return self.conn_to_ctrl1_nic1
self.sma1 = sma_connector
def get_conn_to_ctrl1_nic2(self) -> str:
def get_sma2(self) -> SMAConnector:
"""
Gets the connection to controller 1 NIC2.
Gets the SMAConnector associated with sma2
Returns (str):
The connection to controller 1 NIC2.
Returns (SMAConnector):
The SMAConnector
"""
return self.conn_to_ctrl1_nic2
return self.sma2
def set_sma2(self, sma_connector: SMAConnector) -> None:
"""
Sets the SMAConnector associated with sma2
Args:
sma_connector (SMAConnector): the SMAConnector
Returns: None
"""
self.sma2 = sma_connector
def get_ufl1(self) -> UFLConnector:
"""
Gets the UFLConnector associated with ufl1
Returns (UFLConnector):
The UFLConnector
"""
return self.ufl1
def set_ufl1(self, ufl_connector: UFLConnector) -> None:
"""
Sets the UFLConnector associated with ufl1
Args:
ufl_connector (UFLConnector): the UFLConnector
Returns: None
"""
self.ufl1 = ufl_connector
def get_ufl2(self) -> UFLConnector:
"""
Gets the UFLConnector associated with ufl2
Returns (UFLConnector):
The UFLConnector
"""
return self.ufl2
def set_ufl2(self, ufl_connector: UFLConnector) -> None:
"""
Sets the UFLConnector associated with ufl2
Args:
ufl_connector (UFLConnector): the UFLConnector
Returns: None
"""
self.ufl2 = ufl_connector
def get_conn_to_spirent(self) -> str:
"""

81
config/ptp/objects/ptp_nic_connection.py Normal file
View File

@ -0,0 +1,81 @@
from typing import Dict
class PTPNicConnection:
"""
Class to handle a PTP Connection between nics on different controllers.
"""
def __init__(self, from_nic: str, nic_connection_dict: Dict[str, str]):
"""
Constructor.
Args:
from_nic (str): The nic of this nic_connection
nic_connection_dict (Dict[str, str]): The dictionary read from the JSON config file associated with this NIC Connection.
"""
self.validate_nic_connection_dict(nic_connection_dict)
self.from_nic = from_nic
self.to_host = nic_connection_dict["to_host"]
self.to_nic = nic_connection_dict["to_nic"]
self.interface = nic_connection_dict["interface"]
def __str__(self):
"""
String representation of this object.
Returns (str): String representation of this object.
"""
return f"{self.from_nic} to {self.to_host}:{self.to_nic}"
def validate_nic_connection_dict(self, nic_connection_dict: Dict[str, str]) -> None:
"""
Checks if the nic_connection_dict contains all the necessary fields.
Args:
nic_connection_dict (Dict[str, str]): Dictionary from the config.
"""
required_fields = ["to_host", "to_nic", "interface"]
for field in required_fields:
if field not in nic_connection_dict:
raise Exception(f"Invalid PTP config. NIC connection is missing required field: {field}")
def get_from_nic(self) -> str:
"""
Retrieves the source NIC of the NIC connection.
Returns (str):
The source NIC.
"""
return self.from_nic
def get_to_host(self) -> str:
"""
Retrieves the destination host of the NIC connection.
Returns (str):
The destination host.
"""
return self.to_host
def get_to_nic(self) -> str:
"""
Retrieves the destination NIC of the NIC connection.
Returns (str):
The destination NIC.
"""
return self.to_nic
def get_interface(self) -> str:
"""
Retrieves the interface from the NIC connection.
Returns (str):
The interface.
"""
return self.interface

104
config/ptp/objects/sma_connector.py Normal file
View File

@ -0,0 +1,104 @@
from typing import Dict
class SMAConnector:
"""
Class to handle an SMA connector between two NICs.
"""
def __init__(self, sma_connector_name: str, sma_connector_dict: Dict[str, str]):
"""
Constructor.
Args:
sma_connector_name (str): The name of this sma_connector
sma_connector_dict (Dict[str, str]): The dictionary read from the JSON config file associated with this SMA Connector.
"""
self.name = sma_connector_name
self.validate_sma_connector_dict(sma_connector_dict)
self.input_nic = sma_connector_dict["input"]["nic"]
self.input_sma = sma_connector_dict["input"]["sma"]
self.output_nic = sma_connector_dict["output"]["nic"]
self.output_sma = sma_connector_dict["output"]["sma"]
def validate_sma_connector_dict(self, sma_connector_dict: Dict[str, str]) -> None:
"""
This function validates that sma_connector_dict is formatted appropriately.
Args:
sma_connector_dict (Dict[str, str]): The dictionary read from the JSON config file associated with this SMA Connector.
"""
if "input" not in sma_connector_dict:
raise Exception(f"The sma_connector {self.name} must have an input")
if "nic" not in sma_connector_dict["input"]:
raise Exception(f"The sma_connector {self.name}'s input must have a nic entry.")
if "sma" not in sma_connector_dict["input"]:
raise Exception(f"The sma_connector {self.name}'s input must have an sma entry.")
if "output" not in sma_connector_dict:
raise Exception(f"The sma_connector {self.name} must have an input")
if "nic" not in sma_connector_dict["output"]:
raise Exception(f"The sma_connector {self.name}'s output must have a nic entry.")
if "sma" not in sma_connector_dict["output"]:
raise Exception(f"The sma_connector {self.name}'s output must have an sma entry.")
def __str__(self):
"""
String representation of this object.
Returns (str): String representation of this object.
"""
return self.name
def get_name(self) -> str:
"""
Retrieves the name from the SMA connector.
Returns (str):
The name of the SMA connector.
"""
return self.name
def get_input_nic(self) -> str:
"""
Retrieves the name of the input nic
Returns (str):
The name of the input nic
"""
return self.input_nic
def get_input_sma(self) -> str:
"""
Retrieves the name of the input sma
Returns (str):
The name of the input sma
"""
return self.input_sma
def get_output_nic(self) -> str:
"""
Retrieves the name of the output nic
Returns (str):
The name of the output nic
"""
return self.output_nic
def get_output_sma(self) -> str:
"""
Retrieves the name of the output sma
Returns (str):
The name of the output sma
"""
return self.output_sma

104
config/ptp/objects/ufl_connector.py Normal file
View File

@ -0,0 +1,104 @@
from typing import Dict
class UFLConnector:
"""
Class to handle a UFL connector between two NICs.
"""
def __init__(self, ufl_connector_name: str, ufl_connector_dict: Dict[str, str]):
"""
Constructor.
Args:
ufl_connector_name (str): The name of this ufl_connector
ufl_connector_dict (Dict[str, str]): The dictionary read from the JSON config file associated with this UFL Connector.
"""
self.name = ufl_connector_name
self.validate_ufl_connector_dict(ufl_connector_dict)
self.input_nic = ufl_connector_dict["input"]["nic"]
self.input_ufl = ufl_connector_dict["input"]["ufl"]
self.output_nic = ufl_connector_dict["output"]["nic"]
self.output_ufl = ufl_connector_dict["output"]["ufl"]
def validate_ufl_connector_dict(self, ufl_connector_dict: Dict[str, str]) -> None:
"""
This function validates that ufl_connector_dict is formatted appropriately.
Args:
ufl_connector_dict (Dict[str, str]): The dictionary read from the JSON config file associated with this UFL Connector.
"""
if "input" not in ufl_connector_dict:
raise Exception(f"The ufl_connector {self.name} must have an input")
if "nic" not in ufl_connector_dict["input"]:
raise Exception(f"The ufl_connector {self.name}'s input must have a nic entry.")
if "ufl" not in ufl_connector_dict["input"]:
raise Exception(f"The ufl_connector {self.name}'s input must have a ufl entry.")
if "output" not in ufl_connector_dict:
raise Exception(f"The ufl_connector {self.name} must have an input")
if "nic" not in ufl_connector_dict["output"]:
raise Exception(f"The ufl_connector {self.name}'s output must have a nic entry.")
if "ufl" not in ufl_connector_dict["output"]:
raise Exception(f"The ufl_connector {self.name}'s output must have a ufl entry.")
def __str__(self):
"""
String representation of this object.
Returns (str): String representation of this object.
"""
return self.name
def get_name(self) -> str:
"""
Retrieves the name from the SMA connector.
Returns (str):
The name of the SMA connector.
"""
return self.name
def get_input_nic(self) -> str:
"""
Retrieves the name of the input nic
Returns (str):
The name of the input nic
"""
return self.input_nic
def get_input_ufl(self) -> str:
"""
Retrieves the name of the input ufl
Returns (str):
The name of the input ufl
"""
return self.input_ufl
def get_output_nic(self) -> str:
"""
Retrieves the name of the output nic
Returns (str):
The name of the output nic
"""
return self.output_nic
def get_output_ufl(self) -> str:
"""
Retrieves the name of the output ufl
Returns (str):
The name of the output ufl
"""
return self.output_ufl

2
pyproject.toml
View File

@ -13,4 +13,4 @@ line-length = 9999 # Prevents Black from arbitrarily wrapping lines
style = "google"
check-return-types = true
allow-init-docstring = true
ignore = ["DOC501"] # Ignore Raises section requirement
skip-checking-raises = true # Ignore Raises section requirement

2
resources/ptp/setup/ptp_setup_template.json5
View File

@ -50,7 +50,7 @@
{
name: "phc1",
instance_hostnames : ["controller-0", "controller-1"],
instance_parameters: "'cmdline_opts=-s {{controller_0.nic1.conn_to_sprint}} -O -37 -m'",
instance_parameters: "'cmdline_opts=-s {{controller_0.nic1.conn_to_spirent}} -O -37 -m'",
ptp_interface_names: [
"phc1if1",
],

10
unit_tests/config/ptp/ptp_config_test.py
View File

@ -1,6 +1,4 @@
from config.configuration_file_locations_manager import (
ConfigurationFileLocationsManager,
)
from config.configuration_file_locations_manager import ConfigurationFileLocationsManager
from config.configuration_manager import ConfigurationManagerClass
@ -18,8 +16,8 @@ def test_default_ptp_config():
assert len(default_config.get_all_hosts()) == 2, "There are two hosts in the PTP config"
config_for_controller_0 = default_config.get_host("controller_0")
assert len(config_for_controller_0.get_all_nics()) == 2, "There are two NIC assigned to controller-0 in the PTP config"
assert len(config_for_controller_0.get_all_nics()) == 3, "There are three NICs assigned to controller-0 in the PTP config"
first_nic = config_for_controller_0.get_nic("nic1")
assert first_nic.get_sma1_to_nic2() == "output"
assert first_nic.get_conn_to_ctrl1_nic1() == "enp81s0f1"
assert first_nic.get_sma1().get_name() == "nic_1_2"
assert not first_nic.get_ufl2()