46045-syslab/syslab/physical/EVSE.py

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())