diff --git a/tracetools_trace/tools/lttng.py b/tracetools_trace/tools/lttng.py index dd38953..a5f4fac 100644 --- a/tracetools_trace/tools/lttng.py +++ b/tracetools_trace/tools/lttng.py @@ -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: