We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
pandas is now a mandatory dependency for dscheduler
If pandas is not installed I get an infinite loop with the following logs:

```
tornado.httpclient.HTTPError: HTTP 500: Internal Server Error
bokeh.util.tornado - ERROR - Error thrown from periodic callback: HTTPError(500, 'Internal Server Error', HTTPResponse(_body=None,buffer=<_io.BytesIO object at 0x7f17202a1e08>,code=500,effective_url='http://localhost:9786/tasks.json',error=HTTPError(...),headers=<tornado.httputil.HTTPHeaders object at 0x7f17204ffe48>,reason='Internal Server Error',request=<tornado.httpclient.HTTPRequest object at 0x7f1720578da0>,request_time=0.005105733871459961,time_info={}))
distributed.utils - ERROR - No module named 'pandas'
Traceback (most recent call last):
  File "/volatile/ogrisel/code/distributed/distributed/utils.py", line 229, in log_errors
    yield
  File "/volatile/ogrisel/code/distributed/distributed/http/scheduler.py", line 79, in get
    from ..diagnostics.scheduler import tasks
  File "/volatile/ogrisel/code/distributed/distributed/diagnostics/scheduler.py", line 6, in <module>
    import pandas as pd
ImportError: No module named 'pandas'
```

If think this is not intended? The diagnostics callback should probably be disabled when  pandas is not available, no?


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/formats/csv.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from io import BytesIO
4 
5 from dask import do
6 from dask.dataframe import from_imperative
7 import pandas as pd
8 
9 from .compression import compressors, decompressors
10 
11 from ..executor import default_executor, ensure_default_get
12 from ..utils import ensure_bytes, log_errors
13 
14 
15 def bytes_read_csv(b, header, kwargs):
16     """ Convert a block of bytes to a Pandas DataFrame
17 
18     Parameters
19     ----------
20     b: bytestring
21         The content to be parsed with pandas.read_csv
22     header: bytestring
23         An optional header to prepend to b
24     kwargs: dict
25         A dictionary of keyword arguments to be passed to pandas.read_csv
26 
27     See Also:
28         distributed.formats.csv.read_csv
29     """
30     with log_errors():
31         compression = kwargs.pop('compression', None)
32         b2 = decompressors[compression](b)
33         bio = BytesIO()
34         if header:
35             if not header.endswith(b'\n') and not header.endswith(b'\r'):
36                 header = header + ensure_bytes(kwargs.get('lineterminator', b'\n'))
37             bio.write(header)
38         bio.write(b2)
39         bio.seek(0)
40         return pd.read_csv(bio, **kwargs)
41 
42 
43 def read_csv(block_lists, header, head, kwargs, lazy=True, collection=True,
44         executor=None):
45     """ Convert blocks of bytes to a dask.dataframe or other high-level object
46 
47     This accepts a list of lists of futures/values of bytes where each list
48     corresponds to one file, and the futures/values of bytes concatenate to
49     comprise the entire file, in order.
50 
51     Parameters
52     ----------
53     block_lists: list of lists of futures of bytes
54         The lists of bytestrings with each list corresponding to one logical file
55     header: bytestring
56         The header, found at the front of the first file, to be prepended to
57         all blocks
58     head: pd.DataFrame
59         An example Pandas DataFrame to be used for metadata
60     kwargs: dict
61         Keyword arguments to pass down to ``pd.read_csv``
62     lazy: boolean, optional (defaults to True)
63     collection: boolean, optional (defaults to True)
64 
65     Returns
66     -------
67     A dask.dataframe, or list of futures or values, depending on the value of
68     lazy and collection.
69     """
70     executor = default_executor(executor)
71 
72     dfs1 = [[do(bytes_read_csv)(blocks[0], '', kwargs)] +
73             [do(bytes_read_csv)(b, header, kwargs)
74                 for b in blocks[1:]]
75             for blocks in block_lists]
76     dfs2 = sum(dfs1, [])
77 
78     ensure_default_get(executor)
79 
80     if collection:
81         result = from_imperative(dfs2, head)
82     else:
83         result = dfs2
84 
85     if not lazy:
86         if collection:
87             result = executor.persist(result)
88         else:
89             result = executor.compute(result)
90 
91     return result
92
```

Path: `distributed/diagnostics/scheduler.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from datetime import datetime
4 import os
5 
6 import pandas as pd
7 from toolz import countby, concat, dissoc
8 
9 from ..utils import key_split
10 
11 
12 def tasks(s):
13     """ Task and worker status of scheduler """
14     processing = sum(map(len, s.processing.values()))
15 
16     return {'processing': processing,
17             'total': len(s.tasks),
18             'in-memory': len(s.who_has),
19             'ready': len(s.ready)
20                    + sum(map(len, s.stacks.values())),
21             'waiting': len(s.waiting),
22             'failed': len(s.exceptions_blame)}
23 
24 
25 def workers(s):
26     """ Information about workers
27 
28     Examples
29     --------
30     >>> workers(my_scheduler)  # doctest: +SKIP
31     {'127.0.0.1': {'cores': 3,
32                    'cpu': 0.0,
33                    'last-seen': 0.003068,
34                    'latency': 0.01584628690034151,
35                    'ports': ['54871', '50943'],
36                    'processing': {'inc': 2, 'add': 1},
37                    'disk-read': 1234,
38                    'disk-write': 1234,
39                    'network-send': 1234,
40                    'network-recv': 1234,
41                    'memory': 16701911040,
42                    'memory-percent': 85}}
43     """
44     hosts = {host: ['%s:%s' % (host, port) for port in d['ports']]
45                 for host, d in s.host_info.items()}
46 
47     processing = {host: countby(key_split, concat(s.processing[w] for w in addrs))
48                   for host, addrs in hosts.items()}
49 
50     now = datetime.now()
51 
52     result = {}
53     for host, info in s.host_info.items():
54         info = dissoc(info, 'heartbeat', 'heartbeat-port')
55         info['processing'] = processing[host]
56         result[host] = info
57         info['ports'] = list(info['ports'])
58         if 'last-seen' in info:
59             info['last-seen'] = (now - info['last-seen']).total_seconds()
60 
61     return result
62 
63 
64 def scheduler_progress_df(d):
65     """ Convert status response to DataFrame of total progress
66 
67     Consumes dictionary from status.json route
68 
69     Examples
70     --------
71     >>> d = {"ready": 5, "in-memory": 30, "waiting": 20,
72     ...      "tasks": 70, "failed": 9,
73     ...      "processing": 6,
74     ...      "other-keys-are-fine-too": ''}
75 
76     >>> scheduler_progress_df(d)  # doctest: +SKIP
77                  Count                                  Progress
78     Tasks
79     waiting         20  +++++++++++
80     ready            5  ++
81     failed           9  +++++
82     processing       6  +++
83     in-memory       30  +++++++++++++++++
84     total           70  ++++++++++++++++++++++++++++++++++++++++
85     """
86     d = d.copy()
87     d['total'] = d.pop('tasks')
88     names = ['waiting', 'ready', 'failed', 'processing', 'in-memory', 'total']
89     df = pd.DataFrame(pd.Series({k: d[k] for k in names},
90                                 index=names, name='Count'))
91     if d['total']:
92         barlength = (40 * df.Count / d['total']).astype(int)
93         df['Progress'] = barlength.apply(lambda n: ('%-40s' % (n * '+').rstrip(' ')))
94     else:
95         df['Progress'] = 0
96 
97     df.index.name = 'Tasks'
98 
99     return df
100 
101 
102 def worker_status_df(d):
103     """ Status of workers as a Pandas DataFrame
104 
105     Consumes data from status.json route.
106 
107     Examples
108     --------
109     >>> d = {"other-keys-are-fine-too": '',
110     ...      "ncores": {"192.168.1.107": 4,
111     ...                 "192.168.1.108": 4},
112     ...      "processing": {"192.168.1.108": {'inc': 3, 'add': 1},
113     ...                     "192.168.1.107": {'inc': 2}},
114     ...      "bytes": {"192.168.1.108": 1000,
115     ...                "192.168.1.107": 2000}}
116 
117     >>> worker_status_df(d)
118                    Ncores  Bytes  Processing
119     Workers
120     192.168.1.107       4   2000       [inc]
121     192.168.1.108       4   1000  [add, inc]
122     """
123     names = ['ncores', 'bytes', 'processing']
124     df = pd.DataFrame({k: d[k] for k in names}, columns=names)
125     df['processing'] = df['processing'].apply(sorted)
126     df.columns = df.columns.map(str.title)
127     df.index.name = 'Workers'
128     df = df.sort_index()
129     return df
130
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/diagnostics/scheduler.py b/distributed/diagnostics/scheduler.py
--- a/distributed/diagnostics/scheduler.py
+++ b/distributed/diagnostics/scheduler.py
@@ -3,7 +3,6 @@
 from datetime import datetime
 import os
 
-import pandas as pd
 from toolz import countby, concat, dissoc
 
 from ..utils import key_split
@@ -83,6 +82,7 @@
     in-memory       30  +++++++++++++++++
     total           70  ++++++++++++++++++++++++++++++++++++++++
     """
+    import pandas as pd
     d = d.copy()
     d['total'] = d.pop('tasks')
     names = ['waiting', 'ready', 'failed', 'processing', 'in-memory', 'total']
@@ -120,6 +120,7 @@
     192.168.1.107       4   2000       [inc]
     192.168.1.108       4   1000  [add, inc]
     """
+    import pandas as pd
     names = ['ncores', 'bytes', 'processing']
     df = pd.DataFrame({k: d[k] for k in names}, columns=names)
     df['processing'] = df['processing'].apply(sorted)
diff --git a/distributed/formats/csv.py b/distributed/formats/csv.py
--- a/distributed/formats/csv.py
+++ b/distributed/formats/csv.py
@@ -3,8 +3,6 @@
 from io import BytesIO
 
 from dask import do
-from dask.dataframe import from_imperative
-import pandas as pd
 
 from .compression import compressors, decompressors
 
@@ -27,6 +25,7 @@
     See Also:
         distributed.formats.csv.read_csv
     """
+    import pandas as pd
     with log_errors():
         compression = kwargs.pop('compression', None)
         b2 = decompressors[compression](b)
@@ -67,6 +66,7 @@
     A dask.dataframe, or list of futures or values, depending on the value of
     lazy and collection.
     """
+    from dask.dataframe import from_imperative
     executor = default_executor(executor)
 
     dfs1 = [[do(bytes_read_csv)(blocks[0], '', kwargs)] +

{"golden_diff": "diff --git a/distributed/diagnostics/scheduler.py b/distributed/diagnostics/scheduler.py\n--- a/distributed/diagnostics/scheduler.py\n+++ b/distributed/diagnostics/scheduler.py\n@@ -3,7 +3,6 @@\n from datetime import datetime\n import os\n \n-import pandas as pd\n from toolz import countby, concat, dissoc\n \n from ..utils import key_split\n@@ -83,6 +82,7 @@\n     in-memory       30  +++++++++++++++++\n     total           70  ++++++++++++++++++++++++++++++++++++++++\n     \"\"\"\n+    import pandas as pd\n     d = d.copy()\n     d['total'] = d.pop('tasks')\n     names = ['waiting', 'ready', 'failed', 'processing', 'in-memory', 'total']\n@@ -120,6 +120,7 @@\n     192.168.1.107       4   2000       [inc]\n     192.168.1.108       4   1000  [add, inc]\n     \"\"\"\n+    import pandas as pd\n     names = ['ncores', 'bytes', 'processing']\n     df = pd.DataFrame({k: d[k] for k in names}, columns=names)\n     df['processing'] = df['processing'].apply(sorted)\ndiff --git a/distributed/formats/csv.py b/distributed/formats/csv.py\n--- a/distributed/formats/csv.py\n+++ b/distributed/formats/csv.py\n@@ -3,8 +3,6 @@\n from io import BytesIO\n \n from dask import do\n-from dask.dataframe import from_imperative\n-import pandas as pd\n \n from .compression import compressors, decompressors\n \n@@ -27,6 +25,7 @@\n     See Also:\n         distributed.formats.csv.read_csv\n     \"\"\"\n+    import pandas as pd\n     with log_errors():\n         compression = kwargs.pop('compression', None)\n         b2 = decompressors[compression](b)\n@@ -67,6 +66,7 @@\n     A dask.dataframe, or list of futures or values, depending on the value of\n     lazy and collection.\n     \"\"\"\n+    from dask.dataframe import from_imperative\n     executor = default_executor(executor)\n \n     dfs1 = [[do(bytes_read_csv)(blocks[0], '', kwargs)] +\n", "issue": "pandas is now a mandatory dependency for dscheduler\nIf pandas is not installed I get an infinite loop with the following logs:\n\n```\ntornado.httpclient.HTTPError: HTTP 500: Internal Server Error\nbokeh.util.tornado - ERROR - Error thrown from periodic callback: HTTPError(500, 'Internal Server Error', HTTPResponse(_body=None,buffer=<_io.BytesIO object at 0x7f17202a1e08>,code=500,effective_url='http://localhost:9786/tasks.json',error=HTTPError(...),headers=<tornado.httputil.HTTPHeaders object at 0x7f17204ffe48>,reason='Internal Server Error',request=<tornado.httpclient.HTTPRequest object at 0x7f1720578da0>,request_time=0.005105733871459961,time_info={}))\ndistributed.utils - ERROR - No module named 'pandas'\nTraceback (most recent call last):\n  File \"/volatile/ogrisel/code/distributed/distributed/utils.py\", line 229, in log_errors\n    yield\n  File \"/volatile/ogrisel/code/distributed/distributed/http/scheduler.py\", line 79, in get\n    from ..diagnostics.scheduler import tasks\n  File \"/volatile/ogrisel/code/distributed/distributed/diagnostics/scheduler.py\", line 6, in <module>\n    import pandas as pd\nImportError: No module named 'pandas'\n```\n\nIf think this is not intended? The diagnostics callback should probably be disabled when  pandas is not available, no?\n\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom io import BytesIO\n\nfrom dask import do\nfrom dask.dataframe import from_imperative\nimport pandas as pd\n\nfrom .compression import compressors, decompressors\n\nfrom ..executor import default_executor, ensure_default_get\nfrom ..utils import ensure_bytes, log_errors\n\n\ndef bytes_read_csv(b, header, kwargs):\n    \"\"\" Convert a block of bytes to a Pandas DataFrame\n\n    Parameters\n    ----------\n    b: bytestring\n        The content to be parsed with pandas.read_csv\n    header: bytestring\n        An optional header to prepend to b\n    kwargs: dict\n        A dictionary of keyword arguments to be passed to pandas.read_csv\n\n    See Also:\n        distributed.formats.csv.read_csv\n    \"\"\"\n    with log_errors():\n        compression = kwargs.pop('compression', None)\n        b2 = decompressors[compression](b)\n        bio = BytesIO()\n        if header:\n            if not header.endswith(b'\\n') and not header.endswith(b'\\r'):\n                header = header + ensure_bytes(kwargs.get('lineterminator', b'\\n'))\n            bio.write(header)\n        bio.write(b2)\n        bio.seek(0)\n        return pd.read_csv(bio, **kwargs)\n\n\ndef read_csv(block_lists, header, head, kwargs, lazy=True, collection=True,\n        executor=None):\n    \"\"\" Convert blocks of bytes to a dask.dataframe or other high-level object\n\n    This accepts a list of lists of futures/values of bytes where each list\n    corresponds to one file, and the futures/values of bytes concatenate to\n    comprise the entire file, in order.\n\n    Parameters\n    ----------\n    block_lists: list of lists of futures of bytes\n        The lists of bytestrings with each list corresponding to one logical file\n    header: bytestring\n        The header, found at the front of the first file, to be prepended to\n        all blocks\n    head: pd.DataFrame\n        An example Pandas DataFrame to be used for metadata\n    kwargs: dict\n        Keyword arguments to pass down to ``pd.read_csv``\n    lazy: boolean, optional (defaults to True)\n    collection: boolean, optional (defaults to True)\n\n    Returns\n    -------\n    A dask.dataframe, or list of futures or values, depending on the value of\n    lazy and collection.\n    \"\"\"\n    executor = default_executor(executor)\n\n    dfs1 = [[do(bytes_read_csv)(blocks[0], '', kwargs)] +\n            [do(bytes_read_csv)(b, header, kwargs)\n                for b in blocks[1:]]\n            for blocks in block_lists]\n    dfs2 = sum(dfs1, [])\n\n    ensure_default_get(executor)\n\n    if collection:\n        result = from_imperative(dfs2, head)\n    else:\n        result = dfs2\n\n    if not lazy:\n        if collection:\n            result = executor.persist(result)\n        else:\n            result = executor.compute(result)\n\n    return result\n", "path": "distributed/formats/csv.py"}, {"content": "from __future__ import print_function, division, absolute_import\n\nfrom datetime import datetime\nimport os\n\nimport pandas as pd\nfrom toolz import countby, concat, dissoc\n\nfrom ..utils import key_split\n\n\ndef tasks(s):\n    \"\"\" Task and worker status of scheduler \"\"\"\n    processing = sum(map(len, s.processing.values()))\n\n    return {'processing': processing,\n            'total': len(s.tasks),\n            'in-memory': len(s.who_has),\n            'ready': len(s.ready)\n                   + sum(map(len, s.stacks.values())),\n            'waiting': len(s.waiting),\n            'failed': len(s.exceptions_blame)}\n\n\ndef workers(s):\n    \"\"\" Information about workers\n\n    Examples\n    --------\n    >>> workers(my_scheduler)  # doctest: +SKIP\n    {'127.0.0.1': {'cores': 3,\n                   'cpu': 0.0,\n                   'last-seen': 0.003068,\n                   'latency': 0.01584628690034151,\n                   'ports': ['54871', '50943'],\n                   'processing': {'inc': 2, 'add': 1},\n                   'disk-read': 1234,\n                   'disk-write': 1234,\n                   'network-send': 1234,\n                   'network-recv': 1234,\n                   'memory': 16701911040,\n                   'memory-percent': 85}}\n    \"\"\"\n    hosts = {host: ['%s:%s' % (host, port) for port in d['ports']]\n                for host, d in s.host_info.items()}\n\n    processing = {host: countby(key_split, concat(s.processing[w] for w in addrs))\n                  for host, addrs in hosts.items()}\n\n    now = datetime.now()\n\n    result = {}\n    for host, info in s.host_info.items():\n        info = dissoc(info, 'heartbeat', 'heartbeat-port')\n        info['processing'] = processing[host]\n        result[host] = info\n        info['ports'] = list(info['ports'])\n        if 'last-seen' in info:\n            info['last-seen'] = (now - info['last-seen']).total_seconds()\n\n    return result\n\n\ndef scheduler_progress_df(d):\n    \"\"\" Convert status response to DataFrame of total progress\n\n    Consumes dictionary from status.json route\n\n    Examples\n    --------\n    >>> d = {\"ready\": 5, \"in-memory\": 30, \"waiting\": 20,\n    ...      \"tasks\": 70, \"failed\": 9,\n    ...      \"processing\": 6,\n    ...      \"other-keys-are-fine-too\": ''}\n\n    >>> scheduler_progress_df(d)  # doctest: +SKIP\n                 Count                                  Progress\n    Tasks\n    waiting         20  +++++++++++\n    ready            5  ++\n    failed           9  +++++\n    processing       6  +++\n    in-memory       30  +++++++++++++++++\n    total           70  ++++++++++++++++++++++++++++++++++++++++\n    \"\"\"\n    d = d.copy()\n    d['total'] = d.pop('tasks')\n    names = ['waiting', 'ready', 'failed', 'processing', 'in-memory', 'total']\n    df = pd.DataFrame(pd.Series({k: d[k] for k in names},\n                                index=names, name='Count'))\n    if d['total']:\n        barlength = (40 * df.Count / d['total']).astype(int)\n        df['Progress'] = barlength.apply(lambda n: ('%-40s' % (n * '+').rstrip(' ')))\n    else:\n        df['Progress'] = 0\n\n    df.index.name = 'Tasks'\n\n    return df\n\n\ndef worker_status_df(d):\n    \"\"\" Status of workers as a Pandas DataFrame\n\n    Consumes data from status.json route.\n\n    Examples\n    --------\n    >>> d = {\"other-keys-are-fine-too\": '',\n    ...      \"ncores\": {\"192.168.1.107\": 4,\n    ...                 \"192.168.1.108\": 4},\n    ...      \"processing\": {\"192.168.1.108\": {'inc': 3, 'add': 1},\n    ...                     \"192.168.1.107\": {'inc': 2}},\n    ...      \"bytes\": {\"192.168.1.108\": 1000,\n    ...                \"192.168.1.107\": 2000}}\n\n    >>> worker_status_df(d)\n                   Ncores  Bytes  Processing\n    Workers\n    192.168.1.107       4   2000       [inc]\n    192.168.1.108       4   1000  [add, inc]\n    \"\"\"\n    names = ['ncores', 'bytes', 'processing']\n    df = pd.DataFrame({k: d[k] for k in names}, columns=names)\n    df['processing'] = df['processing'].apply(sorted)\n    df.columns = df.columns.map(str.title)\n    df.index.name = 'Workers'\n    df = df.sort_index()\n    return df\n", "path": "distributed/diagnostics/scheduler.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom io import BytesIO\n\nfrom dask import do\n\nfrom .compression import compressors, decompressors\n\nfrom ..executor import default_executor, ensure_default_get\nfrom ..utils import ensure_bytes, log_errors\n\n\ndef bytes_read_csv(b, header, kwargs):\n    \"\"\" Convert a block of bytes to a Pandas DataFrame\n\n    Parameters\n    ----------\n    b: bytestring\n        The content to be parsed with pandas.read_csv\n    header: bytestring\n        An optional header to prepend to b\n    kwargs: dict\n        A dictionary of keyword arguments to be passed to pandas.read_csv\n\n    See Also:\n        distributed.formats.csv.read_csv\n    \"\"\"\n    import pandas as pd\n    with log_errors():\n        compression = kwargs.pop('compression', None)\n        b2 = decompressors[compression](b)\n        bio = BytesIO()\n        if header:\n            if not header.endswith(b'\\n') and not header.endswith(b'\\r'):\n                header = header + ensure_bytes(kwargs.get('lineterminator', b'\\n'))\n            bio.write(header)\n        bio.write(b2)\n        bio.seek(0)\n        return pd.read_csv(bio, **kwargs)\n\n\ndef read_csv(block_lists, header, head, kwargs, lazy=True, collection=True,\n        executor=None):\n    \"\"\" Convert blocks of bytes to a dask.dataframe or other high-level object\n\n    This accepts a list of lists of futures/values of bytes where each list\n    corresponds to one file, and the futures/values of bytes concatenate to\n    comprise the entire file, in order.\n\n    Parameters\n    ----------\n    block_lists: list of lists of futures of bytes\n        The lists of bytestrings with each list corresponding to one logical file\n    header: bytestring\n        The header, found at the front of the first file, to be prepended to\n        all blocks\n    head: pd.DataFrame\n        An example Pandas DataFrame to be used for metadata\n    kwargs: dict\n        Keyword arguments to pass down to ``pd.read_csv``\n    lazy: boolean, optional (defaults to True)\n    collection: boolean, optional (defaults to True)\n\n    Returns\n    -------\n    A dask.dataframe, or list of futures or values, depending on the value of\n    lazy and collection.\n    \"\"\"\n    from dask.dataframe import from_imperative\n    executor = default_executor(executor)\n\n    dfs1 = [[do(bytes_read_csv)(blocks[0], '', kwargs)] +\n            [do(bytes_read_csv)(b, header, kwargs)\n                for b in blocks[1:]]\n            for blocks in block_lists]\n    dfs2 = sum(dfs1, [])\n\n    ensure_default_get(executor)\n\n    if collection:\n        result = from_imperative(dfs2, head)\n    else:\n        result = dfs2\n\n    if not lazy:\n        if collection:\n            result = executor.persist(result)\n        else:\n            result = executor.compute(result)\n\n    return result\n", "path": "distributed/formats/csv.py"}, {"content": "from __future__ import print_function, division, absolute_import\n\nfrom datetime import datetime\nimport os\n\nfrom toolz import countby, concat, dissoc\n\nfrom ..utils import key_split\n\n\ndef tasks(s):\n    \"\"\" Task and worker status of scheduler \"\"\"\n    processing = sum(map(len, s.processing.values()))\n\n    return {'processing': processing,\n            'total': len(s.tasks),\n            'in-memory': len(s.who_has),\n            'ready': len(s.ready)\n                   + sum(map(len, s.stacks.values())),\n            'waiting': len(s.waiting),\n            'failed': len(s.exceptions_blame)}\n\n\ndef workers(s):\n    \"\"\" Information about workers\n\n    Examples\n    --------\n    >>> workers(my_scheduler)  # doctest: +SKIP\n    {'127.0.0.1': {'cores': 3,\n                   'cpu': 0.0,\n                   'last-seen': 0.003068,\n                   'latency': 0.01584628690034151,\n                   'ports': ['54871', '50943'],\n                   'processing': {'inc': 2, 'add': 1},\n                   'disk-read': 1234,\n                   'disk-write': 1234,\n                   'network-send': 1234,\n                   'network-recv': 1234,\n                   'memory': 16701911040,\n                   'memory-percent': 85}}\n    \"\"\"\n    hosts = {host: ['%s:%s' % (host, port) for port in d['ports']]\n                for host, d in s.host_info.items()}\n\n    processing = {host: countby(key_split, concat(s.processing[w] for w in addrs))\n                  for host, addrs in hosts.items()}\n\n    now = datetime.now()\n\n    result = {}\n    for host, info in s.host_info.items():\n        info = dissoc(info, 'heartbeat', 'heartbeat-port')\n        info['processing'] = processing[host]\n        result[host] = info\n        info['ports'] = list(info['ports'])\n        if 'last-seen' in info:\n            info['last-seen'] = (now - info['last-seen']).total_seconds()\n\n    return result\n\n\ndef scheduler_progress_df(d):\n    \"\"\" Convert status response to DataFrame of total progress\n\n    Consumes dictionary from status.json route\n\n    Examples\n    --------\n    >>> d = {\"ready\": 5, \"in-memory\": 30, \"waiting\": 20,\n    ...      \"tasks\": 70, \"failed\": 9,\n    ...      \"processing\": 6,\n    ...      \"other-keys-are-fine-too\": ''}\n\n    >>> scheduler_progress_df(d)  # doctest: +SKIP\n                 Count                                  Progress\n    Tasks\n    waiting         20  +++++++++++\n    ready            5  ++\n    failed           9  +++++\n    processing       6  +++\n    in-memory       30  +++++++++++++++++\n    total           70  ++++++++++++++++++++++++++++++++++++++++\n    \"\"\"\n    import pandas as pd\n    d = d.copy()\n    d['total'] = d.pop('tasks')\n    names = ['waiting', 'ready', 'failed', 'processing', 'in-memory', 'total']\n    df = pd.DataFrame(pd.Series({k: d[k] for k in names},\n                                index=names, name='Count'))\n    if d['total']:\n        barlength = (40 * df.Count / d['total']).astype(int)\n        df['Progress'] = barlength.apply(lambda n: ('%-40s' % (n * '+').rstrip(' ')))\n    else:\n        df['Progress'] = 0\n\n    df.index.name = 'Tasks'\n\n    return df\n\n\ndef worker_status_df(d):\n    \"\"\" Status of workers as a Pandas DataFrame\n\n    Consumes data from status.json route.\n\n    Examples\n    --------\n    >>> d = {\"other-keys-are-fine-too\": '',\n    ...      \"ncores\": {\"192.168.1.107\": 4,\n    ...                 \"192.168.1.108\": 4},\n    ...      \"processing\": {\"192.168.1.108\": {'inc': 3, 'add': 1},\n    ...                     \"192.168.1.107\": {'inc': 2}},\n    ...      \"bytes\": {\"192.168.1.108\": 1000,\n    ...                \"192.168.1.107\": 2000}}\n\n    >>> worker_status_df(d)\n                   Ncores  Bytes  Processing\n    Workers\n    192.168.1.107       4   2000       [inc]\n    192.168.1.108       4   1000  [add, inc]\n    \"\"\"\n    import pandas as pd\n    names = ['ncores', 'bytes', 'processing']\n    df = pd.DataFrame({k: d[k] for k in names}, columns=names)\n    df['processing'] = df['processing'].apply(sorted)\n    df.columns = df.columns.map(str.title)\n    df.index.name = 'Workers'\n    df = df.sort_index()\n    return df\n", "path": "distributed/diagnostics/scheduler.py"}]}