1# Copyright 2021 The Pigweed Authors 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); you may not 4# use this file except in compliance with the License. You may obtain a copy of 5# the License at 6# 7# https://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 12# License for the specific language governing permissions and limitations under 13# the License. 14"""Device tracing classes to interact with targets via RPC.""" 15 16import os 17import logging 18import tempfile 19 20from pw_rpc.callback_client.errors import RpcError 21from pw_system.device import Device 22from pw_trace import trace 23from pw_trace_tokenized import trace_tokenized 24 25_LOG = logging.getLogger(__package__) 26DEFAULT_TICKS_PER_SECOND = 1000 27 28 29class DeviceWithTracing(Device): 30 """Represents an RPC Client for a device running a Pigweed target with 31 tracing. 32 33 The target must have RPC support for the following services: 34 - tracing 35 36 Note: use this class as a base for specialized device representations. 37 """ 38 39 def __init__( 40 self, 41 *device_args, 42 ticks_per_second: int | None = None, 43 time_offset: int = 0, 44 **device_kwargs, 45 ): 46 super().__init__(*device_args, **device_kwargs) 47 48 self.time_offset = time_offset 49 50 if ticks_per_second: 51 self.ticks_per_second = ticks_per_second 52 else: 53 self.ticks_per_second = self.get_ticks_per_second() 54 _LOG.info('ticks_per_second set to %i', self.ticks_per_second) 55 56 def get_ticks_per_second(self) -> int: 57 trace_service = self.rpcs.pw.trace.proto.TraceService 58 try: 59 resp = trace_service.GetClockParameters() 60 if not resp.status.ok(): 61 _LOG.error( 62 'Failed to get clock parameters: %s. Using default value', 63 resp.status, 64 ) 65 return DEFAULT_TICKS_PER_SECOND 66 except RpcError as rpc_err: 67 _LOG.exception('%s. Using default value', rpc_err) 68 return DEFAULT_TICKS_PER_SECOND 69 70 return resp.response.clock_parameters.tick_period_seconds_denominator 71 72 def start_tracing(self) -> None: 73 """Turns on tracing on this device.""" 74 trace_service = self.rpcs.pw.trace.proto.TraceService 75 trace_service.Start() 76 77 def stop_tracing(self, trace_output_path: str = "trace.json") -> None: 78 """Turns off tracing on this device and downloads the trace file.""" 79 trace_service = self.rpcs.pw.trace.proto.TraceService 80 resp = trace_service.Stop() 81 82 # If there's no tokenizer, there's no need to transfer the trace 83 # file from the device after stopping tracing, as there's not much 84 # that can be done with it. 85 if not self.detokenizer: 86 _LOG.error('No tokenizer specified. Not transfering trace') 87 return 88 89 trace_bin_path = tempfile.NamedTemporaryFile(delete=False) 90 trace_bin_path.close() 91 try: 92 if not self.transfer_file( 93 resp.response.file_id, trace_bin_path.name 94 ): 95 return 96 97 with open(trace_bin_path.name, 'rb') as bin_file: 98 trace_data = bin_file.read() 99 events = trace_tokenized.get_trace_events( 100 [self.detokenizer.database], 101 trace_data, 102 self.ticks_per_second, 103 self.time_offset, 104 ) 105 json_lines = trace.generate_trace_json(events) 106 trace_tokenized.save_trace_file(json_lines, trace_output_path) 107 108 _LOG.info( 109 'Wrote trace file %s', 110 trace_output_path, 111 ) 112 finally: 113 os.remove(trace_bin_path.name) 114