# -*- coding: utf-8 -*- {{{
# vim: set fenc=utf-8 ft=python sw=4 ts=4 sts=4 et:
#
# Copyright 2020, Battelle Memorial Institute.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# This material was prepared as an account of work sponsored by an agency of
# the United States Government. Neither the United States Government nor the
# United States Department of Energy, nor Battelle, nor any of their
# employees, nor any jurisdiction or organization that has cooperated in the
# development of these materials, makes any warranty, express or
# implied, or assumes any legal liability or responsibility for the accuracy,
# completeness, or usefulness or any information, apparatus, product,
# software, or process disclosed, or represents that its use would not infringe
# privately owned rights. Reference herein to any specific commercial product,
# process, or service by trade name, trademark, manufacturer, or otherwise
# does not necessarily constitute or imply its endorsement, recommendation, or
# favoring by the United States Government or any agency thereof, or
# Battelle Memorial Institute. The views and opinions of authors expressed
# herein do not necessarily state or reflect those of the
# United States Government or any agency thereof.
#
# PACIFIC NORTHWEST NATIONAL LABORATORY operated by
# BATTELLE for the UNITED STATES DEPARTMENT OF ENERGY
# under Contract DE-AC05-76RL01830
# }}}
import suds.client
import suds.wsse
import logging
logger = logging.getLogger('chargepoint')
SERVICE_WSDL_URL = "https://webservices.chargepoint.com/cp_api_5.0.wsdl"
CPAPI_SUCCESS = '100'
XMPP_EVENTS = [
'station_charging_session_start',
'station_charging_session_stop',
'station_charging_session_update'
]
[docs]class CPAPIException(Exception):
"""Generic Chargepoint API Exception.
:param response_code: Exception code.
:param response_text: Exception description.
"""
def __init__(self, response_code, response_text):
self._responseCode = response_code
self._responseText = response_text
def __str__(self):
return '{0} : {1}'.format(self._responseCode, self._responseText)
[docs]class CPOrganization(object):
"""Represents an organization within the ChargePoint network.
:param cpn_id: Chargepoint Network ID.
:param organization_id: Chargepoint Org ID.
:param name:
"""
def __init__(self, cpn_id, organization_id, name="Unknown"):
self._cpn_id = cpn_id
self._organization_id = organization_id
self._name = name
[docs] def orgID(self):
"""Returns Chargepoint orgID
:return orgID: 'cpn_id:organization_id'.
"""
return '{0}:{1}'.format(self._cpn_id, self._organization_id)
[docs]class CPGroupManager(object):
"""Manger for a Chargepoint group and its stations.
:param cps: Chargepoint Service object.
:param group: CPStationGroup object.
:param stations: List of CPStation objects belonging to the CPStationGroup.
"""
def __init__(self, cps, group, stations):
self._cps = cps
self._group = group
self._stations = stations
[docs] def refreshGroupStationData(self):
"""For all stations belonging to group, refresh load data."""
stationData = self._cps.getLoad(sgID=self._group.id)
for data in stationData:
for station in self._stations:
if station.id == data.stationID:
station._data['stationLoadData'] = data
[docs]class CPStationGroup(object):
"""Wrapper around the getStationGroups() return by Chargepoint API.
:param cps: Chargepoint Service object.
:param groupsdata: Returned from Chargepoint API. Defined below.
(groupsdata){
sgID = 00001
orgID = "1:ORG00001"
sgName = "Main St Garage"
organizationName = "My Organization Name"
stationData[] =
(stationData){
stationID = "1:00001"
Geo =
(geoData){
Lat = "12.345678901234567"
Long = "-123.456789012345678"
}
},
...
}
:property id: sgID
:property name: sgName
:property organization: CPOrganization __str__ representation
:property station_ids: List of IDs for stations in belonging to the group
"""
def __init__(self, cps, groupsdata):
self._cps = cps
self._groupsdata = groupsdata
def __str__(self):
return '{0} ({1})'.format(self.name, self.id)
@property
def id(self):
return self._groupsdata.sgID
@property
def name(self):
return self._groupsdata.sgName
@property
def organization(self):
cpn_id, organization_id = self._groupsdata.orgID.split(":")
return CPOrganization(cpn_id, organization_id, self._groupsdata.organizationName)
@property
def station_ids(self):
return [s.stationID for s in self._groupsdata.stationData]
[docs]class CPStation(object):
"""Wrapper around the getStations() return by Chargepoint API.
Data surrounding a Chargepoint Station can generally be categorized as static or dynamic. Chargepoint API has two
basic calls, getLoad and getStation, that each return station data. getLoad returns the stationLoadData SUDS
object, and getStation returns the stationDataExtended SUDS object. These are each kept as separate meta-data
parameters.
:param cps: Chargepoint Service object.
:param sld: stationLoadData SUDS object.
:param sde: stationDataExtended SUDS object.
(stationDataExtended){
stationID = "1:00001"
stationManufacturer = "ChargePoint"
stationModel = "CT2100-HD-CDMA-CCR"
stationMacAddr = "0123:4567:89AB:CDEF"
stationSerialNum = "000000000001"
stationActivationDate = 2016-01-01 12:23:45
Address = "1 Main St "
City = "Oakland"
State = "California"
Country = "United States"
postalCode = "94607"
Port[] =
(portData){
portNumber = "1"
stationName = "CHARGEPOINT / MAIN 001"
Geo =
(geoData){
Lat = "12.345678901234567"
Long = "-123.456789012345678"
}
Description = "Use garage entrance on Main St., turn right and follow ...
Reservable = 0
Level = "L1"
Connector = "NEMA 5-20R"
Voltage = "120"
Current = "16"
Power = "1.920"
estimatedCost = 0.0
},
(portData){
portNumber = "2"
stationName = "CHARGEPOINT / MAIN 001"
Geo =
(geoData){
Lat = "12.345678901234567"
Long = "-123.456789012345678"
}
Description = "Use garage entrance on Main St., turn right and follow ...
Reservable = 0
Level = "L2"
Connector = "J1772"
Voltage = "240"
Current = "30"
Power = "6.600"
estimatedCost = 0.0
},
Pricing[] =
(pricingSpecification){
Type = "None"
startTime = 00:00:00
endTime = 23:59:59
minPrice = 0.0
maxPrice = 0.0
unitPricePerHour = 0.0
unitPricePerSession = 1.0
unitPricePerKWh = 0.2
},
numPorts = 2
mainPhone = "1-888-123-4567"
currencyCode = "USD"
orgID = "1:ORG00001"
organizationName = "My Organization Name"
sgID = "00001, 00002, 00003, 00004, 00005, 00006, 00007, 00008, 00009, ...
sgName = "Main St Garage, Public Garages, California Stations, ...
}
(stationloaddata){
stationID = "1:00001"
stationName = "CHARGEPOINT / MAIN 001"
Address = "1 Main St, Oakland, California, 94607, United States"
stationLoad = 5.43
Port[] =
(stationPortData){
portNumber = "1"
userID = None
credentialID = None
shedState = 0
portLoad = 0.0
allowedLoad = 0.0
percentShed = "0"
},
(stationPortData){
portNumber = "2"
credentialID = "ABC000123456"
shedState = 0
portLoad = 5.43
allowedLoad = 0.0
percentShed = "0"
},
}
:property id: sde.stationID
:property manufacturer: sde.stationManufacturer
:property model: sde.stationModel
:property mac: sde.stationMacAddr
:property serial: sde.stationSerialNum
:property activationDate: sde.stationActivationDate
:property name: sld.stationName
:property load: sld.stationLoad
"""
def __init__(self, cps, sld=None, sde=None):
self._cps = cps
self._data = {
'stationLoadData': sld,
'stationDataExtended': sde
}
def __str__(self):
return '{0} ({1})'.format(self.serial, self.id)
@property
def _sld(self):
return self._data['stationLoadData']
@property
def _sde(self):
return self._data['stationDataExtended']
@property
def id(self):
return self._sde.stationID
@property
def manufacturer(self):
return self._sde.stationManufacturer
@property
def model(self):
return self._sde.stationModel
@property
def mac(self):
return self._sde.stationMacAddr
@property
def serial(self):
return self._sde.stationSerialNum
@property
def activationDate(self):
return self._sde.stationActivationDate
@property
def name(self):
return self._sld.stationName
@property
def load(self):
return self._sld.stationLoad
@property
def organization(self):
cpn_id, organization_id = self._sde.orgID.split(":")
return CPOrganization(cpn_id, organization_id, self._sde.organizationName)
@property
def ports(self):
return [CPPort(p) for p in self._sde.Port]
[docs] def refreshStationData(self):
self._data['stationLoadData'] = self._cps.getLoad(stationID=self.id)[0]
[docs] def refreshStationDataExtended(self):
self._data['stationDataExtended'] = self._cps.getStation(stationID=self.id)[0]
[docs]class CPPort(object):
def __init__(self, data=None):
self._data = data
@property
def portNumber(self):
return self._data.portNumber
@property
def level(self):
return self._data.Level
@property
def connector(self):
return self._data.Connector
@property
def voltage(self):
return self._data.Voltage
@property
def current(self):
return self._data.Current
@property
def power(self):
return self._data.Power
[docs]class CPAPIResponse(object):
"""Response object describing a chargepoint API call
:param response: SOAP object containing the API response
:property responseCode: API response Code. '100' is a successful call.
:property responseText: Short description of the designation for the API call
:method is_successful: Returns Boolean value checking whether or not responseCode is set to '100.'
"""
def __init__(self, response):
self.response = response
@property
def responseCode(self):
return self.response.responseCode
@property
def responseText(self):
return self.response.responseText
[docs] def is_successful(self):
return self.responseCode == CPAPI_SUCCESS
[docs] @staticmethod
def is_not_found(name):
logger.warning("{0} not found in result set.".format(name))
return None
[docs] @staticmethod
def get_port_value(port_number, data, attribute):
"""Returns data for a given port
:param port_number: Number of the port to access.
:param data: Larger data structure to scan for Port data.
:param attribute: Which piece of Port data to return.
:return port_data: Accessed data for given port number and attribute. Else None.
"""
if 'Port' in data:
flag = True
for port in data.Port:
if int(port.portNumber) == port_number:
flag = False
if attribute in ['Lat', 'Long']:
if 'Geo' in port:
return CPAPIResponse.check_output(attribute, port['Geo'])
else:
logger.warning('Geo not defined for this port.')
return None
else:
return CPAPIResponse.check_output(attribute, port)
if flag:
logger.warning("Station does not have a definition for port {0}".format(port_number))
else:
logger.warning("Response does not have Ports defined")
return None
[docs] @staticmethod
def check_output(attribute, parent_dict):
"""Helper method for get_port_value"""
if attribute in parent_dict:
if parent_dict[attribute] == "None":
return None
else:
return parent_dict[attribute]
else:
logger.warning("{0} not found in Port result set.".format(attribute))
return None
[docs] @staticmethod
def get_attr_from_response(name_string, response, portNum=None):
list = []
for item in response:
if not portNum:
list.append(getattr(item, name_string)
if name_string in item
else CPAPIResponse.is_not_found(name_string))
else:
list.append(CPAPIResponse.get_port_value(portNum, item, name_string))
return list
[docs]class CPAPIGetAlarmsResponse(CPAPIResponse):
def __init__(self, response):
super(CPAPIGetAlarmsResponse, self).__init__(response)
@property
def alarms(self):
if self.is_successful():
return self.response.Alarms
else:
raise CPAPIException(self.responseCode, self.responseText)
[docs] def alarmType(self, port=None):
return CPAPIResponse.get_attr_from_response('alarmType', self.alarms, port)
[docs] def alarmTime(self, port=None):
return CPAPIResponse.get_attr_from_response('alarmTime', self.alarms, port)
[docs] def clearAlarms(self, port=None):
return CPAPIResponse.get_attr_from_response('clearAlarms', self.alarms, port)
[docs]class CPAPIGetChargingSessionsResponse(CPAPIResponse):
def __init__(self, response):
super(CPAPIGetChargingSessionsResponse, self).__init__(response)
@property
def charging_sessions(self):
if self.is_successful():
return self.response.ChargingSessionData
else:
raise CPAPIException(self.responseCode, self.responseText)
[docs] def sessionID(self, port=None):
return CPAPIResponse.get_attr_from_response('sessionID', self.charging_sessions, port)
[docs] def startTime(self, port=None):
return CPAPIResponse.get_attr_from_response('startTime', self.charging_sessions, port)
[docs] def endTime(self, port=None):
return CPAPIResponse.get_attr_from_response('endTime', self.charging_sessions, port)
[docs] def Energy(self, port=None):
return CPAPIResponse.get_attr_from_response('Energy', self.charging_sessions, port)
[docs] def rfidSerialNumber(self, port=None):
return CPAPIResponse.get_attr_from_response('rfidSerialNumber', self.charging_sessions, port)
[docs] def driverAccountNumber(self, port=None):
return CPAPIResponse.get_attr_from_response('driverAccountNumber', self.charging_sessions, port)
[docs] def driverName(self, port=None):
return CPAPIResponse.get_attr_from_response('driverName', self.charging_sessions, port)
[docs]class CPAPIGetStationStatusResponse(CPAPIResponse):
def __init__(self, response):
super(CPAPIGetStationStatusResponse, self).__init__(response)
@property
def status(self):
if self.is_successful():
return self.response.stationData
else:
raise CPAPIException(self.responseCode, self.responseText)
[docs] def Status(self, port=None):
return CPAPIResponse.get_attr_from_response('Status', self.status, port)
[docs] def TimeStamp(self, port=None):
return CPAPIResponse.get_attr_from_response('TimeStamp', self.status, port)
[docs]class CPAPIGetStationsResponse(CPAPIResponse):
def __init__(self, response):
super(CPAPIGetStationsResponse, self).__init__(response)
@property
def stations(self):
if self.is_successful():
return self.response.stationData
else:
raise CPAPIException(self.responseCode, self.responseText)
[docs] def stationID(self, port=None):
return CPAPIResponse.get_attr_from_response('stationID', self.stations, port)
[docs] def stationManufacturer(self, port=None):
return CPAPIResponse.get_attr_from_response('stationManufacturer', self.stations, port)
[docs] def stationModel(self, port=None):
return CPAPIResponse.get_attr_from_response('stationModel', self.stations, port)
[docs] def stationMacAddr(self, port=None):
return CPAPIResponse.get_attr_from_response('stationMacAddr', self.stations, port)
[docs] def stationSerialNum(self, port=None):
return CPAPIResponse.get_attr_from_response('stationSerialNum', self.stations, port)
[docs] def Address(self, port=None):
return CPAPIResponse.get_attr_from_response('Address', self.stations, port)
[docs] def City(self, port=None):
return CPAPIResponse.get_attr_from_response('City', self.stations, port)
[docs] def State(self, port=None):
return CPAPIResponse.get_attr_from_response('State', self.stations, port)
[docs] def Country(self, port=None):
return CPAPIResponse.get_attr_from_response('Country', self.stations, port)
[docs] def postalCode(self, port=None):
return CPAPIResponse.get_attr_from_response('postalCode', self.stations, port)
[docs] def numPorts(self, port=None):
return CPAPIResponse.get_attr_from_response('numPorts', self.stations, port)
[docs] def currencyCode(self, port=None):
return CPAPIResponse.get_attr_from_response('currencyCode', self.stations, port)
[docs] def orgID(self, port=None):
return CPAPIResponse.get_attr_from_response('orgID', self.stations, port)
[docs] def mainPhone(self, port=None):
return CPAPIResponse.get_attr_from_response('mainPhone', self.stations, port)
[docs] def organizationName(self, port=None):
return CPAPIResponse.get_attr_from_response('organizationName', self.stations, port)
[docs] def sgID(self, port=None):
return CPAPIResponse.get_attr_from_response('sgID', self.stations, port)
[docs] def sgName(self, port=None):
return CPAPIResponse.get_attr_from_response('sgName', self.stations, port)
[docs] def portNumber(self, port=None):
return CPAPIResponse.get_attr_from_response('portNumber', self.stations, port)
[docs] def stationName(self, port=None):
return CPAPIResponse.get_attr_from_response('stationName', self.stations, port)
[docs] def Lat(self, port=None):
return CPAPIResponse.get_attr_from_response('Lat', self.stations, port)
[docs] def Long(self, port=None):
return CPAPIResponse.get_attr_from_response('Long', self.stations, port)
[docs] def Description(self, port=None):
return CPAPIResponse.get_attr_from_response('Description', self.stations, port)
[docs] def Reservable(self, port=None):
return CPAPIResponse.get_attr_from_response('Reservable', self.stations, port)
[docs] def Level(self, port=None):
return CPAPIResponse.get_attr_from_response('Level', self.stations, port)
[docs] def Mode(self, port=None):
return CPAPIResponse.get_attr_from_response('Mode', self.stations, port)
[docs] def Voltage(self, port=None):
return CPAPIResponse.get_attr_from_response('Voltage', self.stations, port)
[docs] def Current(self, port=None):
return CPAPIResponse.get_attr_from_response('Current', self.stations, port)
[docs] def Power(self, port=None):
return CPAPIResponse.get_attr_from_response('Power', self.stations, port)
[docs] def Connector(self, port=None):
return CPAPIResponse.get_attr_from_response('Connector', self.stations, port)
[docs] @staticmethod
def pricing_helper(attribute, station):
if 'Pricing' in station:
return station.Pricing[0][attribute] \
if attribute in station.Pricing[0] \
else CPAPIResponse.is_not_found(attribute)
else:
logger.warning("No Pricing defined for station")
return None
[docs] def Type(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('Type', self.stations, port)
else:
return [self.pricing_helper('Type', station) for station in self.stations]
[docs] def startTime(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('startTime', self.stations, port)
else:
return [self.pricing_helper('startTime', station) for station in self.stations]
[docs] def endTime(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('endTime', self.stations, port)
else:
return [self.pricing_helper('endTime', station) for station in self.stations]
[docs] def minPrice(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('minPrice', self.stations, port)
else:
return [self.pricing_helper('minPrice', station) for station in self.stations]
[docs] def maxPrice(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('maxPrice', self.stations, port)
else:
return [self.pricing_helper('maxPrice', station) for station in self.stations]
[docs] def unitPricePerHour(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('unitPricePerHour', self.stations, port)
else:
return [self.pricing_helper('unitPricePerHour', station) for station in self.stations]
[docs] def unitPricePerSession(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('unitPricePerSession', self.stations, port)
else:
return [self.pricing_helper('unitPricePerSession', station) for station in self.stations]
[docs] def unitPricePerKWh(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('unitPricePerKWh', self.stations, port)
else:
return [self.pricing_helper('unitPricePerKWh', station) for station in self.stations]
[docs] def unitPriceForFirst(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('unitPriceForFirst', self.stations, port)
else:
return [self.pricing_helper('unitPriceForFirst', station) for station in self.stations]
[docs] def unitPricePerHourThereafter(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('unitPricePerHourThereafter', self.stations, port)
else:
return [self.pricing_helper('unitPricePerHourThereafter', station) for station in self.stations]
[docs] def sessionTime(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('sessionTime', self.stations, port)
else:
return [self.pricing_helper('sessionTime', station) for station in self.stations]
[docs]class CPAPIGetStationRightsResponse(CPAPIResponse):
def __init__(self, response):
super(CPAPIGetStationRightsResponse, self).__init__(response)
@property
def rights(self):
if self.is_successful():
return self.response.rightsData
else:
raise CPAPIException(self.responseCode, self.responseText)
[docs]class CPAPIGetLoadResponse(CPAPIResponse):
def __init__(self, response):
super(CPAPIGetLoadResponse, self).__init__(response)
@property
def station_data(self):
if self.is_successful():
return self.response.stationData
else:
raise CPAPIException(self.responseCode, self.responseText)
[docs] def stationLoad(self, port=None):
return CPAPIResponse.get_attr_from_response('stationLoad', self.station_data, port)
[docs] def portLoad(self, port=None):
return CPAPIResponse.get_attr_from_response('portLoad', self.station_data, port)
[docs] def allowedLoad(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('allowedLoad', self.station_data, port)
else:
list = []
for station in self.station_data:
al = 0.0
for port in station.Port:
allowed_load = self.get_port_value(int(port.portNumber), station, 'allowedLoad')
shed_state = self.get_port_value(int(port.portNumber), station, 'shedState')
if shed_state and allowed_load > al:
al = allowed_load
list.append(al)
return list
[docs] def percentShed(self, port=None):
if port:
return CPAPIResponse.get_attr_from_response('percentShed', self.station_data, port)
else:
list = []
for station in self.station_data:
ps = 0.0
for port in station.Port:
percent_shed = self.get_port_value(int(port.portNumber), station, 'percentShed')
shed_state = self.get_port_value(int(port.portNumber), station, 'shedState')
if shed_state and percent_shed > ps:
ps = percent_shed
list.append(ps)
return list
[docs] def shedState(self, port=None):
return CPAPIResponse.get_attr_from_response('shedState', self.station_data, port)
[docs]class CPService(object):
"""
Python wrapper around the Chargepoint WebServices API.
Current Version: 5.0
Docs: ChargePoint_Web_Services_API_Guide_Ver4.1_Rev5.pdf
"""
def __init__(self, username=None, password=None):
"""
Use a default API Username/Password if nothing is provided. These credentials are
created on the chargepoint website: htpp://na.chargepoint.com, tab=organizations
"""
self._username = username
self._password = password
self._suds_client = None
@property
def _client(self):
"""Initialize the SUDS client if necessary."""
if self._suds_client is None:
self._suds_client = suds.client.Client(SERVICE_WSDL_URL)
# Add SOAP Security tokens
self.set_security_token()
return self._suds_client
@property
def _soap_service(self):
return self._client.service
[docs] def set_security_token(self):
# Add SOAP Security tokens
security = suds.wsse.Security()
token = suds.wsse.UsernameToken(self._username, self._password)
security.tokens.append(token)
self._suds_client.set_options(wsse=security)
[docs] def set_client(self, client):
self._suds_client = client
self.set_security_token()
[docs] def clearAlarms(self, **kwargs):
"""Clears the Alarms of given group or station based on given query parameters.
:param **kwargs: any top-level kwarg in the following query. Most frequently queried via stationID.
Query:
(clearAlarmsSearchQuery){
orgID = None
organizationName = None
stationID = None
stationName = None
sgID = None
sgName = None
startTime = None
endTime = None
portNumber = None
alarmType = None
clearReason = None
}
:returns SOAP reply object. If successful, there will be a responseCode of '100'.
"""
searchQuery = self._client.factory.create('clearAlarmsSearchQuery')
for k, v in kwargs.items():
setattr(searchQuery, k, v)
response = self._soap_service.clearAlarms(searchQuery)
return CPAPIResponse(response)
[docs] def clearShedState(self, **kwargs):
"""Clears the shed state of given group or station.
:param sgID (as kwarg): groupID of stations to clear.
:param stationID (as kwarg): (Optional) ID of individual station to clear. If this is used, only that station will have
a cleared shed state, even with the use of sgID.
:returns SOAP reply object. If successful, there will be a responseCode of '100'.
"""
searchQuery = self._client.factory.create('shedQueryInputData')
if 'stationID' in kwargs.keys():
setattr(searchQuery, 'shedStation', {'stationID': kwargs['stationID']})
elif 'sgID' in kwargs.keys():
setattr(searchQuery, 'shedGroup', {'sgID': kwargs['sgID']})
else:
raise Exception('Must have either sgID or stationID as kwarg')
response = self._soap_service.clearShedState(searchQuery)
return CPAPIResponse(response)
[docs] def getAlarms(self, **kwargs):
"""Returns any active alarms matching the search query.
:param **kwargs: any top-level kwarg in the following query. Most frequently queried via stationID.
Query:
(getAlarmsSearchQuery){
orgID = None
organizationName = None
stationID = None
stationName = None
sgID = None
sgName = None
startTime = None
endTime = None
portNumber = None
startRecord = None
numTransactions = None
}
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
Alarms[] =
(oalarms){
stationID = "1:00001"
stationName = "CHARGEPOINT / MAIN 001"
stationModel = "CT2100-HD-CCR"
orgID = "1:ORG00001"
organizationName = "My Organization Name"
stationManufacturer = Chargepoint
stationSerialNum = "000000000001"
portNumber = None
alarmType = "Reachable"
alarmTime = 2016-12-12 12:34:56+00:00
recordNumber = 1
},
...
moreFlag = 0
}
"""
searchQuery = self._client.factory.create('getAlarmsSearchQuery')
for k, v in kwargs.items():
setattr(searchQuery, k, v)
response = self._soap_service.getAlarms(searchQuery)
return CPAPIGetAlarmsResponse(response)
[docs] def getCPNInstances(self):
"""Returns ChargePoint network objects.
Generally not useful expect that it returns the all important CPNID which is needed to construct the orgID,
described as CPNID:CompanyID.
For North America, the CPNID is '1'.
"""
return self._soap_service.getCPNInstances()
[docs] def getChargingSessionData(self, **kwargs):
"""Returns a list of charging sessions based on search query.
Returns a list of Charging Sessions. If there are more than 100
records returned by the query, there will be a MoreFlag return
value of 1.
:param **kwargs: any top-level kwarg in the following query. Most frequently queried via stationID.
Query:
(sessionSearchdata){
stationID = None
sessionID = None
stationName = None
Address = None
City = None
State = None
Country = None
postalCode = None
Proximity = None
proximityUnit = None
fromTimeStamp = None
toTimeStamp = None
startRecord = None
Geo =
(geoData){
Lat = None
Long = None
}
}
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
ChargingSessionData[] =
(sessionSearchResultdata){
stationID = "1:00001"
stationName = "CHARGEPOINT / MAIN 001"
portNumber = "2"
Address = "1 Main St, Oakland, California, 94607, United States"
City = "Oakland"
State = "California"
Country = "United States"
postalCode = "94607"
sessionID = 12345678
Energy = 12.345678
startTime = 2016-01-01 01:01:01+00:00
endTime = 2016-01-01 12:12:02+00:00
userID = "123456"
recordNumber = 1
credentialID = "123456789"
},
...
moreFlag = 0
}
"""
searchQuery = self._client.factory.create('sessionSearchdata')
for k, v in kwargs.items():
setattr(searchQuery, k, v)
response = self._soap_service.getChargingSessionData(searchQuery)
return CPAPIGetChargingSessionsResponse(response)
[docs] def getLoad(self, **kwargs):
"""Returns current load of charging station sessions.
Returns Load on Charging stations/groups as defined by input query. If sgID is not included, many group level
parameters will be returned as 'None.'
:param **kwargs: sgID or stationID.
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
numStations = None
groupName = None
sgLoad = None
stationData[] =
(stationloaddata){
stationID = "1:000013"
stationName = "CHARGEPOINT / MAIN 001"
Address = "1 Main St, Oakland, California, 94607, United States"
stationLoad = 1.1
Port[] =
(stationPortData){
portNumber = "1"
userID = None
credentialID = None
shedState = 0
portLoad = 0.0
allowedLoad = 0.0
percentShed = "0"
},
(stationPortData){
portNumber = "2"
userID = "123456"
credentialID = "123456789"
shedState = 0
portLoad = 1.1
allowedLoad = 0.0
percentShed = "0"
},
},
...
}
"""
# @ToDo: Figure out what type of request searchQuery should be here.
searchQuery = self._client.factory.create('stationSearchRequestExtended')
for k, v in kwargs.items():
setattr(searchQuery, k, v)
response = self._soap_service.getLoad(searchQuery)
return CPAPIGetLoadResponse(response)
[docs] def getOrgsAndStationGroups(self, **kwargs):
"""Returns orgnaizations and their station groups.
Get all organization and station group identifiers.
:param **kwargs: any top-level kwarg in the following query. Most frequently queried via stationID.
Query:
(getOrgsAndStationGroupsSearchQuery){
orgID = None
organizationName = None
sgID = None
sgName = None
}
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
orgData[] =
(ohostdata){
orgID = "1:ORG00001"
organizationName = "My Organization Name"
sgData[] =
(sgData){
sgID = 00001
sgName = "Main St Garage"
parentGroupID = "0"
},
...
},
...
}
"""
searchQuery = self._client.factory.create('getOrgsAndStationGroupsSearchQuery')
for k, v in kwargs.items():
setattr(searchQuery, k, v)
response = self._soap_service.getOrgsAndStationGroups(searchQuery)
return CPAPIResponse(response)
[docs] def getStationGroupDetails(self, sgID, *stationID):
"""Gives details for a given station group.
:param sgID: groupID of stations to clear.
:param stationID: (Optional) ID of individual station to clear. If this is used, only that station will be
returned in the stationData list. If this parameter is given, numStations will return 1
:returns SOAP reply object. If successful, there will be a responseCode of '100'.
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
groupName = "My Group Name"
numStations = 1
stationData[] =
(stationGroupData){
stationID = "1:00001"
stationName = "CHARGEPOINT / MAIN 001"
Address = "1 Main St, Oakland, California, 94607, United States"
},
...
}
"""
if not stationID:
response = self._soap_service.getStationGroupDetails(sgID)
else:
response = self._soap_service.getStationGroupDetails(sgID, stationID)
return CPAPIResponse(response)
[docs] def getStationGroups(self, orgID):
"""Returns a list of groups and their stations belonging to an organization.
:param orgID: Chargepoint Organization ID
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
groupData[] =
(groupsdata){
sgID = 00001
orgID = "1:ORG00001"
sgName = "Main St Garage"
organizationName = "My Organization Name"
stationData[] =
(stationData){
stationID = "1:00001"
Geo =
(geoData){
Lat = "12.345678901234567"
Long = "-123.456789012345678"
}
},
...
},
...
}
"""
response = self._soap_service.getStationGroups(orgID)
return CPAPIResponse(response)
[docs] def getStationRights(self, **kwargs):
"""Returns station rights profiles as defined by the given query parameters.
It is worth noting that there ay be more than one rights profile for a given station. A profile defined the
relationship between a charge station and a group and a charge station may belong to multiple groups.
:param **kwargs: any top-level kwarg in the following query. Most frequently queried via stationID.
Query:
(stationRightsSearchRequest){
stationID = None
stationManufacturer = None
stationModel = None
stationName = None
serialNumber = None
Address = None
City = None
State = None
Country = None
postalCode = None
Proximity = None
proximityUnit = None
Connector = None
Voltage = None
Current = None
Power = None
demoSerialNumber =
(serialNumberData){
serialNumber[] = <empty>
}
Reservable = None
Geo =
(geoData){
Lat = None
Long = None
}
Level = None
Mode = None
Pricing =
(pricingOptions){
startTime = None
Duration = None
energyRequired = None
vehiclePower = None
}
orgID = None
organizationName = None
sgID = None
sgName = None
provisionDateRange =
(provisionDateRange){
startDate = None
endDate = None
}
currentFault = None
portStatus = None
adminStatus = None
networkStatus = None
provisionStatus = None
startRecord = None
}
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
rightsData[] =
(rightsData){
sgID = "00001"
sgName = "Main St Garage"
stationRightsProfile = "network_manager"
stationData[] =
(stationDataRights){
stationID = "1:00001"
stationName = "CHARGEPOINT / MAIN 001"
stationSerialNum = "000000000001"
stationMacAddr = "0123:4567:89AB:CDEF"
},
...
},
...
moreFlag = 0
}
"""
searchQuery = self._client.factory.create('stationRightsSearchRequest')
for k, v in kwargs.items():
setattr(searchQuery, k, v)
response = self._soap_service.getStationRights(searchQuery)
return CPAPIGetStationRightsResponse(response)
[docs] def getStationStatus(self, station):
"""Get port-level charging status for a given station
:param station: stationID to query
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
stationData[] =
(oStatusdata){
stationID = "1:00001"
Port[] =
(portDataStatus){
portNumber = "1"
Status = "AVAILABLE"
TimeStamp = 2016-12-12 12:34:56+00:00
},
...
},
...
moreFlag = 0
}
"""
response = self._soap_service.getStationStatus({'stationID': station})
return CPAPIGetStationStatusResponse(response)
[docs] def getStations(self, **kwargs):
"""Returns a list of Chargepoint Stations based on keyword query args
It is worth noting that only stations the client has access to will be returned.
:param **kwargs: any top-level kwarg in the following query. Most frequently queried via stationID.
Query:
(stationSearchRequestExtended){
stationID = None
stationManufacturer = None
stationModel = None
stationName = None
serialNumber = None
Address = None
City = None
State = None
Country = None
postalCode = None
Proximity = None
proximityUnit = None
Connector = None
Voltage = None
Current = None
Power = None
demoSerialNumber =
(serialNumberData){
serialNumber[] = <empty>
}
Reservable = None
Geo =
(geoData){
Lat = None
Long = None
}
Level = None
Mode = None
Pricing =
(pricingOptions){
startTime = None
Duration = None
energyRequired = None
vehiclePower = None
}
orgID = None
organizationName = None
sgID = None
sgName = None
stationActivationDate = None
startRecord = None
numStations = None
}
Reply:
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
stationData[] =
(stationDataExtended){
stationID = "1:00001"
stationManufacturer = "ChargePoint"
stationModel = "CT2100-HD-CCR"
stationMacAddr = "0123:4567:89AB:CDEF"
stationSerialNum = "000000000001"
stationActivationDate = 2016-01-01 12:23:45
Address = "1 Main St "
City = "Oakland"
State = "California"
Country = "United States"
postalCode = "94607"
Port[] =
(portData){
portNumber = "1"
stationName = "CHARGEPOINT / MAIN 001"
Geo =
(geoData){
Lat = "12.345678901234567"
Long = "-123.456789012345678"
}
Description = "Use garage entrance on Main St., turn right and follow ...
Reservable = 0
Level = "L1"
Connector = "NEMA 5-20R"
Voltage = "120"
Current = "16"
Power = "1.920"
estimatedCost = 0.0
},
...
Pricing[] =
(pricingSpecification){
Type = "None"
startTime = 00:00:00
endTime = 23:59:59
minPrice = 0.0
maxPrice = 0.0
unitPricePerHour = 0.0
unitPricePerSession = 1.0
unitPricePerKWh = 0.2
},
numPorts = 2
mainPhone = "1-888-123-4567"
currencyCode = "USD"
orgID = "1:ORG00001"
organizationName = "My Organization Name"
sgID = "00001, 00002, 00003, 00004, 00005, 00006, 00007, 00008, 00009, ...
sgName = "Main St Garage, Public Garages, California Stations, ...
},
...
moreFlag = 0
}
"""
searchQuery = self._client.factory.create('stationSearchRequestExtended')
for k, v in kwargs.items():
setattr(searchQuery, k, v)
response = self._soap_service.getStations(searchQuery)
return CPAPIGetStationsResponse(response)
[docs] def getUsers(self, **kwargs):
"""Returns a list of Users as defined by the given query parameters
:param **kwargs: any top-level kwarg in the following query. Most frequently queried via userID or credentialID.
Query:
(getUsersSearchRequest){
userID = None
firstName = None
lastName = None
lastModifiedTimeStamp = None
Connection =
(connectionDataRequest){
Status =
(connectedUserStatusTypes){
value = None
}
customInfo =
(customInfoData){
Key = None
Value = None
}
}
managementRealm =
(managementRealmRequest){
Status =
(managedUserStatusTypes){
value = None
}
customInfo =
(customInfoData){
Key = None
Value = None
}
}
credentialID = None
startRecord = None
numUsers = None
}
Reply
(reply){
responseCode = "100"
responseText = "API input request executed successfully."
users =
(userParams){
user[] =
(userData){
lastModifiedTimestamp = 2016-11-11 01:23:45+00:00
userID = "123456"
firstName = "John"
lastName = "Doe"
Connection =
(connectionData){
Status = "APPROVED"
requestTimeStamp = 2016-11-11 01:23:45+00:00
customInfos =
(customInfosData){
customInfo[] =
(customInfoData){
Key = "Custom Key"
Value = "Custom Value"
},
...
}
}
managementRealm = ""
credentialIDs =
(credentialIDsData){
credentialID[] =
"123456789",
...
}
recordNumber = 1
},
...
moreFlag = 0
}
}
"""
searchQuery = self._client.factory.create('getUsersSearchRequest')
for k, v in kwargs.items():
setattr(searchQuery, k, v)
response = self._soap_service.getUsers(searchQuery)
return CPAPIResponse(response)
[docs] def shedLoad(self, **kwargs):
"""Reduce load on a Charegepoint station.
Main functionality for reducing load on a chargepoint station. Can pass either allowedLoadPerStation OR
percentShedPerStation, but not both (one must be None).
:param **kwargs: Input parameters for shedding load. One of allowedLoadPerStation and percentshedPerStation
must be included.
Query:
(shedLoadQueryInputData){
shedGroup =
(shedLoadGroupInputData){
sgID = None
allowedLoadPerStation = None
percentShedPerStation = None
}
shedStation =
(shedLoadStationInputData){
stationID = None
allowedLoadPerStation = None
percentShedPerStation = None
Ports =
(Ports){
Port[] = <empty>
}
}
timeInterval = None
}
:returns SOAP reply object. If successful, there will be a responseCode of '100'.
"""
searchQuery = self._client.factory.create('shedLoadQueryInputData')
port = kwargs.pop('portNumber', None)
query_params = {'stationID': kwargs['stationID']}
if port:
port_params = {'allowedLoadPerPort': kwargs.pop('allowedLoad', None),
'percentShedPerPort': kwargs.pop('percentShed', None),
'portNumber': port}
query_params['Ports'] = {'Port': [port_params]}
else:
query_params['allowedLoadPerStation'] = kwargs.pop('allowedLoad', None)
query_params['percentShedPerStation'] = kwargs.pop('percentShed', None)
setattr(searchQuery, 'shedStation', query_params)
response = self._soap_service.shedLoad(searchQuery)
return CPAPIResponse(response)
[docs] def dump_methods_and_datatypes(self):
"""Debugging tool. Prints out the SOAP methods and datatypes."""
print(self._client)