def runs_todo(withLines):
    runs = dd.from_delayed([dask.delayed(_sa4.provide_runs)(f.parent, f['name'], part.start, part.qty) for (fid, f) in withLines.iterrows() for (_, part) in partition_lines(f.lines).iterrows()]).compute().sort_values(['file_id', 'line'])
    withNames = pd.merge(runs, withLines[['file_id', 'parent', 'name', 'st_size', 'lines']], on='file_id')
    byFile = withNames.groupby('file_id')
    runs = pd.concat([withNames, byFile.apply((lambda g: pd.DataFrame(dict(time_end=g.time.shift((- 1)))))), byFile.apply((lambda g: pd.DataFrame(dict(line_end=g.line.shift((- 1))))))], axis=1)
    runs.line_end = np.where(runs.line_end.isnull(), runs.lines, runs.line_end)
    return runs.sort_values(['st_size', 'file_id', 'line']).reset_index(drop=True)
def runs_todo(withLines):
    runs = dd.from_delayed([dask.delayed(_sa4.provide_runs)(f.parent, f['name'], part.start, part.qty) for (fid, f) in withLines.iterrows() for (_, part) in partition_lines(f.lines).iterrows()]).compute().sort_values(['file_id', 'line'])
    withNames = pd.merge(runs, withLines[['file_id', 'parent', 'name', 'st_size', 'lines']], on='file_id')
    byFile = withNames.groupby('file_id')
    runs = pd.concat([withNames, byFile.apply((lambda g: pd.DataFrame(dict(time_end=g.time.shift((- 1)))))), byFile.apply((lambda g: pd.DataFrame(dict(line_end=g.line.shift((- 1))))))], axis=1)
    runs.line_end = np.where(runs.line_end.isnull(), runs.lines, runs.line_end)
    return runs.sort_values(['st_size', 'file_id', 'line']).reset_index(drop=True)