214 lines
7.5 KiB
Python
214 lines
7.5 KiB
Python
from ..core.SyslabUnit import SyslabUnit
|
|
from ..core.datatypes import CompositeMeasurement
|
|
from typing import Union
|
|
from time import time
|
|
|
|
def cast_to_cm(m: Union[CompositeMeasurement, float]):
|
|
if type(m) == float:
|
|
# TODO: Is there a better way to estimate precision of time.time?
|
|
request = CompositeMeasurement(m, timestampMicros=time()*1e6, timePrecision=1000)
|
|
elif type(m) == CompositeMeasurement:
|
|
request = m
|
|
else:
|
|
raise TypeError(f"Unknown request type: {type(m)}")
|
|
return request
|
|
|
|
class EVSE(SyslabUnit):
|
|
"""The EVSE class represents a charging post in SYSLAB.
|
|
The EVSE class is instantiated using a string with the unique name of the charging post, ie. 'which'
|
|
|
|
A full list of available charging posts can be found by calling 'EVSE.getAvailableChargingPosts()'
|
|
|
|
Alternatively, the user may specify a host and port to connect to via the *host* and *port* arguments.
|
|
"""
|
|
|
|
__CHARGINGPOSTS = {
|
|
'V2G-319': ('syslab-35.syslab.dk', '8080', 'Endesa_V2G'),
|
|
'EVSE-NEVIC-6': ('10.42.245.96', '8080', 'EVSE_NEVIC_6'),
|
|
}
|
|
|
|
def __init__(self, which=None, host=None, port=None, unitname=None):
|
|
|
|
baseurl = 'http://{host}:{port}/typebased_WebService_EVSE/EVSEWebService/{unit_name}/'
|
|
super().__init__(
|
|
baseurl=baseurl,
|
|
which=which,
|
|
units=self.__CHARGINGPOSTS,
|
|
host=host,
|
|
port=port,
|
|
unit_name=unitname,
|
|
unit_type="EVSE")
|
|
|
|
def isEVPresent(self):
|
|
return self._request_resource('isEVPresent')
|
|
|
|
def canRemoteStartStop(self):
|
|
return self._request_resource('canRemoteStartStop')
|
|
|
|
def canLimitChargePower(self):
|
|
return self._request_resource('canLimitChargePower')
|
|
|
|
def canDischarge(self):
|
|
return self._request_resource('canDischarge')
|
|
|
|
def hasSOC(self):
|
|
return self._request_resource('hasSOC')
|
|
|
|
def hasACMeasurements(self):
|
|
return self._request_resource('hasACMeasurements')
|
|
|
|
def hasDCMeasurements(self):
|
|
return self._request_resource('hasDCMeasurements')
|
|
|
|
def getAvailableOperatingStates(self):
|
|
return self._request_resource('getAvailableOperatingStates')
|
|
|
|
def getCurrentOperatingState(self):
|
|
return self._request_resource('getCurrentOperatingState')
|
|
|
|
def getACActivePower(self):
|
|
return self._request_resource('getACActivePower')
|
|
|
|
def getACReactivePower(self):
|
|
return self._request_resource('getACReactivePower')
|
|
|
|
def getACFrequency(self):
|
|
return self._request_resource('getACFrequency')
|
|
|
|
def getACVoltages(self):
|
|
return self._request_resource('getACVoltages')
|
|
|
|
def getACCurrents(self):
|
|
return self._request_resource('getACCurrents')
|
|
|
|
def getSOC(self):
|
|
return self._request_resource('getSOC')
|
|
|
|
def getMinSOC(self):
|
|
return self._request_resource('getMinSOC')
|
|
|
|
def getMaxSOC(self):
|
|
return self._request_resource('getMaxSOC')
|
|
|
|
def startCharge(self):
|
|
return self._request_resource('startCharge', (), 'put')
|
|
|
|
def stopCharge(self):
|
|
return self._request_resource('stopCharge', (), 'put')
|
|
|
|
def startDischarge(self):
|
|
return self._request_resource('startDischarge', (), 'put')
|
|
|
|
def stopDischarge(self):
|
|
return self._request_resource('stopDischarge', (), 'put')
|
|
|
|
def getMinimumChargePower(self):
|
|
return self._request_resource('getMinimumChargePower')
|
|
|
|
def getMaximumChargePower(self):
|
|
return self._request_resource('getMaximumChargePower')
|
|
|
|
def getMinimumDischargePower(self):
|
|
return self._request_resource('getMinimumDischargePower')
|
|
|
|
def getMaximumDischargePower(self):
|
|
return self._request_resource('getMaximumDischargePower')
|
|
|
|
def setPowerSetpoint(self, setPoint):
|
|
|
|
P_UpperLimit = 10.0
|
|
P_LowerLimit = -10.0
|
|
|
|
return self._request_resource('setP', (), 'put', cast_to_cm(max(P_LowerLimit, min(setPoint, P_UpperLimit))).parseToJSON())
|
|
|
|
def isPSetpointEnabled(self):
|
|
return self._request_resource('isPSetpointEnabled')
|
|
|
|
def getActiveEnergyImport(self): # TODO not working/implemented
|
|
return self._request_resource('getActiveEnergyImport')
|
|
|
|
def getActiveEnergyExport(self): # TODO not working/implemented
|
|
return self._request_resource('getActiveEnergyExport')
|
|
|
|
def getReactiveEnergyImport(self): # TODO not working/implemented
|
|
return self._request_resource('getReactiveEnergyImport')
|
|
|
|
def getReactiveEnergyExport(self): # TODO not working/implemented
|
|
return self._request_resource('getReactiveEnergyExport')
|
|
|
|
def getNodeConfiguration(self): # TODO not working/implemented
|
|
return self._request_resource('getNodeConfiguration')
|
|
|
|
def getEVSEName(self):
|
|
return self._request_resource('getEVSEName')
|
|
|
|
def getEVSELogicalNameplate(self): # TODO not working/implemented
|
|
return self._request_resource('getEVSELogicalNameplate')
|
|
|
|
def getEVSEPhysicalNameplate(self): # TODO not working/implemented
|
|
return self._request_resource('getEVSEPhysicalNameplate')
|
|
|
|
def getEVSEHealth(self): # TODO not working/implemented
|
|
return self._request_resource('getEVSEHealth')
|
|
|
|
def getGPSLocation(self): # TODO not working/implemented
|
|
return self._request_resource('getGPSLocation')
|
|
|
|
def getNumberOfEvents(self): # TODO not working/implemented
|
|
return self._request_resource('getNumberOfEvents')
|
|
|
|
def getEvents(self): # TODO not working/implemented
|
|
return self._request_resource('getEvents')
|
|
|
|
def getNumberOfAlarms(self): # TODO not working/implemented
|
|
return self._request_resource('getNumberOfAlarms')
|
|
|
|
def getAlarms(self): # TODO not working/implemented
|
|
return self._request_resource('getAlarms')
|
|
|
|
def getNumberOfUnacknowledgedAlarms(self): # TODO not working/implemented
|
|
return self._request_resource('getNumberOfUnacknowledgedAlarms')
|
|
|
|
def getUnacknowledgedAlarms(self): # TODO not working/implemented
|
|
return self._request_resource('getUnacknowledgedAlarms')
|
|
|
|
def acknowledgeAlarms(self): # TODO not working/implemented
|
|
return self._request_resource('acknowledgeAlarms')
|
|
|
|
def getRatedP(self):
|
|
return self._request_resource('getRatedP')
|
|
|
|
def getRatedQ(self):
|
|
return self._request_resource('getRatedQ')
|
|
|
|
def getRatedS(self):
|
|
return self._request_resource('getRatedS')
|
|
|
|
def getRatedU(self):
|
|
return self._request_resource('getRatedU')
|
|
|
|
def getRatedf(self):
|
|
return self._request_resource('getRatedf')
|
|
|
|
def getInverterName(self): # TODO not working/implemented
|
|
return self._request_resource('getInverterName')
|
|
|
|
def getInverterLogicalNameplate(self): # TODO not working/implemented
|
|
return self._request_resource('getInverterLogicalNameplate')
|
|
|
|
def getInverterHealth(self): # TODO not working/implemented
|
|
return self._request_resource('getInverterHealth')
|
|
|
|
def getDCVoltage(self): # TODO not working/implemented
|
|
return self._request_resource('getDCVoltage')
|
|
|
|
def getDCPower(self): # TODO not working/implemented
|
|
return self._request_resource('getDCPower')
|
|
|
|
def isDCContactorClosed(self): # TODO not working/implemented
|
|
return self._request_resource('isDCContactorClosed')
|
|
|
|
@staticmethod
|
|
def getAvailableEVSE():
|
|
return list(EVSE.__CHARGINGPOSTS.keys())
|