Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _make_evaluation_context() -> typing.Dict[str, typing.Any]:
# This import is super slow, so we do it as late as possible.
# Doing this when generating command-line arguments would be disastrous for performance.
# noinspection PyTypeChecker
pyuavcan.util.import_submodules(pyuavcan.transport)
# Populate the context with all references that may be useful for the transport expression.
context: typing.Dict[str, typing.Any] = {
'pyuavcan': pyuavcan,
}
# Pre-import transport modules for convenience.
for name, module in inspect.getmembers(pyuavcan.transport, inspect.ismodule):
if not name.startswith('_'):
context[name] = module
# Pre-import transport classes for convenience.
transport_base = pyuavcan.transport.Transport
# Suppressing MyPy false positive: https://github.com/python/mypy/issues/5374
for cls in pyuavcan.util.iter_descendants(transport_base): # type: ignore
if not cls.__name__.startswith('_') and cls is not transport_base:
def __repr__(self) -> str:
return pyuavcan.util.repr_attributes_noexcept(self,
dtype=str(pyuavcan.dsdl.get_model(self.dtype)),
input_transport_session=self.input_transport_session,
output_transport_session=self.output_transport_session,
proxy_count=self._proxy_count)
def __repr__(self) -> str:
return pyuavcan.util.repr_attributes(self, str(self))
def __repr__(self) -> str:
return pyuavcan.util.repr_attributes(self, self.specifier, self.payload_metadata)
def __repr__(self) -> str:
return pyuavcan.util.repr_attributes_noexcept(self,
dtype=str(pyuavcan.dsdl.get_model(self.dtype)),
transport_session=self.transport_session,
deserialization_failure_count=self.deserialization_failure_count,
listeners=self._listeners,
closed=self._closed)
b'in the fog?:\xff\xd1\xdd'
"""
assert max_frame_payload_bytes > 0
payload_length = sum(map(len, fragmented_payload))
if payload_length <= max_frame_payload_bytes: # SINGLE-FRAME TRANSFER
payload = fragmented_payload[0] if len(fragmented_payload) == 1 else memoryview(b''.join(fragmented_payload))
assert len(payload) == payload_length
assert max_frame_payload_bytes >= len(payload)
yield frame_factory(0, True, payload)
else: # MULTI-FRAME TRANSFER
crc_bytes = TransferCRC.new(*fragmented_payload).value_as_bytes
refragmented = pyuavcan.transport.commons.refragment(
itertools.chain(fragmented_payload, (memoryview(crc_bytes),)),
max_frame_payload_bytes
)
for frame_index, (end_of_transfer, frag) in enumerate(pyuavcan.util.mark_last(refragmented)):
yield frame_factory(frame_index, end_of_transfer, frag)
def __repr__(self) -> str:
return pyuavcan.util.repr_attributes_noexcept(self,
dtype=str(pyuavcan.dsdl.get_model(self.dtype)),
transport_session=self.transport_session,
proxy_count=self._proxy_count)
def __repr__(self) -> str:
return pyuavcan.util.repr_attributes(self,
dtype=str(pyuavcan.dsdl.get_model(self.dtype)),
transport_session=self.transport_session)
pyuavcan.util.import_submodules(pyuavcan.transport)
# Populate the context with all references that may be useful for the transport expression.
context: typing.Dict[str, typing.Any] = {
'pyuavcan': pyuavcan,
}
# Pre-import transport modules for convenience.
for name, module in inspect.getmembers(pyuavcan.transport, inspect.ismodule):
if not name.startswith('_'):
context[name] = module
# Pre-import transport classes for convenience.
transport_base = pyuavcan.transport.Transport
# Suppressing MyPy false positive: https://github.com/python/mypy/issues/5374
for cls in pyuavcan.util.iter_descendants(transport_base): # type: ignore
if not cls.__name__.startswith('_') and cls is not transport_base:
name = cls.__name__.rpartition(transport_base.__name__)[0]
assert name
context[name] = cls
return context
def __repr__(self) -> str:
"""
Implementations are advised to avoid overriding this method.
"""
return pyuavcan.util.repr_attributes(self, self.descriptor, self.protocol_parameters,
local_node_id=self.local_node_id)