Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if exc_val:
self.thread.exception = exc_val
else:
# remove very-short-lived async actions
if elapsed_s(self.thread) < min_thread_time_s:
vd().threads.remove(self.thread)
class ProfileSheet(Sheet):
columns = [
Column('funcname', getter=lambda col,row: codestr(row.code)),
Column('filename', getter=lambda col,row: os.path.split(row.code.co_filename)[-1] if not isinstance(row.code, str) else ''),
Column('linenum', type=int, getter=lambda col,row: row.code.co_firstlineno if not isinstance(row.code, str) else None),
Column('inlinetime_us', type=int, getter=lambda col,row: row.inlinetime*1000000),
Column('totaltime_us', type=int, getter=lambda col,row: row.totaltime*1000000),
ColumnAttr('callcount', type=int),
Column('avg_inline_us', type=int, getter=lambda col,row: row.inlinetime*1000000/row.callcount),
Column('avg_total_us', type=int, getter=lambda col,row: row.totaltime*1000000/row.callcount),
ColumnAttr('reccallcount', type=int),
ColumnAttr('calls'),
Column('callers', getter=lambda col,row: col.sheet.callers[row.code]),
]
nKeys=3
def reload(self):
self.rows = self.source
self.orderBy(self.column('inlinetime_us'), reverse=True)
self.callers = collections.defaultdict(list) # [row.code] -> list(code)
for r in self.rows:
calls = getattr(r, 'calls', None)
if elapsed_s(self.thread) < min_thread_time_s:
vd().threads.remove(self.thread)
class ProfileSheet(Sheet):
columns = [
Column('funcname', getter=lambda col,row: codestr(row.code)),
Column('filename', getter=lambda col,row: os.path.split(row.code.co_filename)[-1] if not isinstance(row.code, str) else ''),
Column('linenum', type=int, getter=lambda col,row: row.code.co_firstlineno if not isinstance(row.code, str) else None),
Column('inlinetime_us', type=int, getter=lambda col,row: row.inlinetime*1000000),
Column('totaltime_us', type=int, getter=lambda col,row: row.totaltime*1000000),
ColumnAttr('callcount', type=int),
Column('avg_inline_us', type=int, getter=lambda col,row: row.inlinetime*1000000/row.callcount),
Column('avg_total_us', type=int, getter=lambda col,row: row.totaltime*1000000/row.callcount),
ColumnAttr('reccallcount', type=int),
ColumnAttr('calls'),
Column('callers', getter=lambda col,row: col.sheet.callers[row.code]),
]
nKeys=3
def reload(self):
self.rows = self.source
self.orderBy(self.column('inlinetime_us'), reverse=True)
self.callers = collections.defaultdict(list) # [row.code] -> list(code)
for r in self.rows:
calls = getattr(r, 'calls', None)
if calls:
for callee in calls:
self.callers[callee.code].append(r)
c.sheet = self.source[0]
return c
ColumnsSheet.addCommand(None, 'resize-source-rows-max', 'for c in selectedRows or [cursorRow]: c.width = c.getMaxWidth(source.visibleRows)')
ColumnsSheet.addCommand('&', 'join-cols', 'rows.insert(cursorRowIndex, combineColumns(selectedRows or fail("no columns selected to concatenate")))')
class SheetsSheet(Sheet):
rowtype = 'sheets'
precious = False
columns = [
ColumnAttr('name', width=30),
ColumnAttr('nRows', type=int),
ColumnAttr('nCols', type=int),
ColumnAttr('nVisibleCols', type=int),
ColumnAttr('cursorDisplay'),
ColumnAttr('keyColNames'),
ColumnAttr('source'),
ColumnAttr('progressPct'),
]
nKeys = 1
def newRow(self):
return Sheet('', columns=[ColumnItem('', 0)], rows=[])
def reload(self):
self.rows = self.source
SheetsSheet.addCommand(ENTER, 'open-row', 'dest=cursorRow; vd.sheets.remove(sheet) if not sheet.precious else None; vd.push(dest)')
SheetsSheet.addCommand('g'+ENTER, 'open-rows', 'for vs in selectedRows: vd.push(vs)')
SheetsSheet.addCommand('g^R', 'reload-selected', 'for vs in selectedRows or rows: vs.reload()')
SheetsSheet.addCommand('gC', 'columns-selected', 'vd.push(ColumnsSheet("all_columns", source=selectedRows or rows[1:]))')
SheetsSheet.addCommand('gI', 'describe-selected', 'vd.push(DescribeSheet("describe_all", source=selectedRows or rows[1:]))')
# source: vd.allSheets (with BaseSheet as weakref keys)
class GraveyardSheet(SheetsSheet):
rowtype = 'undead sheets' # rowdef: BaseSheet
def reload(self):
self.rows = list(vs for vs in self.source.keys() if vs not in vd().sheets)
class HelpSheet(Sheet):
'Show all commands available to the source sheet.'
rowtype = 'commands'
precious = False
columns = [
ColumnAttr('sheet'),
ColumnAttr('longname'),
Column('keystrokes', getter=lambda col,row: col.sheet.revbinds.get(row.longname)),
Column('description', getter=lambda col,row: col.sheet.cmddict[(row.sheet, row.longname)].helpstr),
ColumnAttr('execstr', width=0),
ColumnAttr('logged', 'replayable', width=0),
]
nKeys = 2
@asyncthread
def reload(self):
from pkg_resources import resource_filename
cmdlist = TsvSheet('cmdlist', source=Path(resource_filename(__name__, 'commands.tsv')))
cmdlist.reload_sync()
self.cmddict = {}
for cmdrow in cmdlist.rows:
self.cmddict[(cmdrow.sheet, cmdrow.longname)] = cmdrow
self.addRow(r)
# if an ordering has been specified, sort the sheet
if self._ordering:
vd.sync(self.sort())
class IndexSheet(Sheet):
rowtype = 'sheets'
precious = False
columns = [
ColumnAttr('name'),
ColumnAttr('rows', 'nRows', type=int),
ColumnAttr('cols', 'nCols', type=int),
ColumnAttr('keys', 'keyColNames'),
ColumnAttr('source'),
]
nKeys = 1
def newRow(self):
return Sheet('', columns=[ColumnItem('', 0)], rows=[])
def openRow(self, row):
return row # rowdef is Sheet
def getSheet(self, k):
for vs in self.rows:
if vs.name == k:
return vs
# add the rest of the rows
for r in vd.Progress(itsource, gerund='loading', total=0):
self.addRow(r)
# if an ordering has been specified, sort the sheet
if self._ordering:
vd.sync(self.sort())
class IndexSheet(Sheet):
rowtype = 'sheets'
precious = False
columns = [
ColumnAttr('name'),
ColumnAttr('rows', 'nRows', type=int),
ColumnAttr('cols', 'nCols', type=int),
ColumnAttr('keys', 'keyColNames'),
ColumnAttr('source'),
]
nKeys = 1
def newRow(self):
return Sheet('', columns=[ColumnItem('', 0)], rows=[])
def openRow(self, row):
return row # rowdef is Sheet
def getSheet(self, k):
for vs in self.rows:
if vs.name == k:
return vs
return c
ColumnsSheet.addCommand(None, 'resize-source-rows-max', 'for c in selectedRows or [cursorRow]: c.width = c.getMaxWidth(source.visibleRows)')
ColumnsSheet.addCommand('&', 'join-cols', 'rows.insert(cursorRowIndex, combineColumns(selectedRows or fail("no columns selected to concatenate")))')
class SheetsSheet(Sheet):
rowtype = 'sheets'
precious = False
columns = [
ColumnAttr('name', width=30),
ColumnAttr('nRows', type=int),
ColumnAttr('nCols', type=int),
ColumnAttr('nVisibleCols', type=int),
ColumnAttr('cursorDisplay'),
ColumnAttr('keyColNames'),
ColumnAttr('source'),
ColumnAttr('progressPct'),
]
nKeys = 1
def newRow(self):
return Sheet('', columns=[ColumnItem('', 0)], rows=[])
def reload(self):
self.rows = self.source
SheetsSheet.addCommand(ENTER, 'open-row', 'dest=cursorRow; vd.sheets.remove(sheet) if not sheet.precious else None; vd.push(dest)')
SheetsSheet.addCommand('g'+ENTER, 'open-rows', 'for vs in selectedRows: vd.push(vs)')
SheetsSheet.addCommand('g^R', 'reload-selected', 'for vs in selectedRows or rows: vs.reload()')
SheetsSheet.addCommand('gC', 'columns-selected', 'vd.push(ColumnsSheet("all_columns", source=selectedRows or rows[1:]))')
SheetsSheet.addCommand('gI', 'describe-selected', 'vd.push(DescribeSheet("describe_all", source=selectedRows or rows[1:]))')
for vs in self.rows:
if vs.name == k:
return vs
class SheetsSheet(IndexSheet):
columns = [
ColumnAttr('name', width=30),
ColumnAttr('shortcut'),
ColumnAttr('nRows', type=int),
ColumnAttr('nCols', type=int),
ColumnAttr('nVisibleCols', type=int),
ColumnAttr('cursorDisplay'),
ColumnAttr('keyColNames'),
ColumnAttr('source'),
ColumnAttr('progressPct'),
# ColumnAttr('threads', 'currentThreads', type=vlen),
]
nKeys = 1
def reload(self):
self.rows = self.source
@VisiData.property
@drawcache
def _evalcontexts(vd):
return {}
## VisiData sheet manipulation
@VisiData.global_api
def replace(vd, vs):
c = type(self.source[0])._coltype()
c.sheet = self.source[0]
return c
ColumnsSheet.addCommand(None, 'resize-source-rows-max', 'for c in selectedRows or [cursorRow]: c.width = c.getMaxWidth(source.visibleRows)')
ColumnsSheet.addCommand('&', 'join-cols', 'rows.insert(cursorRowIndex, combineColumns(selectedRows or fail("no columns selected to concatenate")))')
class SheetsSheet(Sheet):
rowtype = 'sheets'
precious = False
columns = [
ColumnAttr('name', width=30),
ColumnAttr('nRows', type=int),
ColumnAttr('nCols', type=int),
ColumnAttr('nVisibleCols', type=int),
ColumnAttr('cursorDisplay'),
ColumnAttr('keyColNames'),
ColumnAttr('source'),
ColumnAttr('progressPct'),
]
nKeys = 1
def newRow(self):
return Sheet('', columns=[ColumnItem('', 0)], rows=[])
def reload(self):
self.rows = self.source
SheetsSheet.addCommand(ENTER, 'open-row', 'dest=cursorRow; vd.sheets.remove(sheet) if not sheet.precious else None; vd.push(dest)')
SheetsSheet.addCommand('g'+ENTER, 'open-rows', 'for vs in selectedRows: vd.push(vs)')
SheetsSheet.addCommand('g^R', 'reload-selected', 'for vs in selectedRows or rows: vs.reload()')
def newRow(self):
return Sheet('', columns=[ColumnItem('', 0)], rows=[])
def openRow(self, row):
return row # rowdef is Sheet
def getSheet(self, k):
for vs in self.rows:
if vs.name == k:
return vs
class SheetsSheet(IndexSheet):
columns = [
ColumnAttr('name', width=30),
ColumnAttr('shortcut'),
ColumnAttr('nRows', type=int),
ColumnAttr('nCols', type=int),
ColumnAttr('nVisibleCols', type=int),
ColumnAttr('cursorDisplay'),
ColumnAttr('keyColNames'),
ColumnAttr('source'),
ColumnAttr('progressPct'),
# ColumnAttr('threads', 'currentThreads', type=vlen),
]
nKeys = 1
def reload(self):
self.rows = self.source
@VisiData.property