We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add bar chart for worker occupancy
We currently have bar charts for worker memory and the number of tasks processing.  A similar chart to number of tasks processing would be worker occupancy, which records the sum of the amount of compute time of all tasks currently residing on the worker.  This would be exactly like the number of tasks processing chart, but use `ws.occupancy` rather than `len(ws.processing)`

I would have used this today when trying to diagnose work stealing issues.  

The relevant code for the number of processing tasks is here.  It's made slightly complex because the code for this chart is currently merged with the memory chart.  It might make sense to break these apart.

https://github.com/dask/distributed/blob/9d4f0bf2fc804f955a869febd3b51423c4382908/distributed/dashboard/components/scheduler.py#L1017-L1139

--- 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/dashboard/scheduler.py`
Content:
```
1 from urllib.parse import urljoin
2 
3 from tornado import web
4 from tornado.ioloop import IOLoop
5 
6 try:
7     import numpy as np
8 except ImportError:
9     np = False
10 
11 from .components.nvml import gpu_doc  # noqa: 1708
12 from .components.nvml import NVML_ENABLED, gpu_memory_doc, gpu_utilization_doc
13 from .components.scheduler import (
14     AggregateAction,
15     BandwidthTypes,
16     BandwidthWorkers,
17     ComputePerKey,
18     CurrentLoad,
19     MemoryByKey,
20     NBytes,
21     NBytesCluster,
22     SystemMonitor,
23     TaskGraph,
24     TaskProgress,
25     TaskStream,
26     WorkerTable,
27     events_doc,
28     graph_doc,
29     individual_doc,
30     individual_profile_doc,
31     individual_profile_server_doc,
32     profile_doc,
33     profile_server_doc,
34     status_doc,
35     stealing_doc,
36     systemmonitor_doc,
37     tasks_doc,
38     workers_doc,
39 )
40 from .core import BokehApplication
41 from .worker import counters_doc
42 
43 template_variables = {
44     "pages": ["status", "workers", "tasks", "system", "profile", "graph", "info"]
45 }
46 
47 if NVML_ENABLED:
48     template_variables["pages"].insert(4, "gpu")
49 
50 
51 def connect(application, http_server, scheduler, prefix=""):
52     bokeh_app = BokehApplication(
53         applications, scheduler, prefix=prefix, template_variables=template_variables
54     )
55     application.add_application(bokeh_app)
56     bokeh_app.initialize(IOLoop.current())
57 
58     bokeh_app.add_handlers(
59         r".*",
60         [
61             (
62                 r"/",
63                 web.RedirectHandler,
64                 {"url": urljoin((prefix or "").strip("/") + "/", r"status")},
65             )
66         ],
67     )
68 
69 
70 applications = {
71     "/system": systemmonitor_doc,
72     "/stealing": stealing_doc,
73     "/workers": workers_doc,
74     "/events": events_doc,
75     "/counters": counters_doc,
76     "/tasks": tasks_doc,
77     "/status": status_doc,
78     "/profile": profile_doc,
79     "/profile-server": profile_server_doc,
80     "/graph": graph_doc,
81     "/gpu": gpu_doc,
82     "/individual-task-stream": individual_doc(
83         TaskStream, 100, n_rectangles=1000, clear_interval="10s"
84     ),
85     "/individual-progress": individual_doc(TaskProgress, 100, height=160),
86     "/individual-graph": individual_doc(TaskGraph, 200),
87     "/individual-nbytes": individual_doc(NBytes, 100),
88     "/individual-nbytes-cluster": individual_doc(NBytesCluster, 100),
89     "/individual-cpu": individual_doc(CurrentLoad, 100, fig_attr="cpu_figure"),
90     "/individual-nprocessing": individual_doc(
91         CurrentLoad, 100, fig_attr="processing_figure"
92     ),
93     "/individual-workers": individual_doc(WorkerTable, 500),
94     "/individual-bandwidth-types": individual_doc(BandwidthTypes, 500),
95     "/individual-bandwidth-workers": individual_doc(BandwidthWorkers, 500),
96     "/individual-memory-by-key": individual_doc(MemoryByKey, 500),
97     "/individual-compute-time-per-key": individual_doc(ComputePerKey, 500),
98     "/individual-aggregate-time-per-action": individual_doc(AggregateAction, 500),
99     "/individual-scheduler-system": individual_doc(SystemMonitor, 500),
100     "/individual-profile": individual_profile_doc,
101     "/individual-profile-server": individual_profile_server_doc,
102     "/individual-gpu-memory": gpu_memory_doc,
103     "/individual-gpu-utilization": gpu_utilization_doc,
104 }
105
```
--- 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/dashboard/scheduler.py b/distributed/dashboard/scheduler.py
--- a/distributed/dashboard/scheduler.py
+++ b/distributed/dashboard/scheduler.py
@@ -19,6 +19,7 @@
     MemoryByKey,
     NBytes,
     NBytesCluster,
+    Occupancy,
     SystemMonitor,
     TaskGraph,
     TaskProgress,
@@ -90,6 +91,7 @@
     "/individual-nprocessing": individual_doc(
         CurrentLoad, 100, fig_attr="processing_figure"
     ),
+    "/individual-occupancy": individual_doc(Occupancy, 100),
     "/individual-workers": individual_doc(WorkerTable, 500),
     "/individual-bandwidth-types": individual_doc(BandwidthTypes, 500),
     "/individual-bandwidth-workers": individual_doc(BandwidthWorkers, 500),

{"golden_diff": "diff --git a/distributed/dashboard/scheduler.py b/distributed/dashboard/scheduler.py\n--- a/distributed/dashboard/scheduler.py\n+++ b/distributed/dashboard/scheduler.py\n@@ -19,6 +19,7 @@\n     MemoryByKey,\n     NBytes,\n     NBytesCluster,\n+    Occupancy,\n     SystemMonitor,\n     TaskGraph,\n     TaskProgress,\n@@ -90,6 +91,7 @@\n     \"/individual-nprocessing\": individual_doc(\n         CurrentLoad, 100, fig_attr=\"processing_figure\"\n     ),\n+    \"/individual-occupancy\": individual_doc(Occupancy, 100),\n     \"/individual-workers\": individual_doc(WorkerTable, 500),\n     \"/individual-bandwidth-types\": individual_doc(BandwidthTypes, 500),\n     \"/individual-bandwidth-workers\": individual_doc(BandwidthWorkers, 500),\n", "issue": "Add bar chart for worker occupancy\nWe currently have bar charts for worker memory and the number of tasks processing.  A similar chart to number of tasks processing would be worker occupancy, which records the sum of the amount of compute time of all tasks currently residing on the worker.  This would be exactly like the number of tasks processing chart, but use `ws.occupancy` rather than `len(ws.processing)`\r\n\r\nI would have used this today when trying to diagnose work stealing issues.  \r\n\r\nThe relevant code for the number of processing tasks is here.  It's made slightly complex because the code for this chart is currently merged with the memory chart.  It might make sense to break these apart.\r\n\r\nhttps://github.com/dask/distributed/blob/9d4f0bf2fc804f955a869febd3b51423c4382908/distributed/dashboard/components/scheduler.py#L1017-L1139\n", "before_files": [{"content": "from urllib.parse import urljoin\n\nfrom tornado import web\nfrom tornado.ioloop import IOLoop\n\ntry:\n    import numpy as np\nexcept ImportError:\n    np = False\n\nfrom .components.nvml import gpu_doc  # noqa: 1708\nfrom .components.nvml import NVML_ENABLED, gpu_memory_doc, gpu_utilization_doc\nfrom .components.scheduler import (\n    AggregateAction,\n    BandwidthTypes,\n    BandwidthWorkers,\n    ComputePerKey,\n    CurrentLoad,\n    MemoryByKey,\n    NBytes,\n    NBytesCluster,\n    SystemMonitor,\n    TaskGraph,\n    TaskProgress,\n    TaskStream,\n    WorkerTable,\n    events_doc,\n    graph_doc,\n    individual_doc,\n    individual_profile_doc,\n    individual_profile_server_doc,\n    profile_doc,\n    profile_server_doc,\n    status_doc,\n    stealing_doc,\n    systemmonitor_doc,\n    tasks_doc,\n    workers_doc,\n)\nfrom .core import BokehApplication\nfrom .worker import counters_doc\n\ntemplate_variables = {\n    \"pages\": [\"status\", \"workers\", \"tasks\", \"system\", \"profile\", \"graph\", \"info\"]\n}\n\nif NVML_ENABLED:\n    template_variables[\"pages\"].insert(4, \"gpu\")\n\n\ndef connect(application, http_server, scheduler, prefix=\"\"):\n    bokeh_app = BokehApplication(\n        applications, scheduler, prefix=prefix, template_variables=template_variables\n    )\n    application.add_application(bokeh_app)\n    bokeh_app.initialize(IOLoop.current())\n\n    bokeh_app.add_handlers(\n        r\".*\",\n        [\n            (\n                r\"/\",\n                web.RedirectHandler,\n                {\"url\": urljoin((prefix or \"\").strip(\"/\") + \"/\", r\"status\")},\n            )\n        ],\n    )\n\n\napplications = {\n    \"/system\": systemmonitor_doc,\n    \"/stealing\": stealing_doc,\n    \"/workers\": workers_doc,\n    \"/events\": events_doc,\n    \"/counters\": counters_doc,\n    \"/tasks\": tasks_doc,\n    \"/status\": status_doc,\n    \"/profile\": profile_doc,\n    \"/profile-server\": profile_server_doc,\n    \"/graph\": graph_doc,\n    \"/gpu\": gpu_doc,\n    \"/individual-task-stream\": individual_doc(\n        TaskStream, 100, n_rectangles=1000, clear_interval=\"10s\"\n    ),\n    \"/individual-progress\": individual_doc(TaskProgress, 100, height=160),\n    \"/individual-graph\": individual_doc(TaskGraph, 200),\n    \"/individual-nbytes\": individual_doc(NBytes, 100),\n    \"/individual-nbytes-cluster\": individual_doc(NBytesCluster, 100),\n    \"/individual-cpu\": individual_doc(CurrentLoad, 100, fig_attr=\"cpu_figure\"),\n    \"/individual-nprocessing\": individual_doc(\n        CurrentLoad, 100, fig_attr=\"processing_figure\"\n    ),\n    \"/individual-workers\": individual_doc(WorkerTable, 500),\n    \"/individual-bandwidth-types\": individual_doc(BandwidthTypes, 500),\n    \"/individual-bandwidth-workers\": individual_doc(BandwidthWorkers, 500),\n    \"/individual-memory-by-key\": individual_doc(MemoryByKey, 500),\n    \"/individual-compute-time-per-key\": individual_doc(ComputePerKey, 500),\n    \"/individual-aggregate-time-per-action\": individual_doc(AggregateAction, 500),\n    \"/individual-scheduler-system\": individual_doc(SystemMonitor, 500),\n    \"/individual-profile\": individual_profile_doc,\n    \"/individual-profile-server\": individual_profile_server_doc,\n    \"/individual-gpu-memory\": gpu_memory_doc,\n    \"/individual-gpu-utilization\": gpu_utilization_doc,\n}\n", "path": "distributed/dashboard/scheduler.py"}], "after_files": [{"content": "from urllib.parse import urljoin\n\nfrom tornado import web\nfrom tornado.ioloop import IOLoop\n\ntry:\n    import numpy as np\nexcept ImportError:\n    np = False\n\nfrom .components.nvml import gpu_doc  # noqa: 1708\nfrom .components.nvml import NVML_ENABLED, gpu_memory_doc, gpu_utilization_doc\nfrom .components.scheduler import (\n    AggregateAction,\n    BandwidthTypes,\n    BandwidthWorkers,\n    ComputePerKey,\n    CurrentLoad,\n    MemoryByKey,\n    NBytes,\n    NBytesCluster,\n    Occupancy,\n    SystemMonitor,\n    TaskGraph,\n    TaskProgress,\n    TaskStream,\n    WorkerTable,\n    events_doc,\n    graph_doc,\n    individual_doc,\n    individual_profile_doc,\n    individual_profile_server_doc,\n    profile_doc,\n    profile_server_doc,\n    status_doc,\n    stealing_doc,\n    systemmonitor_doc,\n    tasks_doc,\n    workers_doc,\n)\nfrom .core import BokehApplication\nfrom .worker import counters_doc\n\ntemplate_variables = {\n    \"pages\": [\"status\", \"workers\", \"tasks\", \"system\", \"profile\", \"graph\", \"info\"]\n}\n\nif NVML_ENABLED:\n    template_variables[\"pages\"].insert(4, \"gpu\")\n\n\ndef connect(application, http_server, scheduler, prefix=\"\"):\n    bokeh_app = BokehApplication(\n        applications, scheduler, prefix=prefix, template_variables=template_variables\n    )\n    application.add_application(bokeh_app)\n    bokeh_app.initialize(IOLoop.current())\n\n    bokeh_app.add_handlers(\n        r\".*\",\n        [\n            (\n                r\"/\",\n                web.RedirectHandler,\n                {\"url\": urljoin((prefix or \"\").strip(\"/\") + \"/\", r\"status\")},\n            )\n        ],\n    )\n\n\napplications = {\n    \"/system\": systemmonitor_doc,\n    \"/stealing\": stealing_doc,\n    \"/workers\": workers_doc,\n    \"/events\": events_doc,\n    \"/counters\": counters_doc,\n    \"/tasks\": tasks_doc,\n    \"/status\": status_doc,\n    \"/profile\": profile_doc,\n    \"/profile-server\": profile_server_doc,\n    \"/graph\": graph_doc,\n    \"/gpu\": gpu_doc,\n    \"/individual-task-stream\": individual_doc(\n        TaskStream, 100, n_rectangles=1000, clear_interval=\"10s\"\n    ),\n    \"/individual-progress\": individual_doc(TaskProgress, 100, height=160),\n    \"/individual-graph\": individual_doc(TaskGraph, 200),\n    \"/individual-nbytes\": individual_doc(NBytes, 100),\n    \"/individual-nbytes-cluster\": individual_doc(NBytesCluster, 100),\n    \"/individual-cpu\": individual_doc(CurrentLoad, 100, fig_attr=\"cpu_figure\"),\n    \"/individual-nprocessing\": individual_doc(\n        CurrentLoad, 100, fig_attr=\"processing_figure\"\n    ),\n    \"/individual-occupancy\": individual_doc(Occupancy, 100),\n    \"/individual-workers\": individual_doc(WorkerTable, 500),\n    \"/individual-bandwidth-types\": individual_doc(BandwidthTypes, 500),\n    \"/individual-bandwidth-workers\": individual_doc(BandwidthWorkers, 500),\n    \"/individual-memory-by-key\": individual_doc(MemoryByKey, 500),\n    \"/individual-compute-time-per-key\": individual_doc(ComputePerKey, 500),\n    \"/individual-aggregate-time-per-action\": individual_doc(AggregateAction, 500),\n    \"/individual-scheduler-system\": individual_doc(SystemMonitor, 500),\n    \"/individual-profile\": individual_profile_doc,\n    \"/individual-profile-server\": individual_profile_server_doc,\n    \"/individual-gpu-memory\": gpu_memory_doc,\n    \"/individual-gpu-utilization\": gpu_utilization_doc,\n}\n", "path": "distributed/dashboard/scheduler.py"}]}