Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def route_message(self, raw_msg):
"""
Route a message received from a CP.
If the message is a of type Call the corresponding hooks are executed.
If the message is of type CallResult or CallError the message is passed
to the call() function via the response_queue.
"""
try:
msg = unpack(raw_msg)
except OCPPError as e:
LOGGER.exception("Unable to parse message: '%s', it doesn't seem "
"to be valid OCPP: %s", raw_msg, e)
return
if msg.message_type_id == MessageType.Call:
await self._handle_call(msg)
elif msg.message_type_id in \
[MessageType.CallResult, MessageType.CallError]:
self._response_queue.put_nowait(msg)
def create_call_error(self, exception):
error_code = "InternalError"
error_description = "An unexpected error occurred."
error_details = {}
if isinstance(exception, OCPPError):
error_code = exception.code
error_description = exception.description
error_details = exception.details
return CallError(
self.unique_id,
error_code,
error_description,
error_details,
)
def to_exception(self):
""" Return the exception that corresponds to the CallError. """
for error in OCPPError.__subclasses__():
if error.code == self.error_code:
return error(
description=self.error_description,
details=self.error_details
)
raise UnknownCallErrorCodeError("Error code '%s' is not defined by the"
" OCPP specification", self.error_code)