#!/usr/bin/env python3 import requests import json import re # --- Configuration --- DEFAULT_DEVICE_ADDRESS = "192.168.222.1" DEVICE_PORT = 80 REQUEST_TIMEOUT = 10 # seconds class EthLinxControl: """ API client for the EthLinx series of devices. Handles fetching device information, getting/setting Serial parameters, and controlling the device via its HTTP server. """ def __init__(self, device_ip=DEFAULT_DEVICE_ADDRESS, port=DEVICE_PORT, timeout=REQUEST_TIMEOUT, verbose=False): """ Initializes the EthLinxControl client. """ self.base_url = f"http://{device_ip}:{port}" self.settings_url = f"{self.base_url}/settings_data.js" self.device_info_url = f"{self.base_url}/device_info_data.js" self.status_data_url = f"{self.base_url}/status_data.js" self.save_url = f"{self.base_url}/save_serial_settings" self.restart_url = f"{self.base_url}/restart_device" self.clear_stats_url = f"{self.base_url}/clear_serial_stats" # Corrected endpoint self.timeout = timeout self.verbose = verbose self.last_settings = None self._log(f"EthLinxControl initialized for URL: {self.base_url}", level='info') def _log(self, message, level="info"): """ Internal logging helper based on verbosity. """ if level in ["error", "summary"]: print(message) elif self.verbose and level in ["info", "debug"]: print(f"[API-DEBUG] {message}") def _fetch_and_parse_js_object(self, url, variable_name): """ Generic helper to fetch a .js file and parse a JSON object from it. """ self._log(f"Attempting to GET and parse from: {url}", level='info') try: response = requests.get(url, timeout=self.timeout) response.raise_for_status() content = response.text match = re.search(fr'{variable_name}\s*=\s*({{.*}});?', content, re.DOTALL) if not match: self._log(f"Error: Could not find variable '{variable_name}' with JSON object in response.", level='error') return None json_string = match.group(1) json_string = re.sub(r'//.*?\n', '\n', json_string) json_string = re.sub(r'/\*.*?\*/', '', json_string, flags=re.DOTALL) json_string = re.sub(r',\s*([\}\]])', r'\1', json_string) return json.loads(json_string) except requests.exceptions.RequestException as e: self._log(f"Error fetching from {url}: {e}", level='error') return None except json.JSONDecodeError as e: self._log(f"Error decoding JSON from response: {e}", level='error') return None except Exception as e: self._log(f"An unexpected error occurred during fetch from {url}: {e}", level='error') return None def _post_command(self, url, action_name): """Generic helper for sending simple POST commands.""" self._log(f"Sending command to {url}...", level='info') try: response = requests.post(url, timeout=self.timeout) if response.ok: self._log(f"{action_name} command successful.", level='summary') return True else: self._log(f"Failed to {action_name}. Status code: {response.status_code}", level='error') self._log(f"Error Response: {response.text}", level='error') return False except requests.exceptions.RequestException as e: self._log(f"Error during POST request for {action_name}: {e}", level='error') return False def get_device_info(self): """Fetches the main device information from device_info_data.js.""" return self._fetch_and_parse_js_object(self.device_info_url, "deviceInfoValues") def get_settings(self): """Fetches the current serial settings from the device.""" settings = self._fetch_and_parse_js_object(self.settings_url, "settingsValues") if settings: self.last_settings = settings return settings def get_stats(self): """Fetches the current transfer statistics from the device.""" return self._fetch_and_parse_js_object(self.status_data_url, "statusValues") def clear_stats(self): """Sends a command to clear the transfer statistics on the device.""" return self._post_command(self.clear_stats_url, "clear stats") def restart_device(self): """ Sends a command to restart the device. This function expects the connection to drop, so errors are treated as success. """ self._log(f"Sending restart command to {self.restart_url}...", level='info') try: requests.post(self.restart_url, timeout=1) except (requests.exceptions.ConnectionError, requests.exceptions.Timeout): self._log("Restart command sent successfully (connection dropped as expected).", level='summary') return True except requests.exceptions.RequestException as e: self._log(f"An unexpected error occurred during restart command: {e}", level='error') return False self._log("Restart command sent successfully (device responded before reboot).", level='summary') return True def save_settings(self, bus_id: str, label: str, baud_rate: int, terminator: bool, bias: bool, data_bits: int, parity: int, stop_bits: int): """Saves new settings for a specific Serial bus via POST request.""" self._log("Fetching current settings before saving...", level='info') current_settings = self.get_settings() if current_settings is None: self._log("Save aborted: Failed to retrieve current settings.", level='error') return False bus_key = f"serial-{bus_id}" if bus_key not in current_settings: valid_buses = list(current_settings.keys()) self._log(f"Error: Invalid bus_id '{bus_id}'. Key '{bus_key}' not found.", level='error') self._log(f"Available buses reported by device: {valid_buses}", level='error') return False payload = { "busId": str(bus_id), "label": label, "baudRate": str(baud_rate), "terminator": terminator, "bias": bias, "dataBits": str(data_bits), "parity": str(parity), "stopBits": str(stop_bits) } self._log(f"Attempting to save settings for bus {bus_id}: " f"Label='{label}', Baud={baud_rate}, Term={terminator}, Bias={bias}, " f"DataBits={data_bits}, Parity={parity}, Stop={stop_bits}", level='summary') try: response = requests.post(self.save_url, json=payload, timeout=self.timeout) if response.ok: self._log("Settings saved successfully.", level='summary') return True else: self._log(f"Failed to save settings. Status code: {response.status_code}", level='error') self._log(f"Error Response: {response.text}", level='error') return False except requests.exceptions.RequestException as e: self._log(f"Error during POST request: {e}", level='error') return False