randogoth/quanttp2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement libmeterfeeder over libqwqng for multiple device support

Browse source 
WebSockets API isn't updated yet
This commit is contained in:
Simon Nishi McCorkindale 2021-06-04 00:28:38 +08:00
parent 3fe4b5be81
commit 03e4016b4b
11 changed files with 225 additions and 276 deletions
Download patch file Download diff file Expand all files Collapse all files

121
quanttp/data/meterfeeder_wrapper.py Normal file
View file

@ -0,0 +1,121 @@
##
# MeterFeeder Library Wrapper
#
# by soliax
##
from platform import os
from sys import platform
from ctypes import *
cdll = LibraryLoader(CDLL)
class MeterFeederWrapper:
def __init__(self):
if platform == "linux" or platform == "linux2":
# Linux
self._meterfeeder = cdll.LoadLibrary(os.getcwd() + '/libmeterfeeder.so')
elif platform == "darwin":
# OS X
self._meterfeeder = cdll.LoadLibrary(os.getcwd() + '/libmeterfeeder.dylib')
elif platform == "win32":
# Windows
self._meterfeeder = cdll.LoadLibrary(os.getcwd() + '/meterfeeder.dll')
# Declare function bridging
self._meterfeeder.MF_Initialize.argtypes = c_char_p,
self._meterfeeder.MF_Initialize.restype = c_int
self._meterfeeder.MF_GetNumberGenerators.restype = c_int
self._meterfeeder.MF_GetListGenerators.argtypes = [POINTER(c_char_p)]
self._meterfeeder.MF_GetBytes.argtypes = c_int, POINTER(c_ubyte), c_char_p, c_char_p,
self._meterfeeder.MF_RandInt32.argtypes = c_char_p, c_char_p,
self._meterfeeder.MF_RandInt32.restype = c_int
self._meterfeeder.MF_RandUniform.argtypes = c_char_p, c_char_p,
self._meterfeeder.MF_RandUniform.restype = c_double
self._meterfeeder.MF_RandNormal.argtypes = c_char_p, c_char_p,
self._meterfeeder.MF_RandNormal.restype = c_double
self._meterfeeder.MF_Clear.argtypes = c_char_p, c_char_p,
self._meterfeeder.MF_Clear.restype = c_bool
self._meterfeeder.MF_Reset.argtypes = c_char_p,
self._meterfeeder.MF_Reset.restype = c_int
self._medErrorReason = create_string_buffer(256)
# Initialize Meter Feeder
result = self._meterfeeder.MF_Initialize(self._medErrorReason)
if (len(self._medErrorReason.value) > 0):
print(self._medErrorReason)
exit(result)
def deviceIds(self, returnAsList):
# Get the list of connected devices
numGenerators = self._meterfeeder.MF_GetNumberGenerators()
generatorsListBuffers = [create_string_buffer(58) for i in range(numGenerators)]
generatorsListBufferPointers = (c_char_p*numGenerators)(*map(addressof, generatorsListBuffers))
self._meterfeeder.MF_GetListGenerators(generatorsListBufferPointers)
generatorsList = [str(s.value, 'utf-8') for s in generatorsListBuffers]
if returnAsList:
return generatorsList
deviceIdsCsvList = ""
for i in range(numGenerators):
if i < numGenerators - 1:
deviceIdsCsvList += generatorsList[i].split("|")[0] + ","
else:
deviceIdsCsvList += generatorsList[i].split("|")[0]
return deviceIdsCsvList
def randint32(self, deviceId):
try:
return self._meterfeeder.MF_RandInt32(deviceId.encode("utf-8"), self._medErrorReason)
except:
self._meterfeeder.MF_Reset()
return self._meterfeeder.MF_RandInt32(deviceId.encode("utf-8"), self._medErrorReason)
def randuniform(self, deviceId):
try:
return self._meterfeeder.MF_RandUniform(deviceId.encode("utf-8"), self._medErrorReason)
except:
self._meterfeeder.MF_Reset()
return self._meterfeeder.MF_RandUniform(deviceId.encode("utf-8"), self._medErrorReason)
def randnormal(self, deviceId):
try:
return self._meterfeeder.MF_RandNormal(deviceId.encode("utf-8"), self._medErrorReason)
except:
self._meterfeeder.MF_Reset()
return self._meterfeeder.MF_RandNormal(deviceId.encode("utf-8"), self._medErrorReason)
def randbytes(self, deviceId, length):
barray = bytearray(length)
ubuffer = (c_ubyte * length).from_buffer(barray)
try:
self._meterfeeder.MF_GetBytes(length, ubuffer, deviceId.encode("utf-8"), self._medErrorReason)
return barray
except:
self._meterfeeder.MF_Reset()
self._meterfeeder.MF_GetBytes(length, ubuffer, deviceId.encode("utf-8"), self._medErrorReason)
return barray
def clear(self, deviceId):
if self._meterfeeder.MF_Clear(deviceId.encode("utf-8"), self._medErrorReason) == False:
print("unable to clear", deviceId, ": ", self._medErrorReason.value)
def reset(self):
result = self._meterfeeder.MF_Reset(self._medErrorReason)
if (len(self._medErrorReason.value) > 0):
print(self._medErrorReason)
exit(result)
def status(self):
return "ONLINE"

99
quanttp/data/qng_wrapper_linux.py
View file

@ -1,99 +0,0 @@
# Copyright (c) 2020 Andika Wasisto
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import platform
from ctypes import cdll, c_void_p, c_int, c_double, POINTER, c_char, c_char_p, cast
class QngWrapperLinux:
def __init__(self):
self._qwqng_wrapper = cdll.LoadLibrary('./libqwqng-wrapper-x86-64.so' if platform.machine().endswith('64') else './libqwqng-wrapper.so')
self._qwqng_wrapper.GetQwqngInstance.restype = c_void_p
self._qwqng_wrapper.RandInt32.argtypes = [c_void_p]
self._qwqng_wrapper.RandInt32.restype = c_int
self._qwqng_wrapper.RandUniform.argtypes = [c_void_p]
self._qwqng_wrapper.RandUniform.restype = c_double
self._qwqng_wrapper.RandNormal.argtypes = [c_void_p]
self._qwqng_wrapper.RandNormal.restype = c_double
self._qwqng_wrapper.RandBytes.argtypes = [c_void_p, c_int]
self._qwqng_wrapper.RandBytes.restype = POINTER(c_char)
self._qwqng_wrapper.StatusString.argtypes = [c_void_p]
self._qwqng_wrapper.StatusString.restype = POINTER(c_char)
self._qwqng_wrapper.DeviceID.argtypes = [c_void_p]
self._qwqng_wrapper.DeviceID.restype = POINTER(c_char)
self._qwqng_wrapper.Clear.argtypes = [c_void_p]
self._qwqng_wrapper.Reset.argtypes = [c_void_p]
self._qng_pointer = self._qwqng_wrapper.GetQwqngInstance()
def deviceId(self):
try:
return cast(self._qwqng_wrapper.DeviceID(self._qng_pointer), c_char_p).value.decode("utf-8")
except:
self._qwqng_wrapper.Reset(self._qng_pointer)
return cast(self._qwqng_wrapper.DeviceID(self._qng_pointer), c_char_p).value.decode("utf-8")
def statusString(self):
try:
return cast(self._qwqng_wrapper.StatusString(self._qng_pointer), c_char_p).value.decode("utf-8")
except:
self._qwqng_wrapper.Reset(self._qng_pointer)
return cast(self._qwqng_wrapper.StatusString(self._qng_pointer), c_char_p).value.decode("utf-8")
def randint32(self):
try:
return self._qwqng_wrapper.RandInt32(self._qng_pointer)
except:
self._qwqng_wrapper.Reset(self._qng_pointer)
return self._qwqng_wrapper.RandInt32(self._qng_pointer)
def randuniform(self):
try:
return self._qwqng_wrapper.RandUniform(self._qng_pointer)
except:
self._qwqng_wrapper.Reset(self._qng_pointer)
return self._qwqng_wrapper.RandUniform(self._qng_pointer)
def randnormal(self):
try:
return self._qwqng_wrapper.RandNormal(self._qng_pointer)
except:
self._qwqng_wrapper.Reset(self._qng_pointer)
return self._qwqng_wrapper.RandNormal(self._qng_pointer)
def randbytes(self, length):
try:
return self._randbytes_arbitrary_length(length)
except:
self._qwqng_wrapper.Reset(self._qng_pointer)
return self._randbytes_arbitrary_length(length)
def _randbytes_arbitrary_length(self, length):
try:
return self._qwqng_wrapper.RandBytes(self._qng_pointer, length)[:length]
except:
self._qwqng_wrapper.Reset(self._qng_pointer)
def clear(self):
self._qwqng_wrapper.Clear(self._qng_pointer)
def reset(self):
self._qwqng_wrapper.Reset(self._qng_pointer)

40
quanttp/data/qng_wrapper_mock.py
View file

@ -1,40 +0,0 @@
# Copyright (c) 2020 Andika Wasisto
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import random
import secrets
class QngWrapperMock:
def randint32(self):
return random.randint(-2147483648, +2147483647)
def randuniform(self):
return random.random()
def randnormal(self):
return random.gauss(mu=0.0, sigma=1.0)
def randbytes(self, length):
return secrets.token_bytes(nbytes=length)
def clear(self):
pass

76
quanttp/data/qng_wrapper_windows.py
View file

@ -1,76 +0,0 @@
# Copyright (c) 2020 Andika Wasisto
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
class QngWrapperWindows:
def __init__(self):
import win32com.client
self._qng = win32com.client.Dispatch("QWQNG.QNG")
def deviceId(self):
try:
return self._qng.DeviceId
except:
self._qng.Reset()
return self._qng.DeviceId
def randint32(self):
try:
return self._qng.RandInt32
except:
self._qng.Reset()
return self._qng.RandInt32
def randuniform(self):
try:
return self._qng.RandUniform
except:
self._qng.Reset()
return self._qng.RandUniform
def randnormal(self):
try:
return self._qng.RandNormal
except:
self._qng.Reset()
return self._qng.RandNormal
def randbytes(self, length):
try:
return self._randbytes_arbitrary_length(length)
except:
self._qng.Reset()
return self._randbytes_arbitrary_length(length)
def _randbytes_arbitrary_length(self, length):
if length <= 8192:
return bytes(self._qng.RandBytes(length))
else:
data = bytearray()
for x in range(length // 8192):
data.extend(bytearray(self._qng.RandBytes(8192)))
bytes_needed = length % 8192
if bytes_needed != 0:
data.extend(bytearray(self._qng.RandBytes(bytes_needed)))
return bytes(data)
def clear(self):
self._qng.Clear()