trying to script a direct connection to POE connected camera

Jan 1, 2024
337
166
USA
Hello all, So I am getting lazy at my old age and trying to script a direct connection to a POE connected camera on one of my Amcrest POE Nvrs.. As we all know that use POE NVRs you have to connect to your NVR and then go to main menu, camera and click a connected IP camera you want to virtual host to that cameras WebUI.. So I thought I would try and script something that would log in get me a connection for a POE camera so it would just bring up the IP cameras WebUI without having to take all these steps. Now in some NVRs the connection is same idea but they are static in nature so able to just pick NVR info and what channel one would want to open because the port info for each port is always the same. Where with Amcrest and or Dahua first one loads in 10080 and second one you open in same session is 10081 and so. Where the next session if I click say channel 5 first time and then second session of NVR login I went to channel 12 first it would load that using the 10080.. So I am trying to get my script to load one of the channels again doing so without having to log into the NVR first.. I think the issue is the Port Mapping? Not even sure it can be done but would even be willing to add in the direct port info if I had to, so it would work?. IN the test NVR it is a 16ch 4k NVR with 8 POE ports and the one I want to load is port3 aka channel 3

right now this is only thing that is being returned.
An error occurred: ('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))


Here is the code so far. Before I added in some debug I did get some return from the NVR so I know it is talking with it I just don't know what is missing. I do think it is the Port part of my script?
Code:
import requests
import webbrowser

# Replace with your NVR details
NVR_IP = "10.0.0.225"
NVR_USERNAME = "admin"
NVR_PASSWORD = "password"

def login_to_nvr():
    """Login to the NVR and retrieve session ID."""
    url = f"http://{NVR_IP}/RPC2"
    payload = {
        "method": "global.login",
        "params": {
            "userName": NVR_USERNAME,
            "password": NVR_PASSWORD,
            "clientType": "Web3.0"
        },
        "id": 1,
        "session": 0  # Session ID is typically 0 for the first login attempt
    }

    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
        "Content-Type": "application/json"
    }

    try:
        # Attempt login with a timeout of 60 seconds and no SSL verification
        response = requests.post(url, json=payload, headers=headers, timeout=60, verify=False)
        response.raise_for_status()  # Raise an error for HTTP codes 4xx/5xx

        # Debugging: print status and response text
        print(f"Response status code: {response.status_code}")
        print(f"Response content: {response.text}")  # Print response content for debugging

        # Parse the response
        response_data = response.json()
        if "result" in response_data and response_data["result"]:
            session_id = response_data["session"]
            print("Login successful, Session ID:", session_id)
            return session_id
        else:
            print("Login failed:", response_data)
            return None
    except requests.exceptions.RequestException as e:
        # Catch and print any request exceptions (including connection issues)
        print(f"An error occurred: {e}")
        return None

def get_camera_ports(session_id):
    """Retrieve camera port mappings."""
    url = f"http://{NVR_IP}/RPC2"
    payload = {
        "method": "camera.list",
        "params": {},
        "id": 2,
        "session": session_id
    }

    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(url, json=payload, headers=headers, timeout=60, verify=False)  # Timeout increased
        response.raise_for_status()  # Raise an error for HTTP codes 4xx/5xx

        # Debugging: print status and response text
        print(f"Response status code: {response.status_code}")
        print(f"Response content: {response.text}")  # Print response content for debugging

        response_data = response.json()
        if "result" in response_data:
            return response_data["result"]["cameras"]
        else:
            print("Failed to get cameras:", response_data)
            return []
    except requests.exceptions.RequestException as e:
        # Catch and print any request exceptions (including connection issues)
        print(f"An error occurred: {e}")
        return []

def open_camera_webui(camera):
    """Open the camera's web UI in the browser."""
    camera_url = f"http://{NVR_IP}:{camera['virtualHostPort']}"
    print(f"Opening {camera['name']} at {camera_url}")
    webbrowser.open(camera_url)

# Main workflow
session_id = login_to_nvr()
if session_id:
    cameras = get_camera_ports(session_id)
    for camera in cameras:
        open_camera_webui(camera)
 
Try to reuse some of the logic from here:

BTW, there is a connection timeout of 30/60 minutes, so, it think need to login again after timeout or send keep alives messages (not sure which).