Preface

IOTOS Connect with Haikang access control machine , This interface supports normally open door 、 Door normally closed 、 Door state acquisition caused by door opening and door closing . Door normally open operation , The door will remain open , Will not automatically turn off , Perform door closing operation , The door will close ; Door normally closed operation , The door will remain closed , Ordinary card swiping door will not be opened , Perform door opening operation , The door will open ; Door opening operation , Execute the door opening action , Door opening time exceeded , The door will close automatically ; Door closing operation , Close the door , Will close the door immediately .
Call the interface , First, through the interface of obtaining the resource list of access control points , Get the unique number of access control point , Then query the access control point status according to the unique number of the access control point . It should be noted that the door channel must be connected to the door magnet to send the door status change notice normally , If the door magnet is not connected , The platform cannot update the door status through the door status change notification .

Support device

All can go to Haikang integrated security management platform （iSecure Center） Platform access control machine

SDK file

Haikang integrated security management platform （iSecure Center）

Purpose

Obtain the switch status of the access control machine

Scope of application

Hardware equipment of all Haikang access control machines 、IOTOS Collection procedure

Examples of use

Code example

#!coding:utf8
import json
import sys
import threading
sys.path.append("..")
from driver import *
import logging
import urllib3
import hmac  # hex-based message authentication code  Hash message authentication code 
import hashlib  #  Provides many encryption algorithms 
import base64
import certifi

class URL_list():
	def __init__(self):
		#  obtain token, Expiration time 12 Hours  √
		self.token = "/artemis/api/v1/oauth/token"
		#  Query the status of access control points 
		self.door_stat = "/artemis/api/acs/v1/door/states"
		#  Query the list of access control points （ Access control point ）
		self.door_search = "/artemis/api/resource/v2/door/search"


class info_post():
	def __init__(self, _driver_instance):
		self.driver_instance = _driver_instance
		#  Zongping address 
		self.base_url = "https://143.134.201.253:443"
		#  Comprehensive leveling key
		self.appKey = "29740612"
		#  Comprehensive review serect
		self.appSecret = "di1lnAhWkkRUuSzGQBQN"
		#  Request method 
		self.http_method = "POST"
		self.encoding = "utf-8"
		#  Initialization request address 
		self.req_uri = URL_list()
		# https initialization 
		urllib3.disable_warnings()
		self.https = urllib3.PoolManager(cert_reqs='CERT_NONE', ca_certs=certifi.where())
		self.token_header_key = "X-Subject-Token"

	def sign(self, key, value):
		temp = hmac.new(key.encode(), value.encode(), digestmod=hashlib.sha256)
		return base64.b64encode(temp.digest()).decode()

	def request(self, url, headers=None, data=None, method="POST"):
		try:
			if data:
				if isinstance(data, dict):
					data = json.dumps(data)
			response = self.https.urlopen(method, url, body=data, headers=headers)
			response_body = response.data.decode(self.encoding)
			try:
				jsonObj = json.loads(response_body)
				return response.status, jsonObj
			except json.JSONDecodeError:
				return response.status + 100, response_body
		except urllib3.exceptions.HTTPError as e:
			return 500, None
		except Exception as e:
			return 500, None

	def access_token(self):
		# sign_str  The splicing of is very important , Or you won't get the right signature 
		sign_str = "POST\n*/*\napplication/json" + "\nx-ca-key:" + self.appKey + "\n" + \
				   self.req_uri.token

		signature = self.sign(self.appSecret, sign_str)

		headers = {
    
			'Accept': '*/*',
			'Content-Type': 'application/json',
			'x-ca-key': self.appKey,  # appKey, namely  AK
			'x-ca-signature-headers': 'x-ca-key',
			'x-ca-signature': signature,  #  The signature that needs to be calculated , Here you can get... Through the background 
		}
		url = self.base_url + self.req_uri.token
		stauts,results = self.request(method="POST", url=url, headers=headers)
		if stauts == 200:
			self.acc_token = results.get(u"data", {
    }).get(u"access_token", None)

		threading.Timer(10, self.access_token).start()

	def door_state(self):
		url = self.base_url + self.req_uri.door_stat
		headers = {
    
			'access_token': self.acc_token,
			'Content-Type': 'application/json',
		}
		body = {
    
			"doorIndexCodes": []
		}
		door_sta = self.request(url, headers=headers, data=body)
		return door_sta

	def door_search(self):
		value_type = {
    0:" The initial state ",1:" Open the door ",2:" Closing status ",3:" Offline status "}
		code,door_status = self.door_state()
		url = self.base_url + self.req_uri.door_search
		headers = {
    
			'access_token': self.acc_token,
			'Content-Type': 'application/json',
		}
		body = {
    
			"pageNo": 1,
			"pageSize": 200
		}
		status,door_search_info = self.request(url, headers=headers, data=body)
		door = door_search_info.get("data", {
    }).get("list", [])
		door_statu = door_status.get("data", {
    }).get("authDoorList", [])
		if status == 200:
			for i in door:
				for j in door_statu:
					door_in = json.dumps(i)
					door = json.loads(door_in)
					status_in = json.dumps(j)
					status = json.loads(status_in)
					indexcode = door.get(u"indexCode", "")
					name = door.get(u"name", "")
					st =  status.get(u"doorState", "")
					door_index_code = status.get(u"doorIndexCode", "")
					if door_index_code == indexcode:
						self.driver_instance.setValue(name,value_type[st])

		threading.Timer(1, self.door_search).start()

class door(IOTOSDriverI):
	#1、 Communication initialization 
	def InitComm(self,attrs):
		self.online(True)
		self.collectingOneCircle = False  #  Let's collect Collecting Execute only one loop , Traverse the point table once ！
		self.pauseCollect = False
		self.apartment = info_post(self)
		threading.Timer(1, self.apartment.access_token).start()
		threading.Timer(2, self.apartment.door_search).start()

Code instructions

Initialize acquisition

Built in function class Init initialization , So that the cyclic acquisition can start .

    def InitComm(self, attrs):
        self.online(True) # Equipment online 
        self.setCollectingOneCircle(False)# One cycle acquisition false
        self.setPauseCollect(False)# Pause collection false

Circular process requests access control data points

Process request access control data point

Code ( equipment )

def door_search(self):
		value_type = {
    0:" The initial state ",1:" Open the door ",2:" Closing status ",3:" Offline status "}
		code,door_status = self.door_state()
		url = self.base_url + self.req_uri.door_search
		headers = {
    
			'access_token': self.acc_token,
			'Content-Type': 'application/json',
		}
		body = {
    
			"pageNo": 1,
			"pageSize": 200
		}
		status,door_search_info = self.request(url, headers=headers, data=body)
		door = door_search_info.get("data", {
    }).get("list", [])
		door_statu = door_status.get("data", {
    }).get("authDoorList", [])
		if status == 200:
			for i in door:
				for j in door_statu:
					door_in = json.dumps(i)
					door = json.loads(door_in)
					status_in = json.dumps(j)
					status = json.loads(status_in)
					indexcode = door.get(u"indexCode", "")
					name = door.get(u"name", "")
					st =  status.get(u"doorState", "")
					door_index_code = status.get(u"doorIndexCode", "")
					if door_index_code == indexcode:
						self.driver_instance.setValue(name,value_type[st])

		threading.Timer(1, self.door_search).start()

The data points

Running results