Update WiFi-specific network data from scan results and provide aggregated counts
()
| 4033 | def _normalize_network_slug(identifier: Optional[str]) -> Optional[str]: |
| 4034 | """Normalize a requested network identifier to a storage slug.""" |
| 4035 | if not identifier: |
| 4036 | return None |
| 4037 | candidate = identifier.strip() |
| 4038 | if not candidate: |
| 4039 | return None |
| 4040 | |
| 4041 | manager = getattr(shared_data, 'storage_manager', None) |
| 4042 | slugify = getattr(manager, '_slugify', None) if manager else None |
| 4043 | if callable(slugify): |
| 4044 | try: |
| 4045 | return slugify(candidate) |
| 4046 | except Exception as exc: |
| 4047 | logger.debug(f"Failed to slugify network identifier '{candidate}': {exc}") |
| 4048 | |
| 4049 | simplified = re.sub(r'[^a-z0-9]+', '_', candidate.lower()).strip('_') |
| 4050 | return simplified or candidate.lower() |
| 4051 | |
| 4052 | |
| 4053 | @contextmanager |
| 4054 | def _network_context_from_request(): |
| 4055 | """Temporarily switch shared data to the network requested by the client.""" |
| 4056 | identifier = _extract_requested_network_identifier() |
| 4057 | slug = _normalize_network_slug(identifier) |
| 4058 | registry = getattr(shared_data, 'context_registry', None) |
| 4059 | |
| 4060 | if slug and registry: |
| 4061 | try: |
| 4062 | with registry.activate(slug): |
| 4063 | g.requested_network_slug = slug |
| 4064 | yield slug |
| 4065 | return |
| 4066 | except Exception as exc: |
| 4067 | logger.warning(f"Unable to activate network context '{slug}': {exc}") |
| 4068 | finally: |
| 4069 | try: |
| 4070 | g.pop('requested_network_slug', None) |
| 4071 | except Exception: |
| 4072 | pass |
| 4073 | |
| 4074 | yield None |
| 4075 | |
| 4076 | |
| 4077 | def run_targeted_arp_scan(ip, interface=DEFAULT_ARP_SCAN_INTERFACE): |
| 4078 | command = ['sudo', 'arp-scan', f'--interface={interface}', ip] |
| 4079 | logger.info(f"Running targeted arp-scan for {ip}: {' '.join(command)}") |
| 4080 | try: |
| 4081 | result = subprocess.run(command, capture_output=True, text=True, check=False, timeout=60) |
| 4082 | hosts = _parse_arp_scan_output(result.stdout) |
| 4083 | entry = hosts.get(ip) |
| 4084 | return entry.get('mac', '') if entry else '' |
| 4085 | except FileNotFoundError: |
| 4086 | logger.warning(f"arp-scan command not found when resolving MAC for {ip}") |
| 4087 | return '' |
| 4088 | except subprocess.TimeoutExpired as e: |
| 4089 | logger.warning(f"arp-scan timed out for {ip}: {e}") |
| 4090 | hosts = _parse_arp_scan_output(e.stdout or '') |
| 4091 | entry = hosts.get(ip) |
| 4092 | return entry.get('mac', '') if entry else '' |
no test coverage detected