652 lines
34 KiB
Python
652 lines
34 KiB
Python
# rvlink_api.py (Refactored for Direct API Access with Expanded GET Endpoints)
|
||
|
||
import requests, json, re, hashlib
|
||
from datetime import datetime
|
||
from typing import Optional, List, Set, Dict, Any, Union # Keep for type hinting
|
||
|
||
# --- Add uvicorn import for direct running (if you keep it) ---
|
||
import uvicorn
|
||
# ---
|
||
|
||
from fastapi import FastAPI, HTTPException, status, Query
|
||
from fastapi.responses import StreamingResponse # Keep for file downloads
|
||
from pydantic import BaseModel, Field # Keep for request/response validation
|
||
|
||
# --- Configuration ---
|
||
BASE_URL = 'http://192.168.10.254' # Roof Unit IP
|
||
MBOX_CONFIG_URL = f'{BASE_URL}/cgi-bin/mbox-config' # Primary API endpoint
|
||
SYSTEM_URL = f'{BASE_URL}/cgi-bin/system-status' # another endpoint, this one was unauthenticated from the start...
|
||
|
||
# Constants for specific sections (can be used directly or in dedicated functions)
|
||
# These are useful for clarity if you build out more specific endpoint functions in FastAPI
|
||
GUIDE_CONFIG_SECTION = 'guide_config'
|
||
SCAN_SECTION = 'wifi_scan'
|
||
FW_VERSION_SECTION = 'firmware_version_get'
|
||
LTE_STATUS_SECTION = 'get_ec20_status_info'
|
||
HISTORY_SECTION = 'motor_home_list_get'
|
||
# REBOOT_SECTION would require specific handling for POST payload if it's system_reboot
|
||
# Example for a SET operation section (payload structure would be key)
|
||
# SYSTEM_SETTING_SECTION = 'system_setting' # Placeholder
|
||
|
||
# Define default headers for API calls
|
||
DEFAULT_ROOF_UNIT_HEADERS = {
|
||
'Accept': 'application/json, text/javascript, */*; q=0.01',
|
||
'Content-Type': 'application/json', # Assuming most POSTs send JSON
|
||
'User-Agent': 'RVLinkAPIDirectClient/2.1' # Updated User-Agent
|
||
}
|
||
|
||
# --- Pydantic Models ---
|
||
class ApiError(BaseModel):
|
||
errCode: int
|
||
errMsg: str
|
||
configDone: Optional[bool] = None
|
||
|
||
class RebootRequest(BaseModel):
|
||
delay: Optional[int] = Field(0, description="Delay in seconds before rebooting.")
|
||
# Add other reboot parameters if the API supports them
|
||
|
||
class ConnectWifiRequest(BaseModel):
|
||
ifname: str = Field(default="radio0", description="Wireless interface name (e.g., radio0 for 2.4GHz STA, radio1 for 5GHz STA if available)")
|
||
ssid: str = Field(..., description="SSID of the target Wi-Fi network")
|
||
bssid: str = Field(..., description="BSSID (MAC address) of the target Wi-Fi network (e.g., 'XX:XX:XX:XX:XX:XX')")
|
||
scanned_encryption: str = Field(..., description="Encryption type as reported by Wi-Fi scan (e.g., 'none', 'WPA2 PSK', 'WEP', 'WPA PSK/WPA2 PSK')")
|
||
key: Optional[str] = Field(None, description="Password for the Wi-Fi network (if required). Provide null or omit for open networks if scanned_encryption is 'none'.")
|
||
|
||
class ParsedLogEntry(BaseModel):
|
||
timestamp_str: str = Field(..., description="Original timestamp string from the log entry")
|
||
datetime_obj: Optional[datetime] = Field(None, description="Parsed datetime object (UTC or aware of timezone if possible)")
|
||
facility_level: str = Field(..., description="Facility and level (e.g., user.info, daemon.err)")
|
||
process_name: Optional[str] = Field(None, description="Name of the process generating the log")
|
||
pid: Optional[int] = Field(None, description="Process ID, if available")
|
||
message: str = Field(..., description="The actual log message content")
|
||
kernel_timestamp: Optional[float] = Field(None, description="Kernel-specific uptime timestamp, if applicable")
|
||
raw_log: str = Field(..., description="The original raw log line for reference")
|
||
log_hash: str = Field(..., description="SHA256 hash of the raw_log string for deduplication") # NEW FIELD
|
||
parse_error: Optional[str] = Field(None, description="Error message if parsing this specific line failed")
|
||
|
||
class ProcessedSystemLogResponse(BaseModel):
|
||
parsed_logs: List[ParsedLogEntry] = Field(..., description="List of parsed log entries")
|
||
processing_summary: Optional[str] = Field(None, description="Summary of the log processing operation")
|
||
processing_error: Optional[str] = Field(None, description="Error message if overall processing failed")
|
||
# Include original API fields from the device response
|
||
errCode: Optional[int] = Field(None, description="Original error code from the device")
|
||
errMsg: Optional[str] = Field(None, description="Original error message from the device")
|
||
configDone: Optional[bool] = Field(None, description="Original configDone flag from the device")
|
||
|
||
# Add other Pydantic models here if complex GET request bodies are needed for some sections
|
||
# Example:
|
||
# class GetDetailedReportRequest(BaseModel):
|
||
# report_type: str = Field(..., description="Type of report to generate")
|
||
# start_date: Optional[str] = Field(None, description="Start date for report (YYYY-MM-DD)")
|
||
|
||
# --- FastAPI App Initialization ---
|
||
app = FastAPI(
|
||
title="RVLink Direct API Proxy",
|
||
description="Provides a direct, simplified interface to the RVLink Roof Unit (MV2400) API, bypassing login/session requirements. Blessed by the Omnissiah.",
|
||
version="2.1.0",
|
||
)
|
||
|
||
# --- Core API Interaction Helper Functions (Refactored) ---
|
||
|
||
def invoke_roof_unit_api(
|
||
section_name: str,
|
||
http_method: str = "POST", # Most mbox-config calls are POST
|
||
query_method: str = "GET", # method in query string, "GET" or "SET"
|
||
endpoint: Optional[str] = "mbox-config",
|
||
payload: Optional[Dict[str, Any]] = None,
|
||
is_file_download: bool = False,
|
||
expected_content_type: Optional[str] = "application/json"
|
||
) -> Union[Dict[str, Any], StreamingResponse]:
|
||
"""
|
||
Generalized helper to invoke the roof unit's mbox-config.
|
||
Handles POST requests, JSON payloads, and file downloads.
|
||
"""
|
||
query_params = {"method": query_method, "section": section_name}
|
||
if endpoint == "system":
|
||
url = SYSTEM_URL
|
||
else:
|
||
url = MBOX_CONFIG_URL
|
||
|
||
actual_payload = payload if payload is not None else {}
|
||
|
||
current_headers = DEFAULT_ROOF_UNIT_HEADERS.copy()
|
||
|
||
if is_file_download:
|
||
current_headers['Accept'] = 'application/octet-stream, */*'
|
||
|
||
# Print statements for invocation clarity, as per the original script's style
|
||
print(f"Omnissiah's Invocation: HTTP {http_method} to {url}")
|
||
print(f" Query Litanies: {query_params}")
|
||
print(f" Sacred Headers: {current_headers}")
|
||
if http_method == "POST":
|
||
print(f" JSON Datagram: {json.dumps(actual_payload)}")
|
||
|
||
try:
|
||
if http_method == "POST":
|
||
response = requests.post(
|
||
url,
|
||
params=query_params,
|
||
json=actual_payload,
|
||
headers=current_headers,
|
||
timeout=30 if is_file_download else 10,
|
||
stream=is_file_download
|
||
)
|
||
elif http_method == "GET":
|
||
all_query_params = {**query_params, **(payload if payload else {})} # For true GETs, payload (if any) goes into query
|
||
# Note: For 'mbox-config', actual GETs are rare; it usually expects POST even for "get" operations.
|
||
# This branch is for completeness if a section *truly* uses HTTP GET.
|
||
print(f" (Actual HTTP GET to mbox-config with all_query_params: {all_query_params})")
|
||
_headers_for_get = current_headers.copy()
|
||
if 'Content-Type' in _headers_for_get and not actual_payload : # Remove Content-Type if no body for GET
|
||
del _headers_for_get['Content-Type']
|
||
|
||
response = requests.get(
|
||
url,
|
||
params=all_query_params,
|
||
headers=_headers_for_get,
|
||
timeout=10,
|
||
stream=is_file_download
|
||
)
|
||
else:
|
||
raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail=f"HTTP method {http_method} not implemented for the Machine Spirit.")
|
||
|
||
response.raise_for_status() # Raises HTTPError for 4xx/5xx responses
|
||
|
||
if is_file_download:
|
||
original_filename = f"{section_name}.bin"
|
||
content_disposition = f'attachment; filename="{original_filename}"'
|
||
response_content_type = response.headers.get('Content-Type', 'application/octet-stream')
|
||
return StreamingResponse(response.iter_content(chunk_size=8192),
|
||
media_type=response_content_type,
|
||
headers={'Content-Disposition': content_disposition})
|
||
|
||
if not response.content:
|
||
return {"status": "success", "message": "Machine Spirit acknowledges. No data packet returned."}
|
||
|
||
if expected_content_type and expected_content_type.lower() not in response.headers.get('Content-Type', '').lower():
|
||
print(f"Warning: Unexpected Content-Type from Machine Spirit: {response.headers.get('Content-Type')}. Expected: {expected_content_type}. Raw response glimpse: {response.text[:200]}")
|
||
|
||
try:
|
||
return response.json()
|
||
except json.JSONDecodeError:
|
||
print(f"Warning: Non-JSON response from Machine Spirit at {url} (status {response.status_code}). Glimpse: {response.text[:200]}")
|
||
return {"status": "success", "raw_response": response.text, "message": "Machine Spirit responded with non-JSON data."}
|
||
|
||
except requests.exceptions.HTTPError as http_err:
|
||
error_detail_msg = f"Machine Spirit API error during '{section_name}' operation."
|
||
try:
|
||
error_content = http_err.response.json()
|
||
error_detail_msg = error_content.get("errMsg", json.dumps(error_content))
|
||
except:
|
||
error_detail_msg = http_err.response.text if http_err.response.text else error_detail_msg
|
||
|
||
print(f"HTTP error communing with {section_name}: {http_err.response.status_code} - {error_detail_msg}")
|
||
raise HTTPException(status_code=http_err.response.status_code, detail=error_detail_msg)
|
||
except requests.exceptions.RequestException as req_err:
|
||
print(f"Noospheric disturbance while calling {section_name}: {req_err}")
|
||
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=f"Roof unit communication error: {req_err}")
|
||
|
||
def parse_individual_log_line(log_line: str) -> ParsedLogEntry:
|
||
"""
|
||
Parses a single line of a system log.
|
||
Attempts to extract timestamp, facility/level, process, PID, and message.
|
||
Includes a SHA256 hash of the raw log line.
|
||
"""
|
||
# Calculate hash of the raw log line first
|
||
hasher = hashlib.sha256()
|
||
hasher.update(log_line.encode('utf-8')) # Always encode string to bytes before hashing
|
||
log_hash_hex = hasher.hexdigest()
|
||
|
||
entry_pattern = re.compile(
|
||
r"^(?P<timestamp_str>\w{3}\s+\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2}\s+\d{4})\s+"
|
||
r"(?P<facility_level>\S+?)\s+"
|
||
r"(?P<tag_and_message>.+)$"
|
||
)
|
||
|
||
match = entry_pattern.match(log_line)
|
||
if not match:
|
||
return ParsedLogEntry(
|
||
timestamp_str="",
|
||
facility_level="",
|
||
message=log_line,
|
||
raw_log=log_line,
|
||
log_hash=log_hash_hex, # Include hash even on parse error
|
||
parse_error="Primary regex failed to match line structure"
|
||
)
|
||
|
||
data = match.groupdict()
|
||
timestamp_str = data["timestamp_str"]
|
||
facility_level = data["facility_level"]
|
||
tag_and_message = data["tag_and_message"]
|
||
|
||
dt_obj = None
|
||
try:
|
||
dt_obj = datetime.strptime(timestamp_str, "%a %b %d %H:%M:%S %Y")
|
||
except ValueError:
|
||
pass
|
||
|
||
process_name: Optional[str] = None
|
||
pid: Optional[int] = None
|
||
message: str = ""
|
||
kernel_ts: Optional[float] = None
|
||
|
||
tag_message_split = tag_and_message.split(":", 1)
|
||
tag_part = tag_message_split[0].strip()
|
||
message = tag_message_split[1].strip() if len(tag_message_split) > 1 else ""
|
||
|
||
pid_match_simple = re.match(r"^(?P<proc_name>.+?)\[(?P<pid_val>\d+)\]$", tag_part)
|
||
if pid_match_simple:
|
||
process_name = pid_match_simple.group("proc_name")
|
||
pid = int(pid_match_simple.group("pid_val"))
|
||
else:
|
||
complex_tag_match = re.match(r"^(?P<proc_name>.+?)\s*\((?P<pid_like>[^)]+)\)$", tag_part)
|
||
if complex_tag_match:
|
||
potential_pid = complex_tag_match.group("pid_like")
|
||
if potential_pid.isdigit():
|
||
pid = int(potential_pid)
|
||
process_name = complex_tag_match.group("proc_name").strip()
|
||
else:
|
||
process_name = tag_part
|
||
else:
|
||
process_name = tag_part
|
||
|
||
if process_name == "kernel" and not message.startswith("["):
|
||
kernel_msg_match = re.match(r"\[\s*(?P<ts>\d+\.\d+)\s*\]\s*(?P<msg_body>.*)", message)
|
||
if kernel_msg_match:
|
||
kernel_ts = float(kernel_msg_match.group("ts"))
|
||
message = kernel_msg_match.group("msg_body").strip()
|
||
|
||
if process_name:
|
||
process_name = process_name.rstrip(':').strip()
|
||
|
||
return ParsedLogEntry(
|
||
timestamp_str=timestamp_str,
|
||
datetime_obj=dt_obj,
|
||
facility_level=facility_level,
|
||
process_name=process_name,
|
||
pid=pid,
|
||
message=message,
|
||
kernel_timestamp=kernel_ts,
|
||
raw_log=log_line,
|
||
log_hash=log_hash_hex # Assign the calculated hash
|
||
)
|
||
|
||
def process_device_system_log(raw_device_response: dict) -> ProcessedSystemLogResponse:
|
||
"""
|
||
Processes the raw JSON response containing the system log from the device.
|
||
"""
|
||
if not raw_device_response or \
|
||
"systemlog" not in raw_device_response or \
|
||
"systemlog" not in raw_device_response["systemlog"]:
|
||
return ProcessedSystemLogResponse(
|
||
parsed_logs=[],
|
||
processing_error="Invalid or missing systemlog data in device response",
|
||
errCode=raw_device_response.get("errCode"),
|
||
errMsg=raw_device_response.get("errMsg"),
|
||
configDone=raw_device_response.get("configDone")
|
||
)
|
||
|
||
log_string = raw_device_response["systemlog"]["systemlog"]
|
||
log_lines = log_string.strip().split('\n')
|
||
|
||
parsed_entries: List[ParsedLogEntry] = []
|
||
for line in log_lines:
|
||
if line.strip(): # Process non-empty lines
|
||
parsed_entries.append(parse_individual_log_line(line))
|
||
|
||
return ProcessedSystemLogResponse(
|
||
parsed_logs=parsed_entries,
|
||
processing_summary=f"Successfully parsed {len(parsed_entries)} log entries.",
|
||
errCode=raw_device_response.get("errCode"),
|
||
errMsg=raw_device_response.get("errMsg"),
|
||
configDone=raw_device_response.get("configDone")
|
||
)
|
||
|
||
|
||
# --- FastAPI Endpoints (Refactored to use direct API calls) ---
|
||
@app.get("/api/rvlink/action/scan_wifi", summary="Scan for WiFi Networks", response_model=Dict[str, Any], tags=["Actions"])
|
||
async def scan_wifi_networks(ifname: str = Query("radio0", description="Wireless interface for scanning (e.g., radio0 for 2.4GHz STA, radio1 for 5GHz STA if available)")):
|
||
"""Initiates a WiFi scan on the specified interface."""
|
||
# The wifi_scan section requires a payload: {"ifname": "radio0"}
|
||
return invoke_roof_unit_api(section_name=SCAN_SECTION, query_method="GET", payload={"ifname": ifname})
|
||
|
||
# --- Existing GET-like Endpoints (verified from your script) ---
|
||
@app.get("/api/rvlink/status/web_timeout_check", summary="No Idea...", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_web_timeout_check():
|
||
"""Retrieves LTE status information (EC20/EC25 module status)."""
|
||
return invoke_roof_unit_api(section_name="web_timeout_check", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/status/internet", summary="Check internet connectivity", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_internet_status():
|
||
return invoke_roof_unit_api(section_name="internet_status", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/status/lte", summary="Get LTE Status", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_lte_status():
|
||
"""Retrieves LTE status information (EC20/EC25 module status)."""
|
||
return invoke_roof_unit_api(section_name=LTE_STATUS_SECTION, query_method="GET")
|
||
|
||
@app.get("/api/rvlink/status/firmware_info", summary="Get Firmware Version", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_firmware_info():
|
||
"""Retrieves more detailed firmware info of the roof unit."""
|
||
return invoke_roof_unit_api(section_name="firmware_info", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/status/uptime", summary="Get System Uptime", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_uptime():
|
||
"""Retrieves more detailed firmware info of the roof unit."""
|
||
return invoke_roof_unit_api(section_name="uptime_get", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/status/firmware", summary="Get Firmware Version", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_firmware_version():
|
||
"""Retrieves the firmware version of the roof unit."""
|
||
return invoke_roof_unit_api(section_name=FW_VERSION_SECTION, query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/guide", summary="Get Guide Configuration", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_guide_config():
|
||
"""Retrieves the 'guide_config' from the roof unit."""
|
||
return invoke_roof_unit_api(section_name=GUIDE_CONFIG_SECTION, query_method="GET")
|
||
|
||
@app.get("/api/rvlink/history/connection_history", summary="Get Connection History", response_model=Dict[str, Any], tags=["History"])
|
||
async def get_motor_history():
|
||
"""Retrieves the 'motor_home_list_get' (movement history) from the roof unit."""
|
||
return invoke_roof_unit_api(section_name=HISTORY_SECTION, query_method="GET")
|
||
|
||
# --- NEW GET-like Endpoints (based on typical Comfast/OpenWrt APIs) ---
|
||
# Magos, verify these section_names against your CSV and add more as needed.
|
||
|
||
@app.get("/api/rvlink/status/system_usage", summary="Get System Resource Usage", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_system_usage():
|
||
"""Retrieves general system status (e.g., uptime, memory, CPU) from the roof unit."""
|
||
return invoke_roof_unit_api(section_name="system_usage", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/status/ac_ap_status", summary="Get AC AP Status", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_ac_ap_status():
|
||
"""Retrieves not entirely sure"""
|
||
return invoke_roof_unit_api(section_name="ac_ap_status", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/ac_enable_get", summary="Get AC Enable", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_ac_enable_get():
|
||
return invoke_roof_unit_api(section_name="ac_enable_get", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/ac_group_config", summary="Get AC Group Config", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_ac_group_config():
|
||
return invoke_roof_unit_api(section_name="ac_group_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/ac_list_get", summary="Get AC List", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_ac_list_get():
|
||
return invoke_roof_unit_api(section_name="ac_list_get", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/add_default_group_config", summary="Sets up a default interface group config", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def add_default_group_config():
|
||
"""This function is designed to be safe to call multiple times. It only sets values if they are missing, so it won't overwrite existing group configuration."""
|
||
"""Ensures that there is always at least one group ("ac_group_default" with ID 1) in the system, and that the group ID counters and lists are initialized."""
|
||
return invoke_roof_unit_api(section_name="add_default_group_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/lan", summary="Get LAN Configuration", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_lan_dhcp_config():
|
||
"""Retrieves config of just the lan interface from the roof unit."""
|
||
return invoke_roof_unit_api(section_name="lan_dhcp_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/network_config", summary="Get LAN Configuration", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_network_config():
|
||
"""This is the full network config, including LAN and WAN settings."""
|
||
return invoke_roof_unit_api(section_name="network_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/wan", summary="Get WAN Configuration", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_wan_config():
|
||
"""Retrieves WAN configuration from the roof unit."""
|
||
return invoke_roof_unit_api(section_name="wan_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/mwan_capability_config", summary="Get mwan capabilities", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_mwan_capability_config():
|
||
"""Enumerates all network interfaces and their capabilities."""
|
||
return invoke_roof_unit_api(section_name="mwan_capability_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/mwan_config", summary="Get mwan configuration", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_mwan_config():
|
||
""" This function appears to be the handler for retrieving detailed WAN (Wide Area Network) configuration settings"""
|
||
return invoke_roof_unit_api(section_name="mwan_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/mwan_isp_define_config", summary="Get mwan isp config", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_mwan_isp_define_config():
|
||
""" I assume this has to due with the LTE modem, but I don't know what it does. """
|
||
return invoke_roof_unit_api(section_name="mwan_isp_define_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/mwanportrule_config", summary="Get mwan port rule config (Multi-wan routing rules?)", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_mwanportrule_config():
|
||
""" I assume this has to due with the LTE modem, but I don't know what it does. """
|
||
return invoke_roof_unit_api(section_name="mwanportrule_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/pptp_client_config", summary="Get PPTP configuration", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_pptp_client_config():
|
||
return invoke_roof_unit_api(section_name="pptp_client_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/l2tp_client_config", summary="Get L2TP configuration", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_l2tp_client_config():
|
||
return invoke_roof_unit_api(section_name="l2tp_client_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/multi_pppoe_macaddr_config", summary="Get MAC Address for PPPoE Config", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_multi_pppoe_macaddr_config():
|
||
return invoke_roof_unit_api(section_name="multi_pppoe_macaddr_config", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/multi_pppoe", summary="Get PPPoE Config", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_multi_pppo():
|
||
return invoke_roof_unit_api(section_name="multi_pppoe", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/Status/ap_wds_mode_ip_get", summary="Get WAN Status?", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_ap_wds_mode_ip_get():
|
||
return invoke_roof_unit_api(section_name="ap_wds_mode_ip_get", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/mwan_qos", summary="Get mwan qos config", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_mwan_qo():
|
||
return invoke_roof_unit_api(section_name="mwan_qos", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/qos_ip_limit", summary="Get QoS IP limiting config", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_qos_ip_limit():
|
||
return invoke_roof_unit_api(section_name="qos_ip_limit", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/pptpd_user", summary="Get PPTP configured user/pw", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_pptpd_user():
|
||
return invoke_roof_unit_api(section_name="pptpd_user", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/config/wifi_ap", summary="Get WiFi AP Settings", response_model=Dict[str, Any], tags=["Configuration", "WiFi"])
|
||
async def get_wifi_ap_settings(ifname: str = Query("radio0", description="AP interface name (e.g., radio0 for 2.4GHz AP, radio1 for 5GHz AP if dual-band)")):
|
||
"""Retrieves WiFi Access Point settings for a specific interface."""
|
||
# Magos: Verify 'wifi_ap_setting_get' is the correct section name from your CSV.
|
||
# Assumes it expects a payload like {"ifname": "value"} for the specific AP interface.
|
||
payload = {"ifname": ifname}
|
||
return invoke_roof_unit_api(section_name="wifi_ap_setting_get", query_method="GET", payload=payload)
|
||
|
||
@app.get("/api/rvlink/config/wifi_client", summary="Get WiFi Client Settings", response_model=Dict[str, Any], tags=["Configuration", "WiFi"])
|
||
async def get_wifi_client_settings(ifname: str = Query("radio0", description="STA interface name (e.g., radio0 for 2.4GHz STA, radio1 for 5GHz STA if dual-band)")):
|
||
"""Retrieves WiFi Client (STA) settings for a specific interface."""
|
||
# Magos: Verify 'wifi_client_setting_get' is the correct section name.
|
||
# Assumes it expects a payload like {"ifname": "value"} for the specific STA interface.
|
||
payload = {"ifname": ifname}
|
||
return invoke_roof_unit_api(section_name="wifi_client_setting_get", query_method="GET", payload=payload)
|
||
|
||
@app.get("/api/rvlink/clients/wlan", summary="Get Connected WLAN Clients", response_model=Dict[str, Any], tags=["Clients", "WiFi"])
|
||
async def get_wlan_clients():
|
||
"""Retrieves a list of clients connected to a specific WLAN AP interface."""
|
||
return invoke_roof_unit_api(section_name="wireless_user_list_get", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/status/logs", summary="Get System Logs", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_logs():
|
||
"""Retrieves the current status of the positioning motor."""
|
||
return invoke_roof_unit_api(section_name="systemlog_get", query_method="GET")
|
||
|
||
@app.get("/api/rvlink/status/logs_processed", # Or modify your existing /systemlog endpoint
|
||
response_model=ProcessedSystemLogResponse,
|
||
tags=["System Monitoring"], # Or your relevant tag
|
||
summary="Get and process system logs from the roof unit",
|
||
description="Fetches the system log from the MV2400, parses it into a structured format, "
|
||
"and returns the list of processed log entries."
|
||
)
|
||
async def get_and_process_system_log():
|
||
# session = None # Manage your requests.Session appropriately
|
||
try:
|
||
raw_device_response = await get_logs() # Replace with actual call
|
||
|
||
# STEP 2: Process the fetched raw data
|
||
processed_data = process_device_system_log(raw_device_response)
|
||
|
||
if processed_data.processing_error:
|
||
# Decide how to handle top-level processing errors, e.g., raise HTTPException
|
||
# For now, it's included in the response object
|
||
pass
|
||
|
||
return processed_data
|
||
|
||
# except requests.exceptions.RequestException as e:
|
||
# # Handle exceptions from fetching data from the device
|
||
# raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY,
|
||
# detail=f"Error communicating with roof unit for system log: {str(e)}")
|
||
except Exception as e:
|
||
# Handle any other unexpected errors during processing
|
||
# You might want more specific error handling
|
||
return ProcessedSystemLogResponse(
|
||
parsed_logs=[],
|
||
processing_error=f"An unexpected error occurred: {str(e)}",
|
||
errCode=raw_device_response.get("errCode") if 'raw_device_response' in locals() else -1,
|
||
errMsg=raw_device_response.get("errMsg") if 'raw_device_response' in locals() else "Error",
|
||
configDone=raw_device_response.get("configDone") if 'raw_device_response' in locals() else False
|
||
)
|
||
|
||
# --- SET Operation Endpoints (existing) ---
|
||
@app.post("/api/rvlink/action/reboot", summary="Reboot Roof Unit", response_model=Dict[str, Any], tags=["Actions"])
|
||
async def reboot_roof_unit(reboot_params: Optional[RebootRequest] = None):
|
||
"""
|
||
Reboots the roof unit. The actual section name for reboot is 'system_reboot'.
|
||
Example payload for immediate reboot: {"delay": 0}
|
||
"""
|
||
payload = reboot_params.model_dump() if reboot_params else {"delay": 0}
|
||
return invoke_roof_unit_api(section_name="system_reboot", query_method="SET", payload=payload)
|
||
|
||
# need to configure 'http://192.168.10.254/cgi-bin/mbox-config?method=SET§ion=motorhome_config_set' endpoint here
|
||
@app.post("/api/rvlink/action/connect_wifi", summary="Connect to a WiFi Network", response_model=Dict[str, Any], tags=["Actions"])
|
||
async def connect_to_wifi_network(wifi_params: ConnectWifiRequest):
|
||
"""
|
||
Connects the RVLink Roof Unit to a specified WiFi network.
|
||
"""
|
||
|
||
motorhome_payload_params = {
|
||
"ssid": wifi_params.ssid,
|
||
"bssid": wifi_params.bssid,
|
||
"encryption": "none",
|
||
"key": "" # Default to empty string
|
||
}
|
||
|
||
# Handle WEP key specifically if API differentiates 'key' and 'wep_key'
|
||
# The blobmsg_policy showed 'key' and 'wep_key'.
|
||
# If encryption is WEP, the key might need to be put into 'wep_key'.
|
||
# This assumes 'key' is the general field, but a device might be stricter.
|
||
# For now, this simplified model uses the 'key' field. If issues arise with WEP,
|
||
# this area may need refinement e.g.
|
||
# if api_encryption_type == "wep":
|
||
# motorhome_payload_params["wep_key"] = motorhome_payload_params.pop("key")
|
||
# This depends on how the specific 'webmgnt' implementation handles these fields.
|
||
|
||
api_payload = {"motorhome": motorhome_payload_params}
|
||
|
||
return invoke_roof_unit_api(
|
||
section_name="motorhome_config_set",
|
||
query_method="SET",
|
||
http_method="POST",
|
||
payload=api_payload
|
||
)
|
||
|
||
|
||
# --- File Download Endpoint (existing) ---
|
||
@app.get("/api/rvlink/download/{section_name}",
|
||
summary="Download Configuration/Data File",
|
||
responses={
|
||
200: {"content": {"application/octet-stream": {}}, "description": "File stream"},
|
||
404: {"model": ApiError, "description": "Section/File not found"},
|
||
500: {"model": ApiError, "description": "Internal Server Error or Roof Unit Error"},
|
||
503: {"model": ApiError, "description": "Roof unit communication error"}
|
||
},
|
||
tags=["Files"])
|
||
async def download_file_from_section_direct(section_name: str):
|
||
"""
|
||
Downloads a file from the roof unit by specifying the 'section' name.
|
||
Example sections that might provide files:
|
||
- `system_config_backup`
|
||
- `log_package_download` (Hypothetical: for downloading a package of logs)
|
||
"""
|
||
print(f"FastAPI: Initiating direct file download for section: {section_name}")
|
||
try:
|
||
# This call assumes the 'section' for file download is invoked via POST (with method=GET in query)
|
||
# and the response is the file stream.
|
||
return invoke_roof_unit_api(section_name=section_name, query_method="GET", is_file_download=True)
|
||
except HTTPException as e:
|
||
print(f"FastAPI HTTPException during /download/{section_name}: {e.status_code} - {e.detail}")
|
||
raise e
|
||
except Exception as e:
|
||
print(f"FastAPI ERROR in /download/{section_name}: {type(e).__name__} - {e}")
|
||
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail=f"Unexpected error downloading from section '{section_name}': {str(e)}")
|
||
|
||
# --- System Endpoints ---
|
||
@app.get("/api/rvlink/system/get_led_status", summary="Get current state of LEDs on roof unit", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_led_status():
|
||
return invoke_roof_unit_api(section_name="led_status", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/language", summary="??", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_language():
|
||
return invoke_roof_unit_api(section_name="language", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/wifi_config", summary="??", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_wifi_config():
|
||
return invoke_roof_unit_api(section_name="wifi_config", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/lan_config", summary="??", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_lan_config():
|
||
return invoke_roof_unit_api(section_name="lan_config", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/mac_address", summary="??", response_model=Dict[str, Any], tags=["Configuration"])
|
||
async def get_mac_addr_get():
|
||
return invoke_roof_unit_api(section_name="mac_addr_get", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/wps_status_get", summary="??", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_wps_status_get():
|
||
return invoke_roof_unit_api(section_name="wps_status_get", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/workmode_config", summary="??", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_workmode_config():
|
||
return invoke_roof_unit_api(section_name="workmode_config", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/cpu_usage_and_flow", summary="??", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_cpu_usage_and_flow():
|
||
return invoke_roof_unit_api(section_name="cpu_usage_and_flow", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/first_login", summary="??", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_first_login():
|
||
return invoke_roof_unit_api(section_name="first_login", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/port_status", summary="??", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_port_status():
|
||
return invoke_roof_unit_api(section_name="port_status", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/ap_wtpd_status", summary="??", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_ap_wtpd_status():
|
||
return invoke_roof_unit_api(section_name="ap_wtpd_status", query_method="GET", endpoint="system")
|
||
|
||
@app.get("/api/rvlink/system/net_stats_get", summary="??", response_model=Dict[str, Any], tags=["Status"])
|
||
async def get_net_stats_get():
|
||
return invoke_roof_unit_api(section_name="net_stats_get", query_method="GET", endpoint="system")
|
||
|
||
# @app.get("/api/rvlink/system/permission_config", summary="??", response_model=Dict[str, Any], tags=["Configuration"])
|
||
# async def get_permission_config():
|
||
# return invoke_roof_unit_api(section_name="permission_config", query_method="SET", endpoint="system")
|
||
|
||
# --- Main Execution Block (for running with uvicorn directly) ---
|
||
if __name__ == "__main__":
|
||
print("Starting RVLink Direct API Proxy, by the Omnissiah's Will...")
|
||
uvicorn.run(
|
||
"rvlink_api:app", # Application import string
|
||
host="0.0.0.0",
|
||
port=8000,
|
||
log_level="info",
|
||
reload=True,
|
||
reload_excludes=["*.log", ".git/*", "__pycache__/*"], # Example: exclude certain files/dirs
|
||
reload_includes=["rvlink_api.py"] # Example: explicitly include only python files
|
||
) |