gxfxyz/dahua_rpc.py

## dahua_rpc.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Basic Dahua RPC wrapper.

Example:
  from dahua_rpc import DahuaRpc

  dahua = DahuaRpc(host="192.168.1.10", username="admin", password="password")
  dahua.login()

  # Get the current time on the device
  print(dahua.current_time())

  # Set display to 4 grids with first view group
  dahua.set_split(mode=4, view=1)

  # Make a raw RPC request to get serial number
  print(dahua.request(method="magicBox.getSerialNo"))

Dependencies:
  pip install requests
"""

import sys
import hashlib

import requests

if (sys.version_info > (3, 0)):
    unicode = str


class DahuaRpc(object):

    def __init__(self, host, username, password):
        self.host = host
        self.username = username
        self.password = password

        self.s = requests.Session()
        self.session_id = None
        self.id = 0

    def request(self, method, params=None, object_id=None, extra=None, url=None):
        """Make a RPC request."""
        self.id += 1
        data = {'method': method, 'id': self.id}
        if params is not None:
            data['params'] = params
        if object_id:
            data['object'] = object_id
        if extra is not None:
            data.update(extra)
        if self.session_id:
            data['session'] = self.session_id
        if not url:
            url = "http://{}/RPC2".format(self.host)
        r = self.s.post(url, json=data)
        return r.json()

    def login(self):
        """Dahua RPC login.

        Reversed from rpcCore.js (login, getAuth & getAuthByType functions).
        Also referenced:
        https://gist.github.com/avelardi/1338d9d7be0344ab7f4280618930cd0d
        """

        # login1: get session, realm & random for real login
        url = 'http://{}/RPC2_Login'.format(self.host)
        method = "global.login"
        params = {'userName': self.username,
                  'password': "",
                  'clientType': "Dahua3.0-Web3.0"}
        r = self.request(method=method, params=params, url=url)

        self.session_id = r['session']
        realm = r['params']['realm']
        random = r['params']['random']

        # Password encryption algorithm
        # Reversed from rpcCore.getAuthByType
        pwd_phrase = self.username + ":" + realm + ":" + self.password
        if isinstance(pwd_phrase, unicode):
            pwd_phrase = pwd_phrase.encode('utf-8')
        pwd_hash = hashlib.md5(pwd_phrase).hexdigest().upper()
        pass_phrase = self.username + ':' + random + ':' + pwd_hash
        if isinstance(pass_phrase, unicode):
            pass_phrase = pass_phrase.encode('utf-8')
        pass_hash = hashlib.md5(pass_phrase).hexdigest().upper()

        # login2: the real login
        params = {'userName': self.username,
                  'password': pass_hash,
                  'clientType': "Dahua3.0-Web3.0",
                  'authorityType': "Default",
                  'passwordType': "Default"}
        r = self.request(method=method, params=params, url=url)

        if r['result'] is False:
            raise LoginError(str(r))

    def set_config(self, params):
        """Set configurations."""

        method = "configManager.setConfig"
        r = self.request(method=method, params=params)

        if r['result'] is False:
            raise RequestError(str(r))

    def reboot(self):
        """Reboot the device."""

        # Get object id
        method = "magicBox.factory.instance"
        params = ""
        r = self.request(method=method, params=params)
        object_id = r['result']

        # Reboot
        method = "magicBox.reboot"
        r = self.request(method=method, params=params, object_id=object_id)

        if r['result'] is False:
            raise RequestError(str(r))

    def current_time(self):
        """Get the current time on the device."""

        method = "global.getCurrentTime"
        r = self.request(method=method)

        if r['result'] is False:
            raise RequestError(str(r))

        return r['params']['time']

    def ntp_sync(self, address, port, time_zone):
        """Synchronize time with NTP."""

        # Get object id
        method = "netApp.factory.instance"
        params = ""
        r = self.request(method=method, params=params)
        object_id = r['result']

        # NTP sync
        method = "netApp.adjustTimeWithNTP"
        params = {'Address': address, 'Port': port, 'TimeZone': time_zone}
        r = self.request(method=method, params=params, object_id=object_id)

        if r['result'] is False:
            raise RequestError(str(r))

    def get_split(self):
        """Get display split mode."""

        # Get object id
        method = "split.factory.instance"
        params = {'channel': 0}
        r = self.request(method=method, params=params)
        object_id = r['result']

        # Get split mode
        method = "split.getMode"
        params = ""
        r = self.request(method=method, params=params, object_id=object_id)

        if r['result'] is False:
            raise RequestError(str(r))

        mode = int(r['params']['mode'][5:])
        view = int(r['params']['group']) + 1

        return mode, view

    def set_split(self, mode, view):
        """Set display split mode."""

        if isinstance(mode, int):
            mode = "Split{}".format(mode)
        group = view - 1

        # Get object id
        method = "split.factory.instance"
        params = {'channel': 0}
        r = self.request(method=method, params=params)
        object_id = r['result']

        # Set split mode
        method = "split.setMode"
        params = {'displayType': "General",
                  'workMode': "Local",
                  'mode': mode,
                  'group': group}
        r = self.request(method=method, params=params, object_id=object_id)

        if r['result'] is False:
            raise RequestError(str(r))


class LoginError(Exception):
    pass


class RequestError(Exception):
    pass
	#!/usr/bin/env python
	# -- coding: utf-8 --

	"""
	Basic Dahua RPC wrapper.

	Example:
	from dahua_rpc import DahuaRpc

	dahua = DahuaRpc(host="192.168.1.10", username="admin", password="password")
	dahua.login()

	# Get the current time on the device
	print(dahua.current_time())

	# Set display to 4 grids with first view group
	dahua.set_split(mode=4, view=1)

	# Make a raw RPC request to get serial number
	print(dahua.request(method="magicBox.getSerialNo"))

	Dependencies:
	pip install requests
	"""

	import sys
	import hashlib

	import requests

	if (sys.version_info > (3, 0)):
	unicode = str


	class DahuaRpc(object):

	def __init__(self, host, username, password):
	self.host = host
	self.username = username
	self.password = password

	self.s = requests.Session()
	self.session_id = None
	self.id = 0

	def request(self, method, params=None, object_id=None, extra=None, url=None):
	"""Make a RPC request."""
	self.id += 1
	data = {'method': method, 'id': self.id}
	if params is not None:
	data['params'] = params
	if object_id:
	data['object'] = object_id
	if extra is not None:
	data.update(extra)
	if self.session_id:
	data['session'] = self.session_id
	if not url:
	url = "http://{}/RPC2".format(self.host)
	r = self.s.post(url, json=data)
	return r.json()

	def login(self):
	"""Dahua RPC login.

	Reversed from rpcCore.js (login, getAuth & getAuthByType functions).
	Also referenced:
	https://gist.github.com/avelardi/1338d9d7be0344ab7f4280618930cd0d
	"""

	# login1: get session, realm & random for real login
	url = 'http://{}/RPC2_Login'.format(self.host)
	method = "global.login"
	params = {'userName': self.username,
	'password': "",
	'clientType': "Dahua3.0-Web3.0"}
	r = self.request(method=method, params=params, url=url)

	self.session_id = r['session']
	realm = r['params']['realm']
	random = r['params']['random']

	# Password encryption algorithm
	# Reversed from rpcCore.getAuthByType
	pwd_phrase = self.username + ":" + realm + ":" + self.password
	if isinstance(pwd_phrase, unicode):
	pwd_phrase = pwd_phrase.encode('utf-8')
	pwd_hash = hashlib.md5(pwd_phrase).hexdigest().upper()
	pass_phrase = self.username + ':' + random + ':' + pwd_hash
	if isinstance(pass_phrase, unicode):
	pass_phrase = pass_phrase.encode('utf-8')
	pass_hash = hashlib.md5(pass_phrase).hexdigest().upper()

	# login2: the real login
	params = {'userName': self.username,
	'password': pass_hash,
	'clientType': "Dahua3.0-Web3.0",
	'authorityType': "Default",
	'passwordType': "Default"}
	r = self.request(method=method, params=params, url=url)

	if r['result'] is False:
	raise LoginError(str(r))

	def set_config(self, params):
	"""Set configurations."""

	method = "configManager.setConfig"
	r = self.request(method=method, params=params)

	if r['result'] is False:
	raise RequestError(str(r))

	def reboot(self):
	"""Reboot the device."""

	# Get object id
	method = "magicBox.factory.instance"
	params = ""
	r = self.request(method=method, params=params)
	object_id = r['result']

	# Reboot
	method = "magicBox.reboot"
	r = self.request(method=method, params=params, object_id=object_id)

	if r['result'] is False:
	raise RequestError(str(r))

	def current_time(self):
	"""Get the current time on the device."""

	method = "global.getCurrentTime"
	r = self.request(method=method)

	if r['result'] is False:
	raise RequestError(str(r))

	return r['params']['time']

	def ntp_sync(self, address, port, time_zone):
	"""Synchronize time with NTP."""

	# Get object id
	method = "netApp.factory.instance"
	params = ""
	r = self.request(method=method, params=params)
	object_id = r['result']

	# NTP sync
	method = "netApp.adjustTimeWithNTP"
	params = {'Address': address, 'Port': port, 'TimeZone': time_zone}
	r = self.request(method=method, params=params, object_id=object_id)

	if r['result'] is False:
	raise RequestError(str(r))

	def get_split(self):
	"""Get display split mode."""

	# Get object id
	method = "split.factory.instance"
	params = {'channel': 0}
	r = self.request(method=method, params=params)
	object_id = r['result']

	# Get split mode
	method = "split.getMode"
	params = ""
	r = self.request(method=method, params=params, object_id=object_id)

	if r['result'] is False:
	raise RequestError(str(r))

	mode = int(r['params']['mode'][5:])
	view = int(r['params']['group']) + 1

	return mode, view

	def set_split(self, mode, view):
	"""Set display split mode."""

	if isinstance(mode, int):
	mode = "Split{}".format(mode)
	group = view - 1

	# Get object id
	method = "split.factory.instance"
	params = {'channel': 0}
	r = self.request(method=method, params=params)
	object_id = r['result']

	# Set split mode
	method = "split.setMode"
	params = {'displayType': "General",
	'workMode': "Local",
	'mode': mode,
	'group': group}
	r = self.request(method=method, params=params, object_id=object_id)

	if r['result'] is False:
	raise RequestError(str(r))


	class LoginError(Exception):
	pass


	class RequestError(Exception):
	pass