shark
/
Net2Scripting
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

First open source version.

Browse Source
master
Cor Kalis 6 years ago
commit 7ce577bb2f
45 changed files with 3041 additions and 0 deletions
Show Stats Download Patch File Download Diff File

8
.gitignore vendored
Unescape Escape View File

@ -0,0 +1,8 @@
build
dist
dist_installer
docs/*.html
docs/index.htm
*.pyc
#*#
*~

88
README.md
Unescape Escape View File

@ -0,0 +1,88 @@
Net2Scripting
===
Python scripting for the Paxton Net2 SDK.
This project builds a setup file, that will install a Python scripting environment
that can communicate with Paxton Net2, using the SDK (or a direct DB connection).
### **Read this first**
The scripts in this package are currently quite inflexible, in that they assume
that packages are installed in particular directories. For future versions this
should be changed to maybe an environment setting or a 'user' file containg the
definitions. (this file should be in .gitignore)
For now, the install directories are specificly marked in the requirements.
### Requirements
The NetScripting project relies on the following:
* An x86 Windows environment (64-bit did not work). Windows7-32bit should be ok.
* DotNet 4.0
- Installing Paxton Net2 V5.x should take care of that
- The Paxton Net2 OEMClientLibrary dlls (v5.01) are included in this package.
* Python x86 3.4.4
- Is the Python environment
- Download from http://www.python.org
- Install in C:\WinPrg\Python34\ !!!
* pythonnet-2.3.0-cp34-cp34m-win32.whl
- Is Python .net extension
- Download from http://pythonnet.sourceforge.net
- Install with pip install <wheel file>
* py2exe-0.9.2.2-py33.py34-none-any.whl
- To create a python executable
- Download from http://www.py2exe.org
- Install with pip install <wheel file>
- Remove 'clr' in c:/WinPrg/Python34/lib/site-packages/py2exe/hooks.py !!!
- Note that py2exe does/did not support 3.5 and up yet
* Inno Setup 5.4.2 (higher versions most likely also work)
- To create an installer (uses the inno_script.iss)
- Download from http://www.jrsoftware.org
- Install in C:\Program Files\Inno Setup 5\ !!!
The following convenience batch files are available:
* run.bat
- To run de scripting tool while testing sample code
- It will wait after execution and run again after pressing enter (ctrl C=exit)
- Do no just click it in explorer; this will not work
- Create a sendTo shortcut to it, to execute python scripts from the explorer.
+ type "shell:sento" in the file explorer address bar to access the
SentTo folder
+ Create shortcut to run.bat and call it "Test with Net2Scripting"
* make.bat
- Creates an exe-based distro in ./dist
- You can click it in explorer
* build_installer.bat
- Requires make.bat to run first
- Creates a 'SetupNet2Scripting.exe' in ./dist_installer
- You can click it in explorer
Beware of the following:
* That the doc's generated, need to be defined in src/docgen.py
* That for py2exe, new modules must be defined in 'Net2Scripting.py',
to have py2exe include them in the exe generation.
* That the makefile.bat will show warnings (which we ignore for now):
```
8 missing Modules
------------------
? Paxton imported from net2xs
? _dbm imported from dbm.ndbm
? log4net imported from pylog4net
? multiprocessing.SimpleQueue imported from concurrent.futures.process
? netbios imported from uuid
? win32evtlog imported from logging.handlers
? win32evtlogutil imported from logging.handlers
? win32wnet imported from uuid
```
### Configuration
Runtime configuration file is: src/Net2Scripting.exe.config
(complies to the dotnet standard)
### Version updates
To update the version number, adjust the following files:
* inno_script.iss
* src/settings.py
### License
MIT

10
build_installer.bat
Unescape Escape View File

@ -0,0 +1,10 @@
@echo off
del dist_installer\* /s/q > nul
rem taskkill /IM explorer.exe /F
rem cd /d %userprofile%\AppData\Local
rem del IconCache.db /a
rem start explorer.exe
"C:\Program Files\Inno Setup 5\ISCC.exe" inno_script.iss
pause

BIN
docs/Net2Scripting.png
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 1.2 KiB

121
docs/README.txt
Unescape Escape View File

@ -0,0 +1,121 @@
Net2Scripting
=============
The Net2Scripting program is an add-on for the Paxton Net2 software.
It offers the possibility to run Python scripts against the Net2 SDK or
straight against the Net2 database.
Using the SDK, you can access remote Net2 servers. Using the direct database
interface, only local access is possible.
Scripting allows ad-hoc functionality to be created, without compilation.
Useful for proof of concepts, filling Net2 with test data, simple door
control, Net2 data exports, etc.
REQUIREMENTS
* A PC running a recent Windows version. The software has been tested on
WinXP(SP3), Win7 and Win8.
* Administrator rights.
* The Net2 V5.1 or higher software package.
INSTALLATION
* Run the installer program, which will create an IntoAccess folder in the
Programs menu with a shortcut to Net2Scripting, install the executable,
libraries, samples and documentation.
* Optionally edit the Net2Scripting.exe.config file that can be found in the
program installation folder.
(by default it contains a reference to a sample script
'samples/user_script.py')
USAGE
* Option 1: Edit the Net2Scripting.exe.config file, changing the
'user_script' setting to point to your script. Run the Net2Scripting
program.
* Option 2: Edit the Net2Scripting.exe.config file, adding more entries to
the 'appSetting' section. Run the Net2Scripting program adding the parameter
/k=<keyword>, where <keyword> points to the script you want to execute.
* Option 3: Run the Net2Scripting program, adding the path to the script you
wish to execute as parameter.
* If you run the Net2Scripting using the explorer, you may want the window
containing the script result to remain open. Set the appSetting
'confirm_wait' to 'true' to have it wait for the <enter> key before closing.
If you want to run a script using a scheduler, set this option to 'false'.
* By default, the application logs fairly much and both to the console as
well as to a (rolling) file. Change the root 'level' value from DEBUG to
INFO, WARN, ERROR or FATAL to make it less verbose. You can add or remove
any log4net conpatible log appenders to suit your needs.
Beware that the scripting program locks the log file, so if you run more
than one instance, it will complain about not being able to log.
DOCUMENTATION / HELP
* Some pydoc generated 'help' for the API is placed in the 'docs'
subdirectory of the installation folder. It does not hurt however to also
keep the Net2 API doc at hand while scripting.
* A number of script samples are provided, meant for those with some basic
knowledge of Python. For a full Python language reference, check out
http://www.python.org/doc/
Not all Net2 SDK functions are available through the python api yet.
If you would like to add stuff however, check out the'inheritance.py'
sample to get you started.
* The (free) Net2Query tool, available at http://www.intoaccess.com, can
also of use to get some insight on available tables, column names and
data types.
BACKGROUND
* Originally created by CRC Value.
* Maintained (sort of) by IntoAccess from July 2017 - May 2018.
* Open source from May 2018 onward.
* The package websites:
- http://www.net2scripting.nl
- https://github.com/cor-kalis/Net2Scripting
ACKNOWLEDGEMENTS
* The Net2Scripting program is created using the following libraries/tools:
- Python => http://www.python.org
- Python.Net => http://pythonnet.sourceforge.net
- Py2exe => http://www.py2exe.org
- Paxton.Net2.OEMClientLibrary => http://www.paxton.co.uk
- Inno Setup => http://www.jrsoftware.org
LICENSE
* MIT
DISCLAIMER
* You will use this software at your own risk. No responsibility is accepted for
any damage it may accidentally cause.
HISTORY
* V1.0 was published in April 2013. Initial version
* V1.1 was published in July 2013.
- Fixes a module scope issue (imported modules not recognized in functions).
- User image manipulation.
- Confirm wait, even on a sys.exit().
- Allow remote native db access.
- Net2Plus module detection using udp broadcasts.
- SQLServer detection using udp broadcasts.
- Default values for add_user and modify_user (less hassle).
* V1.2 was published in July 2013.
- Enable linecache.getline by default, so 'list' in pdb.set_trace() shows
something useful. (disable in config by setting 'linecache_enable' to false)
- Added __scripting_version__ var, to test script compatibility.
- Fixed bug in Net2XS.deactivate_user fuction.
* V1.3 was published in September 2013.
- Fix / implement acu monitoring
- Fix get_users / get_user_by_id bug
* V1.4 was published in October 2013.
- Fix password encryption on remote native database connection
* V2.0 was published in April 2014.
- Net2 V5 compatible.
* V2.1 was published in July 2014.
- Fix hard coded event and subtype bug in add_event_record.
* V2.2 was published in October 2014.
- Removed misplaced delete_image parameter in add_user
* V2.3 was published in December 2015.
- Upgraded to Python 2.7.10 and included most standard Python modules.
* V2.4 was published in August 2016.
- Added the option to use either a dotnet or python date time objects for
the add_user and modify_user methods.
- Added delete_card and get_cards methods.
* V3.0 was published in August 2017.
- Upgrade to Python 3.4.4
* V4.0 was published in May 2018.
- Made open source.

47
inno_script.iss
Unescape Escape View File

@ -0,0 +1,47 @@
; Script generated by the Inno Setup Script Wizard.
; SEE THE DOCUMENTATION FOR DETAILS ON CREATING INNO SETUP SCRIPT FILES!
#define MyAppName "Net2Scripting"
#define MyAppVersion "4.0"
#define MyAppPublisher "Net2Scripting"
#define MyAppURL "http://www.net2scripting.nl/"
[Setup]
; NOTE: The value of AppId uniquely identifies this application.
; Do not use the same AppId value in installers for other applications.
; (To generate a new GUID, click Tools | Generate GUID inside the IDE.)
AppId={{3940C8D4-5D5F-43CF-B92E-4086ED2198B2}
AppName={#MyAppName}
AppVersion={#MyAppVersion}
;AppVerName={#MyAppName} {#MyAppVersion}
AppPublisher={#MyAppPublisher}
AppPublisherURL={#MyAppURL}
AppSupportURL={#MyAppURL}
AppUpdatesURL={#MyAppURL}
DefaultDirName={pf}\Paxton Access/{#MyAppName}
DefaultGroupName=Paxton Access
InfoAfterFile=.\dist\docs\README.txt
OutputDir=.\dist_installer
OutputBaseFilename=SetupNet2Scripting
SetupIconFile=.\dist\resources\Net2Scripting.ico
Compression=lzma
SolidCompression=yes
[Languages]
Name: "english"; MessagesFile: "compiler:Default.isl"
[Tasks]
Name: "desktopicon"; Description: "{cm:CreateDesktopIcon}"; GroupDescription: "{cm:AdditionalIcons}"; Flags: unchecked
[Files]
Source: ".\dist\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs createallsubdirs
; NOTE: Don't use "Flags: ignoreversion" on any shared system files
[Icons]
Name: "{group}\Net2Scripting"; Filename: "{app}\{#MyAppName}.exe"
Name: "{group}\Net2Scripting API"; Filename: "{app}\docs\index.htm"
Name: "{commondesktop}\Net2Scripting"; Filename: "{app}\{#MyAppName}.exe"; Tasks: desktopicon
[Run]
[UninstallRun]

21
make.bat
Unescape Escape View File

@ -0,0 +1,21 @@
@echo off
echo --- Cleaning...
del build\* /s/q > nul
del dist\* /s/q > nul
del docs\*.html /s/q > nul
echo --- Cleaning done
echo --- Generating docs
cd src
c:/WinPrg/Python34/python docgen.py
cd ..
echo --- Generating docs done
echo --- Building executable
cd src
c:/WinPrg/Python34/python setup.py py2exe
cd ..
echo --- Building executable done
pause

BIN
resources/Net2Scripting.ico
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 320 KiB

6
run.bat
Unescape Escape View File

@ -0,0 +1,6 @@
@echo off
cd src
:again
c:\WinPrg\Python34\python Net2Scripting.py %1
pause
goto again

BIN
runtime/Python.Runtime.dll
View File

Binary file not shown.

43
src/Net2Scripting.exe.config
Unescape Escape View File

@ -0,0 +1,43 @@
<?xml version='1.0' encoding='utf-8'?>
<configuration>
<configSections>
<section name="log4net" type="log4net.Config.Log4NetConfigurationSectionHandler, log4net" />
</configSections>
<appSettings>
<add key="user_script" value="samples/user_script.py" />
<add key="confirm_wait" value="true" />
<add key="log_stacktrace" value="true" />
<add key="enable_linecache" value="true" />
</appSettings>
<log4net>
<appender name="RollingFileAppender" type="log4net.Appender.RollingFileAppender">
<param name="File" value="${ALLUSERSPROFILE}/Application Data/IntoAccess/Net2Scripting/Net2Scripting.log"/>
<param name="AppendToFile" value="true"/>
<param name="rollingStyle" value="Size"/>
<param name="maxSizeRollBackups" value="10"/>
<param name="maximumFileSize" value="10MB"/>
<layout type="log4net.Layout.PatternLayout">
<param name="Header" value=""/>
<param name="Footer" value=""/>
<param name="ConversionPattern" value="%d [%c,%t] %-5p %m%n"/>
</layout>
</appender>
<appender name="ConsoleAppender" type="log4net.Appender.ConsoleAppender">
<layout type="log4net.Layout.PatternLayout">
<param name="Header" value="[Header]\r\n" />
<param name="Footer" value="[Footer]\r\n" />
<param name="ConversionPattern" value="%d [%c,%t] %-5p %m%n" />
</layout>
</appender>
<root>
<level value="DEBUG" />
<appender-ref ref="RollingFileAppender" />
<appender-ref ref="ConsoleAppender" />
</root>
<logger name="Paxton">
<level value="ERROR" />
<appender-ref ref="RollingFileAppender" />
<appender-ref ref="ConsoleAppender" />
</logger>
</log4net>
</configuration>

329
src/Net2Scripting.py
Unescape Escape View File

@ -0,0 +1,329 @@
"""
Main program for Net2Scripting.
"""
import os
import sys
import settings
import traceback
# This is just to force py2exe to include these packages
if False == True:
import net2xs
import net2dbxs
import network
# Not required but nice to have
import abc
import aifc
import antigravity
import argparse
import ast
import asynchat
import asyncio
import asyncore
import base64
import bdb
import binhex
import bisect
import bz2
import cProfile
import calendar
import cgi
import cgitb
import chunk
import cmd
import code
import codecs
import codeop
import collections
import colorsys
import compileall
import concurrent
import configparser
import contextlib
import copy
import copyreg
# import crypt
import csv
import ctypes
# import curses
import datetime
# import dbm
import decimal
import difflib
import dis
import distutils
import doctest
import threading
import email
import encodings
# import ensurepip
import enum
import filecmp
import fileinput
import fnmatch
import formatter
import fractions
import ftplib
import functools
import genericpath
import getopt
import getpass
import gettext
import glob
import gzip
import hashlib
import heapq
import hmac
import html
import http
import idlelib
import imaplib
import imghdr
import imp
import importlib
import inspect
import io
import ipaddress
import json
import keyword
import lib2to3
import linecache
import locale
import logging
import lzma
# import macpath
# import macurl2path
import mailbox
import mailcap
import mimetypes
import modulefinder
import msilib
# import multiprocessing
import netrc
import nntplib
import ntpath
import nturl2path
import numbers
import opcode
import operator
import optparse
import os
import pathlib
import pdb
import pickle
import pickletools
import pipes
import pkgutil
import platform
import plistlib
import poplib
import posixpath
import pprint
import profile
import pstats
import pty
import pyclbr
import pydoc
import pydoc_data
import queue
import quopri
import random
import re
import reprlib
import rlcompleter
import runpy
import sched
import selectors
import shelve
import shlex
import shutil
import site
# import smtpd
import smtplib
import sndhdr
import socket
import socketserver
import sqlite3
# import compile
# import constants
# import parse
import ssl
import stat
import statistics
import string
import stringprep
import struct
import subprocess
import sunau
import symbol
import symtable
import sysconfig
import tabnanny
import tarfile
import telnetlib
import tempfile
import test
import textwrap
import this
import threading
import timeit
import tkinter
import token
import tokenize
import trace
import traceback
import tracemalloc
import tty
import turtle
import turtledemo
import types
import unittest
import urllib
import uu
import uuid
import venv
import warnings
import wave
import weakref
import webbrowser
import wsgiref
import xdrlib
import xml
import xmlrpc
import zipfile
# Setup logging first
from pylog4net import Log4Net
try:
Log4Net.read_config(settings.CONFIG_FILE)
except Exception as e:
print('Log error: %s' % (str(e)))
sys.exit(1)
from config import Config
try:
cfg = Config(settings.CONFIG_FILE)
except Exception as e:
print('Config error: %s' % (str(e)))
sys.exit(1)
# Running as py2exe: check if linecache should be enabled
# if hasattr(sys, 'frozen'):
# if cfg.get("enable_linecache", True, bool):
# # Monkey patch it back to the original value
# import linecache
# linecache.getline = linecache.orig_getline
# del linecache
# Log4net logger
logger = Log4Net.get_logger('Net2Scripting')
def run(script_file):
"""Run script file
"""
try:
if not os.path.isfile(script_file):
raise Exception("User script '%s' does not exist!" % (script_file))
# Get script dir and add it to the module search path
sys.path.insert(0, os.path.dirname(script_file))
print("Press '<ctrl> C' to quit...")
logger.Debug("Calling user script '%s'" % (script_file))
try:
# Strip Net2Scripting param from arguments
sys.argv = sys.argv[1:]
# Explicitly provide dict, to prevent scope issues
with open(script_file) as f:
code = compile(f.read(), script_file, 'exec')
exec(code,
{"__name__": "__main__",
"__file__": script_file,
"__scripting_version__": settings.VERSION})
except KeyboardInterrupt:
logger.Debug("User interrupt")
except Exception as ex:
msg = "Fatal error: " + str(ex)
if cfg.get("log_stacktrace", default=False, vtype=bool):
msg += "\n" + traceback.format_exc()
logger.Fatal(msg)
return 1
def run_key(script_key):
"""Run file referenced by setting keyword
"""
try:
script_key = script_key.strip()
if not script_key:
raise "The provided script keyword is empty!"
# Fetch setting from config
cfg.check_required([(script_key, str)])
return run(cfg.get(script_key))
except Exception as ex:
logger.Fatal("Fatal error: " + str(ex))
return 1
def confirm_wait():
"""Wait for user confirmation
"""
# Fetch setting from config
wait = cfg.get("confirm_wait", default=False, vtype=bool)
if wait:
print()
input("Press enter to continue: ")
# Main program entry
if __name__ == '__main__':
ret_val = 0
try:
if len(sys.argv) > 1:
arg = sys.argv[1]
if arg == "/?":
print()
print("Net2Scripting V%s, by Net2 Solutions" % (settings.VERSION))
print()
print(" /? = help")
print(" /k:keyword = run script referenced by appSettings keyword")
print(" <script file path> = run given script")
print()
print(" By default, the script referenced by the 'user_script' setting is run.")
print(" Settings are located in the 'Net2Scripting.exe.config' file.")
print(" Also modify this file if you prefer different log settings.")
print
ret_val = 1
# Run by keyword
elif arg.startswith("/k:"):
ret_val = run_key(arg.split(":", 1)[1])
# Run as file
elif not arg.startswith("/"):
ret_val = run(arg)
# Unknown option
else:
print("Unknown parameter: " + arg)
ret_val = 1
else:
# Run as default
ret_val = run_key("user_script")
finally:
# Wait for user confirmation when required
confirm_wait()
# Return with value
sys.exit(ret_val)

99
src/config.py
Unescape Escape View File

@ -0,0 +1,99 @@
"""
Module to read dot type config settings from python
"""
import os
import re
from datetime import datetime
from xml.etree import ElementTree as ET
class ConfigError(Exception):
"""Config exception class
"""
pass
class Config(object):
"""Class for reading and writing to the config file
"""
def __init__(self, config_file):
"""Read config file
"""
self.config_file = config_file
self._app_settings = {}
if not os.path.isfile(config_file):
raise ConfigError('Failed to find config file "%s"' % config_file)
# Read document
tree = ET.ElementTree()
tree.parse(config_file)
# Obtain appSettings
element = tree.find('appSettings')
if element is None:
raise ConfigError(
'Required "appSettings" section is missing in config file' %
self.config_file)
for item in element.findall('add'):
if 'key' in item.attrib and 'value' in item.attrib:
key = item.attrib['key']
val = item.attrib['value']
self._app_settings[key] = val
else:
raise ConfigError(
'Encountered appSetting with missing attributes')
def _parse_as_datetime(self, val):
"""Attempt to parse val as datetime
"""
if re.match(r'\d+:\d+:\d+', val):
fmt = '%H:%M:%S'
elif re.match(r'\d\d\d\d-\d+-\d+', val):
fmt = '%Y-%m-%d'
elif re.match(r'\d\d\d\d-\d+-\d+ \d+:\d+:\d+', val):
fmt = '%Y-%m-%d %H:%M:%S'
else:
raise ValueError('Illegal datetime format')
return datetime.strptime(val, fmt)
def check_required(self, item_list):
for key, vtype in item_list:
val = self._app_settings.get(key)
if val is None:
raise ConfigError('Required setting for "%s" is missing' % key)
if vtype != str:
try:
if vtype == datetime:
self._parse_as_datetime(val)
else:
vtype(val)
except ValueError:
raise ConfigError(
'Required setting for "%s" (%s), is no %s type' %
(key, val, vtype.__name__))
def get(self, key, default=None, vtype=str):
"""Obtain typecasted value
"""
val = self._app_settings.get(key, default)
if val is None:
return None
if vtype == str:
return val
if vtype == datetime:
try:
return self._parse_as_datetime(val)
except ValueError:
return default
if vtype == bool:
return val.lower() == 'true'
try:
return vtype(val)
except ValueError:
return default

78
src/docgen.py
Unescape Escape View File

@ -0,0 +1,78 @@
"""
Helper module to generate the pydoc based
"""
import os
import settings
import sys
from glob import glob
from pydoc import writedoc
# So we can find the libs
sys.path.append('libs')
PACKAGES = ["net2base",
"net2xs", "net2xs.conversions", "net2xs.deftypes",
"net2dbxs",
"network", "network.net2plus", "network.sqlserver"]
INDEX = "index.htm"
def read_file(file_name):
"""Read file content
"""
with open(file_name) as f:
return f.read()
def alter_doc(content):
"""Change doc content
- Strip reference to local python file
- Replace index with our own index file
"""
pos1 = content.find("<a href=\"file:///")
if pos1 >= 0:
pos2 = content.find("</a>", pos1)
content = content[:pos1] + content[pos2 + len("</a>"):]
content = content.replace('<a href=".">index</a>',
'<a href="index.htm">index</a>')
return content
def alter_index(content):
"""Change index content
- Replace version place holder with actual version
"""
content = content.replace('#VERSION#', settings.VERSION)
return content
def write_file(file_name, content):
"""Write content as file
"""
with open(file_name, "w") as f:
f.write(content)
if __name__ == "__main__":
# Create docs
for p in PACKAGES:
writedoc(p)
# For each html, generated into src, write altered to docs
for src in glob("*.html"):
dest = '../docs/' + src
content = read_file(src)
content = alter_doc(content)
write_file(dest, content)
os.remove(src)
# Write index with correct version to docs
content = read_file(INDEX)
dest = '../docs/' + INDEX
content = alter_index(content)
write_file(dest, content)

29
src/index.htm
Unescape Escape View File

@ -0,0 +1,29 @@
<html>
<head>
<style type="text/css">
body {
color: #142313;
font-family: Tahoma, sans-serif;
font-size: 14px;
font-weight: normal;
text-align: left;
color: darkSlateGrey;
background-color: #f0f0f8;
}
</style>
</head>
<body>
<center>
<h2>Net2Scripting V#VERSION# API documentation</h2>
<br>
<img src="Net2Scripting.png">
<br>
<br>
<p><a href="net2xs.html">Net2XS for access through the SDK</a></p>
<p><a href="net2dbxs.html">Net2DBXS for direct database access</a></p>
<p><a href="network.html">Network tools related to Net2 access</a></p>
<p><a href="README.txt">Application README</a></p>
<br>
<p><h5><a href="http://www.net2scripting.nl">www.net2scripting.nl</a></h5></p>
</body>
</html>

BIN
src/libs/log4netdll/log4net.dll
View File

Binary file not shown.

BIN
src/libs/paxton/Paxton.Cryptography.dll
View File

Binary file not shown.

BIN
src/libs/paxton/Paxton.Net2.OEMClientLibrary.dll
View File

Binary file not shown.

BIN
src/libs/paxton/Paxton.Net2.Utils.dll
View File

Binary file not shown.

BIN
src/libs/paxton/Paxton.Remoting.Channel.dll
View File

Binary file not shown.

BIN
src/libs/paxton/Paxton.Remoting.RemotingLayerInterfaces.dll
View File

Binary file not shown.

43
src/net2base/__init__.py
Unescape Escape View File

@ -0,0 +1,43 @@
"""
Module offering base class for net2 access
"""
class Net2Base(object):
"""Net2 access base class
"""
@classmethod
def safe_get_row_val(cls, row, col):
"""Null aware fetching of row data
Returns None if the database column is NULL
"""
if row.IsNull(col):
return None
return row[col]
@classmethod
def dataset_to_str(cls, dataset):
"""Convert dataset to string presentation
"""
if not dataset:
return "None"
if dataset.Tables.Count < 1:
return "Empty"
res = []
for table in dataset.Tables:
res.append("Table: %s" % (table.TableName))
cols = []
for col in table.Columns:
cols.append(col.ColumnName)
for row in table.Rows:
row_res = []
for i in range(len(cols)):
row_res.append("%s=%s" % (cols[i], row.get_Item(i)))
res.append(", ".join(row_res))
return "\n".join(res)

146
src/net2dbxs/__init__.py
Unescape Escape View File

@ -0,0 +1,146 @@
"""
The module that offers direct access to the Paxton Net2 database,
not using the SDK.
"""
import clr
from threading import RLock
clr.AddReference('System.Data')
from System.Data.SqlClient import SqlConnection, SqlDataAdapter, SqlCommand
from System.Data import DataSet, CommandType, ParameterDirection
from net2base import Net2Base
from pylog4net import Log4Net
class Net2DBXSException(Exception):
"""Exception class for net2 database xs
"""
pass
class DBCommand(object):
"""With wrapper for SqlCommand
"""
def __init__(self, query, con):
self._cmd = SqlCommand(query, con)
def __enter__(self):
return self._cmd
def __exit__(self, type, value, traceback):
if (self._cmd):
self._cmd.Dispose()
class Net2DBXS(Net2Base):
"""Net2 direct database access class
"""
# Class variables
_logger = Log4Net.get_logger('Net2DBXS')
_lock = RLock()
def __init__(self):
"""Class constructor
"""
self._con = None
self._connected = False
def __enter__(self):
"""With enter
"""
return self
def __exit__(self, type, value, traceback):
"""With exit
"""
self.dispose()
def _check_connect(self):
"""Check database connection
Raises Net2DBXSException if not connected.
"""
if not self._connected:
raise Net2DBXSException("Not connected to the database")
def connect(self,
user_name="sdk_user",
password="E56ABED4-2918-44F9-A110-71B61B47142A",
db='Net2',
server=None):
"""Connect to database
If the server parameter is None, named pipes are used.
For remote connections, the server parameter should contain the
windows name.
"""
self.dispose()
with Net2DBXS._lock:
if not server:
Net2DBXS._logger.Debug('Connecting through named pipes')
self._con = SqlConnection(
r"server=\\.\pipe\MSSQL$NET2\sql\query;database=%s;uid=%s;password=%s" %
(db, user_name, password))
else:
Net2DBXS._logger.Debug('Connecting over the network')
self._con = SqlConnection(
r"server=%s\NET2;database=%s;uid=%s;password=%s" %
(server, db, user_name, password))
self._con.Open()
self._connected = True
Net2DBXS._logger.Debug('Connected')
def query_db(self, query):
"""Perform database query and return data in dataset
Returns a dataset.
"""
self._check_connect()
with DBCommand(query, self._con) as cmd:
adapter = SqlDataAdapter()
adapter.SelectCommand = cmd
data = DataSet()
adapter.Fill(data)
return data
def run_stored_procedure(self, sp_name, **params):
"""Run stored procedured
Returns the stored procedure result value.
"""
with DBCommand(sp_name, self._con) as cmd:
cmd.CommandType = CommandType.StoredProcedure
# Add parameters
for name, value in params.iteritems():
cmd.Parameters.AddWithValue("@%s" % name, value)
# Add return code
cmd.Parameters.AddWithValue("@result", -1)
cmd.Parameters["@result"].Direction = ParameterDirection.ReturnValue
# Execute
cmd.ExecuteNonQuery()
return cmd.Parameters["@result"].Value
def get_all_tables(self):
"""Convenience function to retrieve all known tables / views
Returns a dataset.
"""
return self.query_db(
"select * from INFORMATION_SCHEMA.Tables order by TABLE_NAME")
def dispose(self):
"""Dispose database connection
"""
with Net2DBXS._lock:
if self._con:
try:
self._con.Close()
self._con = None
Net2DBXS._logger.Debug('Disposed')
except:
pass

999
src/net2xs/__init__.py
Unescape Escape View File

@ -0,0 +1,999 @@
"""
The module that offers access to the Paxton Net2 server
"""
import clr
import os
import sys
import settings
from net2xs.conversions import date_time_to_net, flex_date_time_to_net, \
time_zones_to_py, access_levels_to_py, access_level_detail_to_py
from datetime import datetime
from net2base import Net2Base
from pylog4net import Log4Net
from threading import RLock
from System import Array
from System.Data import DataSet
from System.Reflection import Assembly
# Minimal required net2 version
MIN_NET2_VERSION = 501
# Paxton assembly
PAXTON_ASSEMBLY = 'Paxton.Net2.OEMClientLibrary'
def readable_min_version():
"""Show a user readable minimum version
"""
major = MIN_NET2_VERSION / 100
minor = MIN_NET2_VERSION - major * 100
return "%d.%d" % (major, minor)
class Net2XSException(Exception):
"""Exception class for net2 xs
"""
pass
# The code below is required to determine if the already installed
# Net2 version can be used, or that the packaged version is required.
try:
# Obtain paxton assembly reference
asm = Assembly.LoadWithPartialName(PAXTON_ASSEMBLY)
# Found: check the version
if asm:
ver = asm.GetName().Version
if ver.Major * 100 + ver.Minor < MIN_NET2_VERSION:
raise Net2XSException(
'Only Net2 V%s or higher is supported' %
readable_min_version())
# Not found: enable the packaged paxton libs
else:
# Add path lib path to search path
PAXTON_LIB_DIR = os.path.join(settings.LIB_DIR, 'paxton')
if PAXTON_LIB_DIR not in sys.path:
sys.path.append(PAXTON_LIB_DIR)
try:
Assembly.LoadWithPartialName(PAXTON_ASSEMBLY)
clr.AddReference(PAXTON_ASSEMBLY)
from Paxton.Net2.OemClientLibrary import OemClient as OC
from Paxton.Net2.OemClientLibrary import AccessLevelDetailSet
from Paxton.Net2.OemClientLibrary import TimezonesSet
from Paxton.Net2.OemClientLibrary import EventViewEnums
except:
raise Net2XSException('Failed to load the library')
except Exception as e:
Log4Net.get_logger('Net2XS').Fatal('Paxton error: %s' % str(e))
sys.exit(1)
# end of paxton loading
class Net2XS(Net2Base):
"""Net2 Access class
"""
# Class variables
_logger = Log4Net.get_logger('Net2XS')
_lock = RLock()
def __init__(self, host='localhost', port=8025):
"""Class constructor
"""
self._client = None
self._host = host
self._port = port
self._connected = False
self._on_acu_event = None
def __enter__(self):
"""With enter handler
"""
return self
def __exit__(self, type, value, traceback):
"""With exit handler
"""
self.dispose()
def _check_client(self):
"""Check is client connection is valid
"""
if not self._client or not self._connected:
raise Net2XSException('Not connected')
def authenticate(self, user_id, password):
"""Authenticate to Net2
"""
with Net2XS._lock:
# Save for re authentication
self._user_id = user_id
self._password = password
self.dispose()
Net2XS._logger.Debug('Connecting to net2 server on %s:%d' %
(self._host, self._port))
self._client = OC(self._host, self._port)
methods = self._client.AuthenticateUser(user_id, password)
if not methods:
raise Net2XSException('Authentication failed: ' +
self._client.LastErrorMessage)
else:
self._connected = True
# Add disconnect reconnect handlers
self._client.Net2ServerReconnected += (
OC.Net2ServerReconnectedHandler(self._reconnected))
self._client.Net2ServerDisconnected += (
OC.Net2ServerDisconnectedHandler(self._disconnected))
# Add acu event handler
self._client.Net2AccessEvent += (
OC.Net2AcuEventHandler(self._acu_event))
Net2XS._logger.Debug('Authenticated')
@property
def client_version(self):
"""Client version
"""
asm = Assembly.GetAssembly(OC)
ver = asm.GetName().Version
return (ver.Major, ver.Minor)
def query_db(self, query):
"""Perform a db query
Returns a dataset
"""
with Net2XS._lock:
self._check_client()
return self._client.QueryDb(query)
def are_doors_being_synchronised(self):
"""Return if synchronization is taking place
Returns a boolean
"""
with Net2XS._lock:
self._check_client()
return self._client.AreDoorsBeingSynchronised()
def get_doors(self, device_address=-1):
"""Get all doors (ViewDoors)
Device_address is an optional integer
Returns a dataset
"""
with Net2XS._lock:
self._check_client()
if device_address < 0:
return self._client.ViewDoors().DoorsDataSource
else:
return self._client.ViewDoors(device_address).DoorsDataSource
def get_door_name(self, device_address):
"""Obtain door name
Returns a string
"""
with Net2XS._lock:
self._check_client()
dataset = self._client.ViewDoors(device_address).DoorsDataSource
if (not dataset or
dataset.Tables.Count < 1 or
dataset.Tables[0].Rows.Count == 0):
return None
return dataset.Tables[0].Rows[0].Name
def get_departments(self):
"""Get all departments (ViewDepartments)
Returns a dataset
"""
with Net2XS._lock:
self._check_client()
return self._client.ViewDepartments().DepartmentsDataSource
def close_door(self, device_address):
"""Close a door
Returns True on success
"""
with Net2XS._lock:
self._check_client()
return self._client.CloseDoor(device_address)
def hold_door_open(self, device_address):
"""Hold a door open
Returns True on success
"""
with Net2XS._lock:
self._check_client()
return self._client.HoldDoorOpen(device_address)
def control_door(self, device_address, relay, function, door_open_time, led_flash):
"""Control a door
Relay is 0 or 1, for relay 1 or 2.
Function 0 Close, 1 Timed open, 2 Hold Open.
Door_open_time in ms.
Led_flash see Net2 API.
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
return self._client.ControlDoorEx(
device_address, relay, function, door_open_time, led_flash)
def get_department_name(self, department_id):
"""Get department name by id
Returns a string or None if department was not found
"""
dataset = self.query_db(
'select DepartmentName from sdk.Departments'
' where DepartmentID=%d' % department_id)
if (not dataset or
dataset.Tables.Count < 1 or
dataset.Tables[0].Rows.Count == 0):
return None
return dataset.Tables[0].Rows[0][0]
def get_users_ex(self, name=None):
"""Get users, from the UsersEx view.
Name param is optional tuple (first_name, sur_name)
Returns a dataset
"""
query = 'select * from sdk.UsersEx'
if name and len(name) == 2:
first_name, sur_name = name
query = ("%s where FirstName='%s' and Surname='%s'" %
(query, first_name, sur_name))
return self.query_db(query)
def get_user_id_by_name(self, name):
"""Get user id by name
Name is a (first_name, sur_name) tuple
Returns the id or -1 if not found.
"""
dataset = self.get_users(name)
if (not dataset or
dataset.Tables.Count < 1 or
dataset.Tables[0].Rows.Count == 0):
return -1
return dataset.Tables[0].Rows[0].get_Item("UserID")
def get_users(self, name=None):
"""Get users, from the ViewUserRecords call
Name param is optional tuple (first_name, sur_name)
Returns a dataset
"""
with Net2XS._lock:
self._check_client()
wheres = ['Active=1']
if name and len(name) == 2:
first_name, sur_name = name
if first_name:
wheres.append("FirstName='%s'" % first_name)
if sur_name:
wheres.append("Surname='%s'" % sur_name)
return self._client.ViewUserRecords(
' and '.join(wheres)).UsersDataSource
def get_user_record(self, user_id):
"""Get user record, from the ViewUserRecords call
Returns an IUserView user record or None if not found
"""
with Net2XS._lock:
self._check_client()
# Fetch current user info
users = self._client.ViewUserRecords(
'UserID=%d' % user_id).UsersList()
if users.Count != 1 or not users.ContainsKey(user_id):
return None
return users[user_id]
def get_user_name(self, user_id):
"""Get user name
Returns a string or None when user is not found
"""
dataset = self.query_db(
'select Username from sdk.UsersEx where UserID=%d' % user_id)
if (not dataset or
dataset.Tables.Count < 1 or
dataset.Tables[0].Rows.Count == 0):
return None
return dataset.Tables[0].Rows[0][0]
def add_user(
self,
access_level_id=1,
department_id=0,
anti_passback_ind=False,
alarm_user_ind=False,
first_name=None,
middle_name=None,
sur_name=None,
telephone_no=None,
telephone_extension=None,
pin_code=None,
activation_date=None,
active=True,
fax_no=None,
expiry_date=None,
custom_fields=None,
user_picture=None):
"""Add user record
DateTime fields can be either python or dotnet objects.
If activation date is None, the current date will be used.
If expiry date is None, the user entry will not expire.
Custom_fields is a string array (15) of which the first element is
ignored.
Returns True on success.
"""
# If no user name is given at all: create one (required by Net2)
if not first_name and not sur_name and not middle_name:
first_name = "New"
sur_name = "User"
with Net2XS._lock:
self._check_client()
return self._client.AddUserRecord(
access_level_id,
department_id,
anti_passback_ind,
alarm_user_ind,
first_name,
middle_name,
sur_name,
telephone_no,
telephone_extension,
pin_code,
None,
flex_date_time_to_net(activation_date) or self.now_date,
0,
0,
active,
fax_no,
flex_date_time_to_net(expiry_date) or self.no_expiration_date,
custom_fields,
user_picture)
def modify_user(
self,
user_id,
access_level_id=None,
department_id=None,
anti_passback_ind=None,
alarm_user_ind=None,
first_name=None,
middle_name=None,
sur_name=None,
telephone_no=None,
telephone_extension=None,
pin_code=None,
activation_date=None,
active=None,
fax_no=None,
expiry_date=None,
custom_fields=None,
user_picture=None,
delete_image=False):
"""Modify user record
Fields omitted keep their original value.
Custom_fields is a string array (15) of which the first element is
ignored.
Providing None custom values in the array, leaves the original value
unchanged.
Returns True on success.
"""
# Fetch current user info
uview = self.get_user_record(user_id)
if not uview:
return False
with Net2XS._lock:
self._check_client()
return self._client.UpdateUserRecord(
user_id,
uview.AccessLevelId if access_level_id is None else access_level_id,
uview.DepartmentId if department_id is None else department_id,
uview.AntiPassbackUser if anti_passback_ind is None else anti_passback_ind,
uview.AlarmUser if alarm_user_ind is None else alarm_user_ind,
first_name,
middle_name,
sur_name,
telephone_no,
telephone_extension,
pin_code,
None,
flex_date_time_to_net(activation_date) or uview.ActivationDate,
uview.Active if active is None else active,
fax_no,
flex_date_time_to_net(expiry_date) or uview.ExpiryDate,
custom_fields,
user_picture,
delete_image)
def delete_user(self, user_id):
"""Delete user record
Returns True on success
"""
with Net2XS._lock:
self._check_client()
return self._client.PurgeUser(user_id)
def deactivate_user(self, user_id):
"""Deactivate user. Preferred method iso delete.
Returns True on success
"""
return self.modify_user(
user_id=user_id,
access_level_id=0,
department_id=0,
pin_code="",
active=False)
def modify_user_access_level(self, user_id, access_level_id):
"""Alter user access level
Returns True on success
"""
return self.modify_user(
user_id=user_id,
access_level_id=access_level_id)
def modify_user_picture(self, user_id, user_picture):
"""Alter user picture. If user_picture is None, remove the picture.
Returns True on success
"""
return self.modify_user(
user_id=user_id,
user_picture=user_picture,
delete_image=True if not user_picture else False)
def get_area_ids(self, device_address):
"""Get area ids of a device
Returns a tuple with area id's (in, out)
"""
dataset = self.query_db(
'select b.ToAreaID from sdk.PeripheralNames a'
' inner join sdk.AreaGateways b on'
' a.PeripheralID=b.PeripheralID and'
' a.SerialNumber=%d order by a.SubAddress' %
(device_address))
if (not dataset or
dataset.Tables.Count < 1 or
dataset.Tables[0].Rows.Count != 2):
return None
return (dataset.Tables[0].Rows[0][0], dataset.Tables[0].Rows[1][0])
def get_device_addr_info(self):
"""Obtain all relevant device address info
Returns a dataset
"""
return self.query_db(
'select a.SerialNumber, a.SubAddress, a.PeripheralID, b.ToAreaID'
' from sdk.PeripheralNames a'
' inner join AreaGateways b on a.PeripheralID=b.PeripheralID'
' order by a.SerialNumber')
def get_time_slots(self, access_level_id, area_id):
"""Obtain allocated timeslots
Returns a dataset
"""
return self.query_db(
'select * from sdk.Timeslots'
' where TimezoneID='
'(select TimezoneID from AccessLevelMembers'
' where AccessLevelID=%d and AreaID=%d)' %
(access_level_id, area_id or -1))
def get_time_zones(self, time_zone_id=-1):
"""Obtain timezones, including slot data
Time_zone_id is optional. If not provided, all timezones are returned.
Returns a dataset
"""
with Net2XS._lock:
self._check_client()
if time_zone_id < 0:
return self._client.ViewTimezones().TimezonesDataSource
else:
return self._client.ViewTimezones(
time_zone_id, False).TimezonesDataSource
def get_py_time_zones(self, time_zone_id=-1):
"""Like get_time_zones, but returns a python array of TimeZone objects
Returns an array of TimeZone objects
"""
dataset = self.get_time_zones(time_zone_id)
return time_zones_to_py(dataset)
def add_time_zone(self, time_zone):
"""Add new time zone
Time_zone is a python TimeZone object type.
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
tz_set = TimezonesSet()
for slot in time_zone.slots:
tz_set.Tables[1].AddTimeslotsRow(
slot.id,
time_zone.id,
date_time_to_net(slot.start_time),
date_time_to_net(slot.end_time),
slot.day)
# To array
ts_array = Array[TimezonesSet.TimeslotsRow](
len(time_zone.slots) * [None])
for i, row in enumerate(tz_set.Tables[1].Rows):
ts_array[i] = row
# Dispose timezones set
tz_set.Dispose()
return self._client.AddTimezone(time_zone.name, ts_array)
def delete_time_zone(self, time_zone_id):
"""Remove time zone with given id
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
return self._client.DeleteTimezone(time_zone_id)
def update_time_zone(self, time_zone):
"""Update existing time zone
Time_zone is a python TimeZone object type.
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
tz_set = TimezonesSet()
nr_slots = len(time_zone.slots)
if nr_slots:
for slot in time_zone.slots:
tz_set.Tables[1].AddTimeslotsRow(
slot.id,
time_zone.id,
date_time_to_net(slot.start_time),
date_time_to_net(slot.end_time),
slot.day)
# Work around for bug "Update Timezone with empty timeslot" (forum)
# At least 1 row is required for the update to work correctly.
# Should be fixed in V4.21
# TODO: only perform for lower versions?
else:
nr_slots = 1
tz_set.Tables[1].AddTimeslotsRow(
-1,
time_zone.id,
date_time_to_net(datetime(1899, 12, 30, 23, 59, 59)),
date_time_to_net(datetime(1899, 12, 30, 23, 59, 59)),
0)
# To array (bit awkward in Python.Net)
ts_array = Array[TimezonesSet.TimeslotsRow](nr_slots * [None])
for i, row in enumerate(tz_set.Tables[1].Rows):
ts_array[i] = row
# Dispose timezones set
tz_set.Dispose()
return self._client.UpdateTimezone(time_zone.id, ts_array)
def get_time_zone_id_by_name(self, name):
"""Get time zone id of time zone with given name
Returns a string or None when the timezone could not be found.
"""
dataset = self.query_db(
"select TimezoneID from sdk.Timezones"
" where Name='%s'" % name)
if (not dataset or
dataset.Tables.Count < 1 or
dataset.Tables[0].Rows.Count == 0):
return None
return dataset.Tables[0].Rows[0][0]
def get_access_levels(self, prefix=None):
"""Obtain access levels, optinally profiding a name prefix filter
Prefix is an optional parameter to obtain only access levels with the
given prefix.
Details need to be fetched separately.
Returns a dataset.
"""
query = 'select * from sdk.AccessLevels'
if prefix:
query = "%s where AccessLevelName like '%s%%'" % (query, prefix)
return self.query_db(query)
def get_access_level_details(self, access_level_id):
"""Obtain access level details
Returns a dataset.
"""
query = 'select * from sdk.AccessLevelMembers where AccessLevelID=%d' % (access_level_id)
return self.query_db(query)
def get_py_access_levels(self, prefix=None):
"""Like get_access_level, but returning a python array of AccessLevel
objects including details!
Returns an array of AccessLevel objects.
"""
dataset = self.get_access_levels(prefix)
if not dataset:
return []
access_levels = access_levels_to_py(dataset)
for al in access_levels:
dataset = self.get_access_level_details(al.id)
if dataset:
access_level_detail_to_py(al, dataset)
return access_levels
def add_access_level(self, access_level):
"""Add new access level
Access_level is a python AccessLevel object type.
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
detail_set = AccessLevelDetailSet()
if access_level.details:
for detail in access_level.details:
detail_set.Tables[0].AddAccessLevelDetailRow(
None,
detail[0], # time zone id
None,
detail[1], # area id
None,
-1, # address
-1, # sub address
-1) # access level id
return self._client.AddAccessLevel(access_level.name, detail_set)
def delete_access_level(self, access_level_id):
"""Delete access level
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
return self._client.DeleteAccessLevel(access_level_id)
def update_access_level(self, access_level):
"""Update access level
Access_level is a python AccessLevel object type.
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
if access_level.details:
detail_set = AccessLevelDetailSet()
for detail in access_level.details:
detail_set.Tables[0].AddAccessLevelDetailRow(
None,
detail[0], # time zone id
None,
detail[1], # area id
None,
-1, # address
-1, # sub address
-1) # access level id
# Work around for similar bug as with the time zone slots
# When details are empty, existing data will not be cleared
# Therefore set TimezoneID to 0 for any existing entries
else:
ald = self._client.ViewAccessLevelDetail(access_level.id)
detail_set = ald.AccessLevelDetailsDataSource
for row in detail_set.Tables[0].Rows:
row.TimezoneID = 0
return self._client.UpdateAccessLevel(
access_level.id, access_level.name, detail_set)
def get_access_level_id_by_user(self, user_id):
"""Get access level id of given user
Returns an integer or None if the user could not be found
"""
with Net2XS._lock:
self._check_client()
dataset = self._client.ViewUserRecords(
'UserID=%d' % (user_id)).UsersDataSource
if (not dataset or
dataset.Tables.Count < 1 or
dataset.Tables[0].Rows.Count == 0):
return None
return dataset.Tables[0].Rows[0].AccessLevelID
def get_access_level_id_by_name(self, name):
"""Get access level id of level with given name
Returns a string or None if the accesslevel could not be found.
"""
dataset = self.query_db(
"select AccessLevelID from sdk.AccessLevels"
" where AccessLevelName='%s'" % name)
if (not dataset or
dataset.Tables.Count < 1 or
dataset.Tables[0].Rows.Count == 0):
return None
return dataset.Tables[0].Rows[0][0]
def get_operator_level(self, user_id):
"""Get operator level of the given user id
Returns an integer.
"""
with Net2XS._lock:
self._check_client()
return self._client.GetOperatorLevel(user_id)
def add_card(self, card_nr, type_id, user_id):
"""Add a new card to the given user
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
return self._client.AddCard(card_nr, type_id, user_id)
def delete_card(self, card_nr):
"""Delete a card with the given card_nr
Returns True on success (even if card_nr did not exist).
"""
with Net2XS._lock:
self._check_client()
return self._client.DeleteCard(card_nr)
def get_cards(self, user_id=None):
"""Obtain all 'cards' of the given user id
If no user_id is given, all cards are returned.
Returns a dataset.
"""
with Net2XS._lock:
self._check_client()
# Setup dataset to return
dataset = DataSet()
table = dataset.Tables.Add()
table.Columns.Add("UserID")
table.Columns.Add("CardNumber")
table.Columns.Add("LostCard")
table.Columns.Add("CardTypeID")
# Fetch normal cards
query = ("select * from sdk.Cards" +
" where CardTypeID is not null" +
" and CardTypeID <> 7" +
" and CardNumber <> 100000000 + UserID")
if type(user_id) is int:
query += " and UserID=%d" % (user_id)
card_dataset = self.query_db(query)
if card_dataset and card_dataset.Tables.Count > 0:
for row in card_dataset.Tables[0].Rows:
table.Rows.Add(row['UserID'],
row['CardNumber'],
row['LostCard'],
row['CardTypeID'])
# Fetch vehicle registrations
query = "select * from sdk.UserVehicleIndex"
if type(user_id) is int:
query += " where UserID=%d" % (user_id)
vehicle_dataset = self.query_db(query)
if vehicle_dataset and vehicle_dataset.Tables.Count > 0:
for row in vehicle_dataset.Tables[0].Rows:
table.Rows.Add(row['UserID'],
row['VehicleIndex'],
False,
7)
return dataset
def add_event_record(
self,
#CK: check!
#event_type, event_subtype=EventViewEnums.Net2EventSubTypes.None,
event_type, event_subtype=None,
address=0, subaddress=0, user_id=None, card_nr=0,
event_detail=None,
linked_event_id=0, ioboard_id=0, input_id=0, output_id=0):
"""Add an event record
Returns True on success.
"""
with Net2XS._lock:
self._check_client()
return self._client.AddEventRecord(
event_type,
event_subtype,
address, subaddress, user_id, card_nr, event_detail,
linked_event_id, ioboard_id, input_id, output_id)
def add_acu(self, address, acu_type):
"""Add new acu device
Acu_types: Classic = 1, Nano = 2, Plus = 3, EasyProxNano = 4, PaxLoc = 7, PaxLocMifare = 8
Returns True on success. (up till 4.27, there is a bug in the SDK causing a False on success)
"""
with Net2XS._lock:
self._check_client()
return self._client.AddACU(address, acu_type)
def delete_acu(self, address):
"""Delete acu device
Return True on success.
"""
with Net2XS._lock:
self._check_client()
return self._client.DeleteACU(address)
@property
def last_error_message(self):
"""Last error message
"""
with Net2XS._lock:
self._check_client()
return self._client.LastErrorMessage
@property
def no_expiration_date(self):
"""The 'special' no expiration date of Net2 (1753-01-01)
Returns dotnet DateTime object
"""
return date_time_to_net(datetime(year=1753, month=1, day=1))
@property
def now_date(self):
"""The now datetime
Returns dotnet DateTime object
"""
return date_time_to_net(datetime.now())
def dispose(self):
"""Dispose Net2 client
"""
with Net2XS._lock:
if self._client:
try:
self._client.Dispose()
self._client = None
Net2XS._logger.Debug('Disposed')
except:
pass
def _reconnected(self, sender, event_args):
"""Reconnect handler
"""
if event_args.ReAuthenticationRequired:
try:
self.authenticate(self._user_id, self._password)
self._connected = True
Net2XS._logger.Debug('Reconnected')
except Exception as e:
Net2XS._logger.Debug('Reconnect error: %s' % str(e))
else:
self._connected = True
def _disconnected(self, sender, event_args):
"""Disconnect handler
"""
Net2XS._logger.Debug('Disconnected')
self._connected = False
@property
def on_acu_event(self):
return self._on_acu_event
@on_acu_event.setter
def on_acu_event(self, value):
self._on_acu_event = value
def _acu_event(self, sender, event_view):
"""Acu event handler
"""
if self.on_acu_event:
try:
self.on_acu_event(sender, event_view)
except Exception as e:
Net2XS._logger.Debug('Acu even handler error: %s' % str(e))
def monitor_acu(self, address):
"""Monitor events of given acu
Returns True on success
"""
with Net2XS._lock:
self._check_client()
return self._client.MonitorAcu(address)
def stop_monitoring_acu(self, address):
"""Stop monitoring events of given acu
Returns True on success
"""
with Net2XS._lock:
self._check_client()
return self._client.StopMonitoringAcu(address)
def activate_ioboard_relays(self, ioboard_id, settings):
"""Control ioboard relays
Returns True on success
"""
with Net2XS._lock:
self._check_client()
return self._client.ActivateIoBoardRelays(
ioboard_id, settings)
@classmethod
def door_status_str(cls, status):
"""Translate door status to human readable string
Returns a csv string with the combined status bits.
"""
ret = []
if status & OC.DoorStatusFlags.PSUIsOK:
ret.append('psu ok')
if status & OC.DoorStatusFlags.IntruderAlarm:
ret.append('intruder')
if not status & OC.DoorStatusFlags.TamperStatusGood:
ret.append('tampered')
if status & OC.DoorStatusFlags.DoorContactClosed:
ret.append('door contact closed')
if status & OC.DoorStatusFlags.AlarmTripped:
ret.append('alarm')
if status & OC.DoorStatusFlags.DoorOpen:
ret.append('door open')
# When psu not ok, other state bits are irrelevant
else:
ret.append('psu not ok')
return ', '.join(ret)

151
src/net2xs/conversions.py
Unescape Escape View File

@ -0,0 +1,151 @@
"""
Module holding conversion definitions
"""
from datetime import datetime
from System import DateTime
from . deftypes import TimeZone, TimeSlot, AccessLevel
def date_time_to_py(val):
"""Convert dotnet date time format to python presentation
Returns python datetime object.
"""
return datetime(
year=val.Year, month=val.Month, day=val.Day,
hour=val.Hour, minute=val.Minute, second=val.Second)
def date_time_to_net(val):
"""Convert python to dotnet date time presentation
Returns dotnet DateTime object.
"""
return DateTime(
val.year, val.month, val.day,
val.hour, val.minute, val.second)
def flex_date_time_to_net(val):
"""Convert python to dotnet date time presentation when required
Returns dotnet DateTime object or None.
"""
if val is None:
return val
if type(val) is datetime:
return date_time_to_net(val)
if not type(val) is DateTime:
raise TypeError(
("The time value supplied (%s), " +
"can not be interpreted as a dotnet DateTime object") %
(str(val)))
return val
def time_zones_to_py(dataset):
"""Convert Net2 time zones dataset to a python presentation
Returns array of TimeZone objects.
"""
if not dataset or dataset.Tables.Count != 2:
raise Exception('Timezone dataset is invalid')
result = []
for tz_row in dataset.Tables[0].Rows:
time_zone = TimeZone(
id=tz_row.TimezoneID,
name=tz_row.Name)
ts_rows = dataset.Tables[1].Select('TimezoneID=%d' % time_zone.id)
for ts_row in ts_rows:
time_slot = TimeSlot(
id=ts_row.SlotID,
day=ts_row.Day,
start_time=date_time_to_py(ts_row.StartTime),
end_time=date_time_to_py(ts_row.EndTime))
time_zone.slots.append(time_slot)
result.append(time_zone)
return result
def access_levels_to_py(dataset):
"""Convert Net2 access_level dataset to a python presentation
Returns array of AccessLevel objects.
"""
if not dataset or dataset.Tables.Count != 1:
raise Exception('AccessLevel dataset is invalid')
result = []
for al_row in dataset.Tables[0].Rows:
access_level = AccessLevel(
id=al_row.get_Item("AccessLevelID"),
name=al_row.get_Item("AccessLevelName"))
result.append(access_level)
return result
def access_level_detail_to_py(access_level, dataset):
"""Convert Net2 access_level detail dataset
time_zones is a net2-id key based lookup dict.
"""
if not dataset or dataset.Tables.Count != 1:
raise Exception('AccessLevelDetail dataset is invalid')
# Clear existing
access_level.details = []
# Add details
for ald_row in dataset.Tables[0].Rows:
time_zone_id = ald_row.get_Item("TimezoneID")
area_id = ald_row.get_Item("AreaID")
access_level.details.append((time_zone_id, area_id))
def user_view_to_py(user_view):
"""Convert a Net2 UserView object (partially) to a dict
"""
res = {}
res["user_id"] = user_view.UserId
res["access_level_id"] = user_view.AccessLevelId
res["department_id"] = user_view.DepartmentId
res["anti_passback_user"] = user_view.AntiPassbackUser
res["alarm_user"] = user_view.AlarmUser
res["first_name"] = user_view.FirstName
res["middle_name"] = user_view.MiddleName
res["sur_name"] = user_view.Surname
res["telephone"] = user_view.Telephone
res["extension"] = user_view.Extension
res["pin"] = user_view.PIN
res["activation_date"] = date_time_to_py(user_view.ActivationDate)
res["active"] = user_view.Active
res["fax"] = user_view.Fax
res["expiry_date"] = date_time_to_py(user_view.ExpiryDate)
res["custom_fields"] = [
None,
user_view.Field1_100,
user_view.Field2_100,
user_view.Field3_50,
user_view.Field4_50,
user_view.Field5_50,
user_view.Field6_50,
user_view.Field7_50,
user_view.Field8_50,
user_view.Field9_50,
user_view.Field10_50,
user_view.Field11_50,
user_view.Field12_50,
user_view.Field13_Memo,
user_view.Field14_50]
res["user_guid"] = user_view.UserGuid.ToString()
res["last_area_id"] = user_view.LastAreaId
res["last_access_time"] = date_time_to_py(user_view.LastAccessTime)
res["last_updated"] = date_time_to_py(user_view.LastUpdated)
return res

120
src/net2xs/deftypes.py
Unescape Escape View File

@ -0,0 +1,120 @@
"""
Module holding definition data types
"""
class TimeSlot(object):
"""Time zone slot definition
Member id: Timeslot id
Member day: Timeslot day number
Member start_time: Start time of slot
Member start_time: End time of slot
"""
def __init__(self, id=-1, day=-1, start_time=None, end_time=None):
self.id = id
self.start_time = start_time
self.end_time = end_time
self.day = day
def __eq__(self, other):
"""Compare
"""
return (
self.start_time == other.start_time and
self.end_time == other.end_time and
self.day == other.day)
def __ne__(self, other):
"""Compare negative
"""
return not self.__eq__(other)
def __str__(self):
return ('%s id=%d, day=%d, start_time=%s, end_time=%s' %
(self.__class__.__name__,
self.id, self.day, self.start_time, self.end_time))
class TimeZone(object):
"""Time zone definition
Member id: Timezone id
Member name: Timezone name
Member slots: Array of TimeSlot objects
"""
def __init__(self, id=-1, name=None):
self.id = id
self.name = name
self.slots = []
def __eq__(self, other):
"""Compare
"""
if (type(self) != type(other) or
len(self.slots) != len(other.slots)):
return False
for s_slot in self.slots:
for o_slot in other.slots:
if s_slot == o_slot:
break
else:
return False
return True
def __ne__(self, other):
"""Compare negative
"""
return not self.__eq__(other)
def __str__(self):
out = []
out.append('%s id=%d, name=%s' %
(self.__class__.__name__, self.id, self.name))
for slot in self.slots:
out.append('-%s' % slot)
return '\n'.join(out)
class AccessLevel(object):
"""Access level definition
Member id: Accesslevel id
Member name: Accesslevel name
Member details: Array of (timezone id, area id) tuples
"""
def __init__(self, id, name):
self.id = id
self.name = name
self.details = []
def __eq__(self, other):
"""Compare
"""
if (type(self) != type(other) or
len(self.details) != len(other.details)):
return False
for s_detail in self.details:
for o_detail in other.details:
if s_detail == o_detail:
break
else:
return False
return True
def __ne__(self, other):
"""Compare negative
"""
return not self.__eq__(other)
def __str__(self):
out = []
out.append('%s id=%d, name=%s' % (
self.__class__.__name__, self.id, self.name))
for tz_id, area_id in self.details:
out.append('-tz=%d, area=%d' % (tz_id, area_id))
return '\n'.join(out)

8
src/network/__init__.py
Unescape Escape View File

@ -0,0 +1,8 @@
"""
The module that offers several Net2 related network utilities
"""
# This is just to force py2exe to include these packages
if False == True:
import network.net2plus
import network.sqlserver

66
src/network/net2plus.py
Unescape Escape View File

@ -0,0 +1,66 @@
"""
Net2Plus network utilities
"""
import socket
class Net2PlusFinder(object):
"""Class to help find Net2Plus modules
"""
def __init__(self, port=30718, timeout=4, ip="0.0.0.0"):
self._port = port
self._timeout = timeout
self._ip = ip
def _broadcast(self):
"""Broadcast on udp
"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.bind((self._ip, self._port))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
# Send twice
s.sendto(bytearray([0x00, 0x00, 0x00, 0xf6]), ("255.255.255.255", self._port))
s.sendto(bytearray([0x00, 0x00, 0x00, 0xf6]), ("255.255.255.255", self._port))
finally:
s.close()
def find(self):
"""Find Net2Plus modules responsing to a Broadcast
Returns an array of dicts with info about each module
"""
res = []
# Broadcast
self._broadcast()
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.settimeout(self._timeout)
s.bind((self._ip, self._port))
# Wait for result
try:
while True:
data, addr = s.recvfrom(1024)
if len(data) >= 30:
mac_nrs = tuple(ord(i) for i in data[24:30])
mac = "%02x:%02x:%02x:%02x:%02x:%02x" % mac_nrs
mod_nr = (mac_nrs[3] << 16) + (mac_nrs[4] << 8) + mac_nrs[5]
version = "%d.%02d (%d.%02d)" % (ord(data[19]), ord(data[18]), ord(data[21]), ord(data[20]))
dev_dict = dict(
ip= addr[0],
mac=mac,
module=mod_nr,
version=version)
res.append(dev_dict)
except socket.timeout:
pass
return res
finally:
s.close()

54
src/network/sqlserver.py
Unescape Escape View File

@ -0,0 +1,54 @@
"""
SQLServer network utilities
"""
import socket
class SqlServerFinder(object):
"""Class to help find SQLServer instances
"""
def __init__(self, port=1434, timeout=4, ip="0.0.0.0"):
self._port = port
self._timeout = timeout
self._ip = ip
def find(self):
"""Find SQLServer instances responsing to a Broadcast
Returns an array of dicts with info about each instance
"""
res = []
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.settimeout(self._timeout)
s.bind((self._ip, 0))
self._local_port = s.getsockname()[1]
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
s.sendto(bytearray([0x02]), ("255.255.255.255", self._port))
# Wait for result
try:
while True:
data, addr = s.recvfrom(1024)
if len(data) > 3:
data = data.decode()
entry = {}
parts = data[3:].split(";")
nr_values = int(len(parts) / 2)
for i in range(nr_values):
key = parts[i * 2].strip()
value = parts[1 + i * 2].strip()
if (key):
entry[key] = value
res.append(entry)
except socket.timeout:
pass
return res
finally:
s.close()

64
src/pylog4net/__init__.py
Unescape Escape View File

@ -0,0 +1,64 @@
"""
Module to use log4net from python
"""
import os
import clr
import sys
import settings
# Add path lib to search path
LOG4NET_LIB_DIR = os.path.join(settings.LIB_DIR, 'log4netdll')
if LOG4NET_LIB_DIR not in sys.path:
sys.path.append(LOG4NET_LIB_DIR)
clr.AddReference('log4net')
import log4net
from log4net.Config import XmlConfigurator
clr.AddReference('System.Xml')
from System.Xml import XmlDocument
class Log4NetError(Exception):
"""Log4Net exception class
"""
pass
class Log4Net(object):
"""Class that offers log4net usage, taking its config from an xml file
"""
@classmethod
def read_config(cls, config_file):
"""Read and process config file
"""
if not os.path.isfile(config_file):
raise Log4NetError('Failed to find config file "%s"' % config_file)
# Read document
doc = XmlDocument()
try:
doc.Load(config_file)
except Exception as e:
raise Log4NetError(str(e))
# Obtain 1st element with log4net tag
for element in doc.GetElementsByTagName('log4net'):
XmlConfigurator.Configure(element)
break
@classmethod
def get_logger(cls, name):
"""Obtain logger instance
"""
return log4net.LogManager.GetLogger(name)
@classmethod
def get_file_appender_logfile(cls):
"""Obtain file appender log file
"""
for appender in log4net.LogManager.GetRepository().GetAppenders():
if isinstance(appender, log4net.Appender.FileAppender):
return str(appender.File)
return None

51
src/samples/add_user.py
Unescape Escape View File

@ -0,0 +1,51 @@
"""
Sample script to add a new user
"""
from net2xs import Net2XS
# Uncomment to use dotnet DateTime objects
# from System import DateTime
# Uncomment to use python datetime objects
# from datetime import datetime
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
if __name__ == "__main__":
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Add new user
res = net2.add_user(
access_level_id=1, # Always, all doors
department_id=0, # No department
anti_passback_ind=False,
alarm_user_ind=False,
first_name="John",
middle_name=None,
sur_name="Doe",
telephone_no="12345678",
telephone_extension=None,
pin_code=None,
activation_date=None, # Now
# Or supply a dotnet DateTime object (also see import)
# activation_date=DateTime(2018, 1, 2),
# Or supply a python datetime object (also see import)
# activation_date=datetime(2018, 1, 2),
active=True,
fax_no=None,
expiry_date=None) # Never expire (also see activation_date)
if res:
print("Success")
else:
print("Failure")

46
src/samples/add_users_with_card.py
Unescape Escape View File

@ -0,0 +1,46 @@
"""
Sample script to add many users with cards
"""
from net2xs import Net2XS
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
if __name__ == "__main__":
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Add 1000 users
for i in range(1, 1001):
first_name = "Test"
sur_name = "User #%04d" % (i)
card_nr = 77770000 + i
# Add user
res = net2.add_user(
first_name=first_name,
sur_name=sur_name)
if not res:
print("Failed to add user %s %s: %s" %
(first_name, sur_name, net2.last_error_message))
continue
# Get user id
user_id = net2.get_user_id_by_name((first_name, sur_name))
if user_id < 0:
print("Failed to find user %s %s" % (first_name, sur_name))
continue
# Create card
res = net2.add_card(card_nr, 1, user_id)
if not res:
print("Failed to add card to user %s %s: %s" %
(first_name, sur_name, net2.last_error_message))

35
src/samples/direct_db_query.py
Unescape Escape View File

@ -0,0 +1,35 @@
"""
Sample script to query the database directly
"""
from net2dbxs import Net2DBXS
if __name__ == "__main__":
with Net2DBXS() as net2db:
# Connect
net2db.connect()
# Get last log entry relating a device
dataset = net2db.query_db(
"select * from EventsEx "
"where EventID = "
"(select max(EventID) from sdk.EventsEx "
" where SerialNumber is not NULL)")
# Non result
if (not dataset
or dataset.Tables.Count < 1
or dataset.Tables[0].Rows.Count < 1):
print("Nothing relevant found")
else:
# Just to demonstrate how to get values from a dataset
# Typically Table[0]
table = dataset.Tables[0]
# In this case only interested in the first row
row = dataset.Tables[0].Rows[0]
# For each column
for col in table.Columns:
val = row.get_Item(col.ColumnName)
print("%s = %s" % (col.ColumnName, val))

17
src/samples/find_modules.py
Unescape Escape View File

@ -0,0 +1,17 @@
"""
Sample script to detect Net2Plus modules
"""
from network.net2plus import Net2PlusFinder
if __name__ == "__main__":
inst = Net2PlusFinder()
print("Please wait for the modules to respond...")
devs = inst.find()
if devs:
for d in devs:
print(d)
else:
print("No devices detected")

17
src/samples/find_sqlserver.py
Unescape Escape View File

@ -0,0 +1,17 @@
"""
Sample script to detect all SQLServer instances
"""
from network.sqlserver import SqlServerFinder
if __name__ == "__main__":
inst = SqlServerFinder()
print("Please wait for the servers to respond...")
srvs = inst.find()
if srvs:
for s in srvs:
print(s)
else:
print("No servers detected")

22
src/samples/get_access_levels.py
Unescape Escape View File

@ -0,0 +1,22 @@
"""
Sample script to fetch all access levels
"""
from net2xs import Net2XS
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
if __name__ == "__main__":
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Show all access levels and their details
for al in net2.get_py_access_levels():
print(al)

21
src/samples/get_cards.py
Unescape Escape View File

@ -0,0 +1,21 @@
"""
Sample script to fetch all cards
"""
from net2xs import Net2XS
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
if __name__ == "__main__":
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Obtain all Net2 cards
print(Net2XS.dataset_to_str(net2.get_cards()))

22
src/samples/get_time_zones.py
Unescape Escape View File

@ -0,0 +1,22 @@
"""
Sample script to fetch all time zones
"""
from net2xs import Net2XS
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
if __name__ == "__main__":
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Show all time zones and their details
for tz in net2.get_py_time_zones():
print(tz)

47
src/samples/inheritance.py
Unescape Escape View File

@ -0,0 +1,47 @@
"""
Sample script to demonstrate inheritance
"""
from net2xs import Net2XS
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
class MyNet2XS(Net2XS):
"""Inherited class for additional functionality
"""
global Net2XS
def get_current_user_id(self):
"""Return logged on user id
"""
# Place a lock, for thread safety
# (if you don't have threads you kan skip this)
with Net2XS._lock:
# Basic check if client connection is valid
self._check_client()
return self._client.CurrentUserID
def get_client_members(self):
"""Return all Net2 client object members using the introspective
Python dir function
"""
with Net2XS._lock:
self._check_client()
return dir(self._client)
if __name__ == "__main__":
with MyNet2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Show current user id
print("Current used id:", net2.get_current_user_id())
# Show all members
print("Net2 client members:", net2.get_client_members())

48
src/samples/modify_user.py
Unescape Escape View File

@ -0,0 +1,48 @@
"""
Sample script to modify a user.
It assumes that a John Doe user exists (see add_user sample).
"""
from net2xs import Net2XS
from datetime import datetime
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
# First name
FIRST_NAME = "John"
# Surname
SUR_NAME = "Doe"
if __name__ == "__main__":
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Get user id
user_id = net2.get_user_id_by_name((FIRST_NAME, SUR_NAME))
print("Found user id %d" % (user_id))
# Found a valid user id
if user_id >= 0:
# Modify expiration date
res = net2.modify_user(
user_id=user_id,
expiry_date=datetime(2022, 12, 31))
if res:
print("Success")
else:
print("Failure")
else:
print("Failed to find user %s %s" % (FIRST_NAME, SUR_NAME))

48
src/samples/monitor_acu.py
Unescape Escape View File

@ -0,0 +1,48 @@
"""
Sample script to monitor acu
"""
from time import sleep
from net2xs import Net2XS
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
# Acu to monitor
ACU_ADDRESS = 1219195
def handle_acu_event(sender, event_view):
"""Handler for ACU event
"""
print("-----------------------")
print("address=", event_view.Address)
print("card=", event_view.CardNumber)
print("department=", event_view.Department)
print("userid=", event_view.UserId)
if __name__ == "__main__":
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Set function to handle ACU event
net2.on_acu_event = handle_acu_event
res = net2.monitor_acu(ACU_ADDRESS)
if not res:
print("Failed to monitor ACU %d" % (ACU_ADDRESS))
else:
print("Monitoring ACU %d; press <ctrl>C to quit" % (ACU_ADDRESS))
while True:
try:
sleep(1)
except KeyboardInterrupt:
break
net2.stop_monitoring_acu(ACU_ADDRESS)

60
src/samples/open_all_doors.py
Unescape Escape View File

@ -0,0 +1,60 @@
"""
Sample script to hold all known doors open for a few seconds.
Also shows how to use the logger function.
"""
import time
from net2xs import Net2XS
from pylog4net import Log4Net
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
def get_doors(net2):
"""Obtain a list of all known doors
"""
res = []
dataset = net2.get_doors()
if dataset and dataset.Tables.Count > 0:
for row in dataset.Tables[0].Rows:
res.append(row.Address)
return res
if __name__ == "__main__":
# Create logger object
logger = Log4Net.get_logger('open_all_doors')
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Get list off door addresses
doors = get_doors(net2)
# Open each door
for door in doors:
if not net2.hold_door_open(door):
logger.Error(
"Failed to hold door %d open: %s." %
(door, net2.last_error_message))
else:
logger.Info("Set door %d open." % (door))
logger.Info("Now all doors are open...")
time.sleep(3)
# Close each door
for door in doors:
if not net2.close_door(door):
logger.Error(
"Failed to close door %d: %s." %
(door, net2.last_error_message))
else:
logger.Info("Set door %d closed." % (door))
logger.Info("Now all doors are closed again")

21
src/samples/user_script.py
Unescape Escape View File

@ -0,0 +1,21 @@
"""
Sample script to fetch a dataset with all active users
"""
from net2xs import Net2XS
# Operator id 0 is System Engineer
OPERATOR_ID = 0
# Default Net2 password
OPERATOR_PWD = "net2"
# When running on the machine where Net2 is installed
NET2_SERVER = "localhost"
if __name__ == "__main__":
with Net2XS(NET2_SERVER) as net2:
# Authenticate
net2.authenticate(OPERATOR_ID, OPERATOR_PWD)
# Obtain all Net2 users
print(Net2XS.dataset_to_str(net2.get_users()))

21
src/settings.py
Unescape Escape View File

@ -0,0 +1,21 @@
import sys
import os
# Running as py2exe
if hasattr(sys, 'frozen'):
DEBUG = False
BASE_DIR = os.path.split(sys.executable)[0]
# Running as interpreter
else:
DEBUG = True
BASE_DIR = '.'
# Config file to use for dotnet stuff
CONFIG_FILE = os.path.join(
BASE_DIR, 'Net2Scripting.exe.config')
# Generic lib dir
LIB_DIR = os.path.join(BASE_DIR, 'libs')
# Version
VERSION = '4.0'

35
src/setup.py
Unescape Escape View File

@ -0,0 +1,35 @@
import os
from distutils.core import setup
from glob import glob
import py2exe
# Data files (resources)
data_files = [('.', ['Net2Scripting.exe.config']),
('.', glob(r'../runtime/*.*')),
('libs/log4netdll', glob(r'libs/log4netdll/*.*')),
('libs/paxton', glob(r'libs/paxton/*.*')),
('resources', glob(r'../resources/*.*')),
('docs', glob(r'../docs/*.*')),
('samples', glob(r'samples/*.py'))]
# Source files (python modules)
source_files = [os.path.splitext(os.path.basename(f))[0] for f in glob('*.py')]
setup(
# basic console exe
console=[{'script': 'Net2Scripting.py',
'icon_resources': [(1, '../resources/Net2Scripting.ico')]}],
py_modules=source_files,
data_files=data_files,
# zipfile=None,
# options={'py2exe': {'bundle_files': 1, 'compressed': True}}
options={
'py2exe': {
'dist_dir': '../dist',
},
'build': {
'build_base': '../build'
}
}
)
Write Preview
Loading…
Cancel
Save