1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
#!/usr/bin/env python3
#
# Copyright 2019, The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Helper util libraries for parsing logcat logs."""
import asyncio
import re
from datetime import datetime
from typing import Optional, Pattern
# local import
import print_utils
def parse_logcat_datetime(timestamp: str) -> Optional[datetime]:
"""Parses the timestamp of logcat.
Params:
timestamp: for example "2019-07-01 16:13:55.221".
Returns:
a datetime of timestamp with the year now.
"""
try:
# Match the format of logcat. For example: "2019-07-01 16:13:55.221",
# because it doesn't have year, set current year to it.
timestamp = datetime.strptime(timestamp,
'%Y-%m-%d %H:%M:%S.%f')
return timestamp
except ValueError as ve:
print_utils.debug_print('Invalid line: ' + timestamp)
return None
def _is_time_out(timeout: datetime, line: str) -> bool:
"""Checks if the timestamp of this line exceeds the timeout.
Returns:
true if the timestamp exceeds the timeout.
"""
# Get the timestampe string.
cur_timestamp_str = ' '.join(re.split(r'\s+', line)[0:2])
timestamp = parse_logcat_datetime(cur_timestamp_str)
if not timestamp:
return False
return timestamp > timeout
async def _blocking_wait_for_logcat_pattern(timestamp: datetime,
pattern: Pattern,
timeout: datetime) -> Optional[str]:
# Show the year in the timestampe.
logcat_cmd = 'adb logcat -v year -v threadtime -T'.split()
logcat_cmd.append(str(timestamp))
print_utils.debug_print('[LOGCAT]:' + ' '.join(logcat_cmd))
# Create subprocess
process = await asyncio.create_subprocess_exec(
*logcat_cmd,
# stdout must a pipe to be accessible as process.stdout
stdout=asyncio.subprocess.PIPE)
while (True):
# Read one line of output.
data = await process.stdout.readline()
line = data.decode('utf-8').rstrip()
# 2019-07-01 14:54:21.946 27365 27392 I ActivityTaskManager: Displayed
# com.android.settings/.Settings: +927ms
# TODO: Detect timeouts even when there is no logcat output.
if _is_time_out(timeout, line):
print_utils.debug_print('DID TIMEOUT BEFORE SEEING ANYTHING ('
'timeout={timeout} seconds << {pattern} '
'>>'.format(timeout=timeout, pattern=pattern))
return None
if pattern.match(line):
print_utils.debug_print(
'WE DID SEE PATTERN << "{}" >>.'.format(pattern))
return line
def blocking_wait_for_logcat_pattern(timestamp: datetime,
pattern: Pattern,
timeout: datetime) -> Optional[str]:
"""Selects the line that matches the pattern and within the timeout.
Returns:
the line that matches the pattern and within the timeout.
"""
loop = asyncio.get_event_loop()
result = loop.run_until_complete(
_blocking_wait_for_logcat_pattern(timestamp, pattern, timeout))
return result
|
