niklas/ros2_tracing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add typing to tracetools_trace.tools.lttng

Browse source
This commit is contained in:
Christophe Bedard 2019-06-17 14:45:48 +02:00
parent c61b2f5db5
commit 8e51b87b57
1 changed files with 48 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

90
tracetools_trace/tools/lttng.py
View file

@ -3,6 +3,7 @@
import sys
# Temporary workaround
sys.path = ['/usr/local/lib/python3.6/site-packages'] + sys.path
from typing import List
import lttng # noqa: E402
@ -14,19 +15,20 @@ from .names import ( # noqa: E402
def lttng_setup(
session_name,
full_path,
ros_events=DEFAULT_EVENTS_ROS,
kernel_events=DEFAULT_EVENTS_KERNEL,
context_names=DEFAULT_CONTEXT):
session_name: str,
full_path: str,
ros_events: List[str] = DEFAULT_EVENTS_ROS,
kernel_events: List[str] = DEFAULT_EVENTS_KERNEL,
context_names: List[str] = DEFAULT_CONTEXT
) -> None:
"""
Set up LTTng session, with events and context.
:param session_name (str): the name of the session
:param full_path (str): the full path to the main directory to write trace data to
:param ros_events (list(str)): list of ROS events to enable
:param kernel_events (list(str)): list of kernel events to enable
:param context_names (list(str)): list of context elements to enable
:param session_name: the name of the session
:param full_path: the full path to the main directory to write trace data to
:param ros_events: list of ROS events to enable
:param kernel_events: list of kernel events to enable
:param context_names: list of context elements to enable
"""
ust_enabled = ros_events is not None and len(ros_events) > 0
kernel_enabled = kernel_events is not None and len(kernel_events) > 0
@ -80,45 +82,45 @@ def lttng_setup(
_add_context(enabled_handles, context_list)
def lttng_start(session_name):
def lttng_start(session_name: str) -> None:
"""
Start LTTng session, and check for errors.
:param session_name (str): the name of the session
:param session_name: the name of the session
"""
result = lttng.start(session_name)
if result < 0:
raise RuntimeError(f'failed to start tracing: {lttng.strerror(result)}')
def lttng_stop(session_name):
def lttng_stop(session_name: str) -> None:
"""
Stop LTTng session, and check for errors.
:param session_name (str): the name of the session
:param session_name: the name of the session
"""
result = lttng.stop(session_name)
if result < 0:
raise RuntimeError(f'failed to stop tracing: {lttng.strerror(result)}')
def lttng_destroy(session_name):
def lttng_destroy(session_name: str) -> None:
"""
Destroy LTTng session, and check for errors.
:param session_name (str): the name of the session
:param session_name: the name of the session
"""
result = lttng.destroy(session_name)
if result < 0:
raise RuntimeError(f'failed to destroy tracing session: {lttng.strerror(result)}')
def _create_events(event_names_list):
def _create_events(event_names_list: List[str]) -> List[lttng.Event]:
"""
Create events list from names.
:param event_names_list (list(str)): a list of names to create events for
:return (list(Event)): the list of events
:param event_names_list: a list of names to create events for
:return: the list of events
"""
events_list = []
for event_name in event_names_list:
@ -130,12 +132,12 @@ def _create_events(event_names_list):
return events_list
def _create_session(session_name, full_path):
def _create_session(session_name: str, full_path: str) -> None:
"""
Create session from name and full directory path, and check for errors.
:param session_name (str): the name of the session
:param full_path (str): the full path to the main directory to write trace data to
:param session_name: the name of the session
:param full_path: the full path to the main directory to write trace data to
"""
result = lttng.create(session_name, full_path)
LTTNG_ERR_EXIST_SESS = 28
@ -148,13 +150,13 @@ def _create_session(session_name, full_path):
raise RuntimeError(f'session creation failed: {lttng.strerror(result)}')
def _create_handle(session_name, domain):
def _create_handle(session_name: str, domain: lttng.Domain) -> lttng.Handle:
"""
Create a handle for a given session name and a domain, and check for errors.
:param session_name (str): the name of the session
:param domain (Domain): the domain to be used
:return (Handle): the handle
:param session_name: the name of the session
:param domain: the domain to be used
:return: the handle
"""
handle = None
handle = lttng.Handle(session_name, domain)
@ -163,25 +165,29 @@ def _create_handle(session_name, domain):
return handle
def _enable_channel(handle, channel):
def _enable_channel(handle: lttng.Handle, channel: lttng.Channel) -> None:
"""
Enable channel for a handle, and check for errors.
:param handle (Handle): the handle to be used
:param channel (Channel): the channel to enable
:param handle: the handle to be used
:param channel: the channel to enable
"""
result = lttng.enable_channel(handle, channel)
if result < 0:
raise RuntimeError(f'channel enabling failed: {lttng.strerror(result)}')
def _enable_events(handle, events_list, channel_name):
def _enable_events(
handle: lttng.Handle,
events_list: List[lttng.Event],
channel_name: str
) -> None:
"""
Enable events list for a given handle and channel name, and check for errors.
:param handle (Handle): the handle to be used
:param events_list (list(Event)): the list of events to enable
:param channel_name (str): the name of the channel to associate
:param handle: the handle to be used
:param events_list: the list of events to enable
:param channel_name: the name of the channel to associate
"""
for event in events_list:
result = lttng.enable_event(handle, event, channel_name)
@ -197,22 +203,22 @@ context_map = {
}
def _context_name_to_type(context_name):
def _context_name_to_type(context_name: str) -> int:
"""
Convert from context name to LTTng enum/constant type.
:param context_name (str): the generic name for the context
:return (int): the associated type
:param context_name: the generic name for the context
:return: the associated type
"""
return context_map.get(context_name)
def _create_context_list(context_names_list):
def _create_context_list(context_names_list: List[str]) -> List[lttng.EventContext]:
"""
Create context list from names, and check for errors.
:param context_names_list (list(str)): the list of context names
:return (list(EventContext)): the event context list
:param context_names_list: the list of context names
:return: the event context list
"""
context_list = []
for c in context_names_list:
@ -224,12 +230,12 @@ def _create_context_list(context_names_list):
return context_list
def _add_context(handles, context_list):
def _add_context(handles: List[lttng.Handle], context_list: List[lttng.EventContext]) -> None:
"""
Add context list to given handles, and check for errors.
:param handles (list(Handle)): the list of handles for which to add context
:param context_list (list(EventContext)): the list of event contexts to add to the handles
:param handles: the list of handles for which to add context
:param context_list: the list of event contexts to add to the handles
"""
for handle in handles:
for contex in context_list: