123 lines
4.7 KiB
Python
123 lines
4.7 KiB
Python
import csv
|
|
from io import StringIO
|
|
from lxml import etree
|
|
from lxml.builder import ElementMaker
|
|
import requests
|
|
|
|
E_plain = ElementMaker(nsmap={"hid": "http://www.hidglobal.com/VertX"})
|
|
E = ElementMaker(namespace="http://www.hidglobal.com/VertX",
|
|
nsmap={"hid": "http://www.hidglobal.com/VertX"})
|
|
E_corp = ElementMaker(namespace="http://www.hidcorp.com/VertX", #stupid
|
|
nsmap={"hid": "http://www.hidcorp.com/VertX"})
|
|
ROOT = E_plain.VertXMessage
|
|
|
|
fieldnames = "CardNumber,CardFormat,PinRequired,PinCode,ExtendedAccess,ExpiryDate,Forename,Initial,Surname,Email,Phone,Custom1,Custom2,Schedule1,Schedule2,Schedule3,Schedule4,Schedule5,Schedule6,Schedule7,Schedule8".split(",")
|
|
|
|
class RemoteError(Exception):
|
|
def __init__(self, r):
|
|
super().__init__(
|
|
f"Door Updating Error: {r.status_code} {r.reason}\n{r.text}")
|
|
|
|
class DoorController():
|
|
def __init__(self, ip, username, password, name="", access=""):
|
|
self.ip = ip
|
|
self.username = username
|
|
self.password = password
|
|
self.name = name
|
|
self.access = access
|
|
|
|
def doImportRequest(self, params=None, files=None):
|
|
"""Send a request to the door control import script"""
|
|
r = requests.post(
|
|
'https://' + self.ip + '/cgi-bin/import.cgi',
|
|
params=params,
|
|
files=files,
|
|
auth=requests.auth.HTTPDigestAuth(self.username, self.password),
|
|
timeout=60,
|
|
verify=False) # ignore insecure SSL
|
|
xml = etree.XML(r.content)
|
|
if r.status_code != 200 \
|
|
or len(xml.findall("{http://www.hidglobal.com/VertX}Error")) > 0:
|
|
raise RemoteError(r)
|
|
|
|
def doCSVImport(self, csv):
|
|
"""Do the CSV import procedure on a door control"""
|
|
self.doImportRequest({"task": "importInit"})
|
|
self.doImportRequest({"task": "importCardsPeople", "name": "cardspeopleschedule.csv"},
|
|
{"importCardsPeopleButton": ("cardspeopleschedule.csv", csv, 'text/csv')})
|
|
self.doImportRequest({"task": "importDone"})
|
|
|
|
def doXMLRequest(self, xml, prefix=b'<?xml version="1.0" encoding="UTF-8"?>'):
|
|
if not isinstance(xml, bytes):
|
|
xml = etree.tostring(xml)
|
|
r = requests.get(
|
|
'https://' + self.ip + '/cgi-bin/vertx_xml.cgi',
|
|
params={'XML': prefix + xml},
|
|
auth=requests.auth.HTTPDigestAuth(self.username, self.password),
|
|
verify=False)
|
|
resp_xml = etree.XML(r.content)
|
|
# probably meed to be more sane about this
|
|
if r.status_code != 200 \
|
|
or len(resp_xml.findall("{*}Error")) > 0:
|
|
raise RemoteError(r)
|
|
return resp_xml
|
|
|
|
def sendSchedules(self, schedules):
|
|
# clear all people
|
|
outString = StringIO()
|
|
writer = csv.DictWriter(outString, fieldnames)
|
|
writer.writeheader()
|
|
writer.writerow({})
|
|
outString.seek(0)
|
|
self.doCSVImport(outString)
|
|
|
|
# clear all schedules
|
|
delXML = ROOT(
|
|
*[E.Schedules({"action": "DD", "scheduleID": str(ii)})
|
|
for ii in range(1, 8)])
|
|
try:
|
|
self.doXMLRequest(delXML)
|
|
except RemoteError:
|
|
# don't care about failure to delete, they probably just didn't exist
|
|
pass
|
|
|
|
# load new schedules
|
|
self.doXMLRequest(schedules)
|
|
|
|
def getSchedules(self):
|
|
# TODO: might be able to do in one request
|
|
schedules = self.doXMLRequest(ROOT(
|
|
E.Schedules({"action": "LR"})))
|
|
etree.dump(schedules)
|
|
|
|
data = self.doXMLRequest(ROOT(
|
|
*[E.Schedules({"action": "LR",
|
|
"scheduleID": schedule.attrib["scheduleID"]})
|
|
for schedule in schedules[0]]))
|
|
return ROOT(E_corp.Schedules({"action": "AD"},
|
|
*[s[0] for s in data]))
|
|
|
|
def sendCardFormat(self, formatName, templateID, facilityCode):
|
|
# TODO: add ability to delete formats
|
|
# delete example: <hid:CardFormats action="DD" formatID="7-1-244"/>
|
|
|
|
el = ROOT(
|
|
E.CardFormats({"action": "AD"},
|
|
E.CardFormat({"formatName": formatName,
|
|
"templateID": str(templateID)},
|
|
E.FixedField({"value": str(facilityCode)}))))
|
|
return self.doXMLRequest(el)
|
|
|
|
def lockOrUnlockDoor(self, lock=True):
|
|
el = ROOT(
|
|
E.Doors({"action": "CM",
|
|
"command": "lockDoor" if lock else "unlockDoor"}))
|
|
return self.doXMLRequest(el)
|
|
|
|
def getStatus(self):
|
|
el = ROOT(
|
|
E.Doors({"action": "LR", "responseFormat": "status"}))
|
|
xml = self.doXMLRequest(el)
|
|
relayState = xml.find('./{*}Doors/{*}Door').attrib['relayState']
|
|
return "unlocked" if relayState == "set" else "locked"
|