-
Notifications
You must be signed in to change notification settings - Fork 28
Expand file tree
/
Copy pathstats.py
More file actions
executable file
·109 lines (92 loc) · 3.07 KB
/
stats.py
File metadata and controls
executable file
·109 lines (92 loc) · 3.07 KB
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
105
106
107
108
109
import datetime as dt
import json
import os
import sys
from typing import Dict, List, Optional, Union
from . import __version__
from .config import (
get_bustools_binary_path,
get_kallisto_binary_path,
is_dry,
)
from .dry import dummy_function
from .dry import dryable
class Stats:
"""Class used to collect kb run statistics.
"""
def __init__(self):
self.workdir = None
self.kallisto = None
self.bustools = None
self.start_time = None
self.call = None
self.commands = []
self.runtimes = []
self.end_time = None
self.elapsed = None
self.version = __version__
def start(self):
"""Start collecting statistics.
Sets start time, the command line call, and the commands array to an empty list.
Additionally, sets the kallisto and bustools paths and versions.
"""
self.start_time = dt.datetime.now()
self.call = ' '.join(sys.argv)
self.commands = []
self.workdir = os.getcwd()
if not is_dry():
# Import here to prevent circular imports
from .utils import get_bustools_version, get_kallisto_version
self.kallisto = {
'path': get_kallisto_binary_path(),
'version': '.'.join(str(i) for i in get_kallisto_version())
}
self.bustools = {
'path': get_bustools_binary_path(),
'version': '.'.join(str(i) for i in get_bustools_version())
}
def command(self, command: List[str], runtime: Optional[float] = None):
"""Report a shell command was run.
Args:
command: A shell command, represented as a list
runtime: Command runtime
"""
cmd = ' '.join(command)
self.commands.append(cmd)
self.runtimes.append(runtime or 'not measured')
def end(self):
"""End collecting statistics.
"""
self.end_time = dt.datetime.now()
self.elapsed = (self.end_time - self.start_time).total_seconds()
@dryable(dummy_function)
def save(self, path: str) -> str:
"""Save statistics as JSON to path.
Args:
path: Path to JSON
Returns:
Path to saved JSON
"""
if not is_dry():
with open(path, 'w') as f:
json.dump(self.to_dict(), f, indent=4)
return path
def to_dict(self) -> Dict[str, Union[str, float]]:
"""Convert statistics to dictionary, so that it is easily parsed
by the report-rendering functions.
Returns:
Statistics dictionary
"""
return {
'workdir': self.workdir,
'version': self.version,
'kallisto': self.kallisto,
'bustools': self.bustools,
'start_time': self.start_time.isoformat(),
'end_time': self.end_time.isoformat(),
'elapsed': self.elapsed,
'call': self.call,
'commands': self.commands,
'runtimes': self.runtimes,
}
STATS = Stats()