mirror of
https://github.com/KevinMidboe/mktxp-no-cli.git
synced 2025-10-29 17:50:23 +00:00
DS refactor, fixes/optimizations
This commit is contained in:
@@ -21,21 +21,13 @@ from mktxp.cli.config.config import config_handler
|
|||||||
class BaseOutputProcessor:
|
class BaseOutputProcessor:
|
||||||
OutputCapsmanEntry = namedtuple('OutputCapsmanEntry', ['dhcp_name', 'dhcp_address', 'mac_address', 'rx_signal', 'interface', 'ssid', 'tx_rate', 'rx_rate', 'uptime'])
|
OutputCapsmanEntry = namedtuple('OutputCapsmanEntry', ['dhcp_name', 'dhcp_address', 'mac_address', 'rx_signal', 'interface', 'ssid', 'tx_rate', 'rx_rate', 'uptime'])
|
||||||
OutputWiFiEntry = namedtuple('OutputWiFiEntry', ['dhcp_name', 'dhcp_address', 'mac_address', 'signal_strength', 'signal_to_noise', 'interface', 'tx_rate', 'rx_rate', 'uptime'])
|
OutputWiFiEntry = namedtuple('OutputWiFiEntry', ['dhcp_name', 'dhcp_address', 'mac_address', 'signal_strength', 'signal_to_noise', 'interface', 'tx_rate', 'rx_rate', 'uptime'])
|
||||||
OutputDHCPEntry = namedtuple('OutputDHCPEntry', ['host_name', 'comment', 'address', 'active_address', 'mac_address', 'server', 'expires_after'])
|
OutputDHCPEntry = namedtuple('OutputDHCPEntry', ['host_name', 'server', 'mac_address', 'address', 'active_address', 'expires_after'])
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def augment_record(router_entry, registration_record, dhcp_lease_records):
|
def augment_record(router_entry, registration_record, dhcp_lease_records):
|
||||||
try:
|
try:
|
||||||
dhcp_lease_record = next((dhcp_lease_record for dhcp_lease_record in dhcp_lease_records if dhcp_lease_record['mac_address']==registration_record['mac_address']))
|
dhcp_lease_record = next((dhcp_lease_record for dhcp_lease_record in dhcp_lease_records if dhcp_lease_record['mac_address']==registration_record['mac_address']))
|
||||||
dhcp_name = dhcp_lease_record.get('host_name')
|
dhcp_name = BaseOutputProcessor.dhcp_name(router_entry, dhcp_lease_record)
|
||||||
dhcp_comment = dhcp_lease_record.get('comment')
|
|
||||||
|
|
||||||
if dhcp_name and dhcp_comment:
|
|
||||||
dhcp_name = f'{dhcp_name[0:20]} ({dhcp_comment[0:20]})' if not router_entry.config_entry.use_comments_over_names else dhcp_comment
|
|
||||||
elif dhcp_comment:
|
|
||||||
dhcp_name = dhcp_comment
|
|
||||||
else:
|
|
||||||
dhcp_name = dhcp_lease_record.get('mac_address') if not dhcp_name else dhcp_name
|
|
||||||
dhcp_address = dhcp_lease_record.get('address', '')
|
dhcp_address = dhcp_lease_record.get('address', '')
|
||||||
except StopIteration:
|
except StopIteration:
|
||||||
dhcp_name = registration_record['mac_address']
|
dhcp_name = registration_record['mac_address']
|
||||||
@@ -62,6 +54,23 @@ class BaseOutputProcessor:
|
|||||||
if registration_record.get('rx_signal'):
|
if registration_record.get('rx_signal'):
|
||||||
registration_record['rx_signal'] = BaseOutputProcessor.parse_signal_strength(registration_record['rx_signal'])
|
registration_record['rx_signal'] = BaseOutputProcessor.parse_signal_strength(registration_record['rx_signal'])
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def dhcp_name(router_entry, dhcp_lease_record, drop_comment = False):
|
||||||
|
dhcp_name = dhcp_lease_record.get('host_name')
|
||||||
|
dhcp_comment = dhcp_lease_record.get('comment')
|
||||||
|
|
||||||
|
if dhcp_name and dhcp_comment:
|
||||||
|
dhcp_name = f'{dhcp_name[0:20]} ({dhcp_comment[0:20]})' if not router_entry.config_entry.use_comments_over_names else dhcp_comment
|
||||||
|
elif dhcp_comment:
|
||||||
|
dhcp_name = dhcp_comment
|
||||||
|
else:
|
||||||
|
dhcp_name = dhcp_lease_record.get('mac_address') if not dhcp_name else dhcp_name
|
||||||
|
|
||||||
|
if drop_comment:
|
||||||
|
del dhcp_lease_record['comment']
|
||||||
|
|
||||||
|
return dhcp_name
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def parse_rates(rate):
|
def parse_rates(rate):
|
||||||
wifi_rates_rgx = config_handler.re_compiled.get('wifi_rates_rgx')
|
wifi_rates_rgx = config_handler.re_compiled.get('wifi_rates_rgx')
|
||||||
|
|||||||
@@ -40,9 +40,12 @@ class DHCPOutput:
|
|||||||
output_table = []
|
output_table = []
|
||||||
for key in dhcp_by_server.keys():
|
for key in dhcp_by_server.keys():
|
||||||
for record in dhcp_by_server[key]:
|
for record in dhcp_by_server[key]:
|
||||||
|
record['host_name'] = BaseOutputProcessor.dhcp_name(router_entry, record, drop_comment = True)
|
||||||
output_table.append(BaseOutputProcessor.OutputDHCPEntry(**record))
|
output_table.append(BaseOutputProcessor.OutputDHCPEntry(**record))
|
||||||
num_records += 1
|
num_records += 1
|
||||||
output_table.append({})
|
output_table.append({})
|
||||||
print()
|
print()
|
||||||
print(tabulate(output_table, headers = "keys", tablefmt="github"))
|
print(tabulate(output_table, headers = "keys", tablefmt="github"))
|
||||||
print(tabulate([{0:'DHCP Clients:', 1:num_records}], tablefmt="text"))
|
print(tabulate([{0:'DHCP Clients:', 1:num_records}], tablefmt="text"))
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user