We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
IPCProvider argument type bug

### What was wrong?

IPCProvider correctly handles a `pathlib.Path` as its `ipc_path` argument, but the type annotation says it only accepts `str`.

### How can it be fixed?

Change it to `Union[str, Path]`


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `web3/providers/ipc.py`
Content:
```
1 from json import (
2     JSONDecodeError,
3 )
4 import logging
5 import os
6 from pathlib import (
7     Path,
8 )
9 import socket
10 import sys
11 import threading
12 from types import (
13     TracebackType,
14 )
15 from typing import (
16     Any,
17     Type,
18 )
19 
20 from web3._utils.threads import (
21     Timeout,
22 )
23 from web3.types import (
24     RPCEndpoint,
25     RPCResponse,
26 )
27 
28 from .base import (
29     JSONBaseProvider,
30 )
31 
32 
33 def get_ipc_socket(ipc_path: str, timeout: float=0.1) -> socket.socket:
34     if sys.platform == 'win32':
35         # On Windows named pipe is used. Simulate socket with it.
36         from web3._utils.windows import NamedPipe
37 
38         return NamedPipe(ipc_path)
39     else:
40         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
41         sock.connect(ipc_path)
42         sock.settimeout(timeout)
43         return sock
44 
45 
46 class PersistantSocket:
47     sock = None
48 
49     def __init__(self, ipc_path: str) -> None:
50         self.ipc_path = ipc_path
51 
52     def __enter__(self) -> socket.socket:
53         if not self.ipc_path:
54             raise FileNotFoundError("cannot connect to IPC socket at path: %r" % self.ipc_path)
55 
56         if not self.sock:
57             self.sock = self._open()
58         return self.sock
59 
60     def __exit__(
61         self, exc_type: Type[BaseException], exc_value: BaseException, traceback: TracebackType
62     ) -> None:
63         # only close the socket if there was an error
64         if exc_value is not None:
65             try:
66                 self.sock.close()
67             except Exception:
68                 pass
69             self.sock = None
70 
71     def _open(self) -> socket.socket:
72         return get_ipc_socket(self.ipc_path)
73 
74     def reset(self) -> socket.socket:
75         self.sock.close()
76         self.sock = self._open()
77         return self.sock
78 
79 
80 # type ignored b/c missing return statement is by design here
81 def get_default_ipc_path() -> str:  # type: ignore
82     if sys.platform == 'darwin':
83         ipc_path = os.path.expanduser(os.path.join(
84             "~",
85             "Library",
86             "Ethereum",
87             "geth.ipc"
88         ))
89         if os.path.exists(ipc_path):
90             return ipc_path
91 
92         ipc_path = os.path.expanduser(os.path.join(
93             "~",
94             "Library",
95             "Application Support",
96             "io.parity.ethereum",
97             "jsonrpc.ipc"
98         ))
99         if os.path.exists(ipc_path):
100             return ipc_path
101 
102         base_trinity_path = Path('~').expanduser() / '.local' / 'share' / 'trinity'
103         ipc_path = str(base_trinity_path / 'mainnet' / 'ipcs-eth1' / 'jsonrpc.ipc')
104         if Path(ipc_path).exists():
105             return str(ipc_path)
106 
107     elif sys.platform.startswith('linux') or sys.platform.startswith('freebsd'):
108         ipc_path = os.path.expanduser(os.path.join(
109             "~",
110             ".ethereum",
111             "geth.ipc"
112         ))
113         if os.path.exists(ipc_path):
114             return ipc_path
115 
116         ipc_path = os.path.expanduser(os.path.join(
117             "~",
118             ".local",
119             "share",
120             "io.parity.ethereum",
121             "jsonrpc.ipc"
122         ))
123         if os.path.exists(ipc_path):
124             return ipc_path
125 
126         base_trinity_path = Path('~').expanduser() / '.local' / 'share' / 'trinity'
127         ipc_path = str(base_trinity_path / 'mainnet' / 'ipcs-eth1' / 'jsonrpc.ipc')
128         if Path(ipc_path).exists():
129             return str(ipc_path)
130 
131     elif sys.platform == 'win32':
132         ipc_path = os.path.join(
133             "\\\\",
134             ".",
135             "pipe",
136             "geth.ipc"
137         )
138         if os.path.exists(ipc_path):
139             return ipc_path
140 
141         ipc_path = os.path.join(
142             "\\\\",
143             ".",
144             "pipe",
145             "jsonrpc.ipc"
146         )
147         if os.path.exists(ipc_path):
148             return ipc_path
149 
150     else:
151         raise ValueError(
152             "Unsupported platform '{0}'.  Only darwin/linux/win32/freebsd are "
153             "supported.  You must specify the ipc_path".format(sys.platform)
154         )
155 
156 
157 # type ignored b/c missing return statement is by design here
158 def get_dev_ipc_path() -> str:  # type: ignore
159     if sys.platform == 'darwin':
160         tmpdir = os.environ.get('TMPDIR', '')
161         ipc_path = os.path.expanduser(os.path.join(
162             tmpdir,
163             "geth.ipc"
164         ))
165         if os.path.exists(ipc_path):
166             return ipc_path
167 
168     elif sys.platform.startswith('linux') or sys.platform.startswith('freebsd'):
169         ipc_path = os.path.expanduser(os.path.join(
170             "/tmp",
171             "geth.ipc"
172         ))
173         if os.path.exists(ipc_path):
174             return ipc_path
175 
176     elif sys.platform == 'win32':
177         ipc_path = os.path.join(
178             "\\\\",
179             ".",
180             "pipe",
181             "geth.ipc"
182         )
183         if os.path.exists(ipc_path):
184             return ipc_path
185 
186         ipc_path = os.path.join(
187             "\\\\",
188             ".",
189             "pipe",
190             "jsonrpc.ipc"
191         )
192         if os.path.exists(ipc_path):
193             return ipc_path
194 
195     else:
196         raise ValueError(
197             "Unsupported platform '{0}'.  Only darwin/linux/win32/freebsd are "
198             "supported.  You must specify the ipc_path".format(sys.platform)
199         )
200 
201 
202 class IPCProvider(JSONBaseProvider):
203     logger = logging.getLogger("web3.providers.IPCProvider")
204     _socket = None
205 
206     def __init__(self, ipc_path: str=None, timeout: int=10, *args: Any, **kwargs: Any) -> None:
207         if ipc_path is None:
208             self.ipc_path = get_default_ipc_path()
209         elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):
210             self.ipc_path = str(Path(ipc_path).expanduser().resolve())
211         else:
212             raise TypeError("ipc_path must be of type string or pathlib.Path")
213 
214         self.timeout = timeout
215         self._lock = threading.Lock()
216         self._socket = PersistantSocket(self.ipc_path)
217         super().__init__()
218 
219     def __str__(self) -> str:
220         return f"<{self.__class__.__name__} {self.ipc_path}>"
221 
222     def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
223         self.logger.debug("Making request IPC. Path: %s, Method: %s",
224                           self.ipc_path, method)
225         request = self.encode_rpc_request(method, params)
226 
227         with self._lock, self._socket as sock:
228             try:
229                 sock.sendall(request)
230             except BrokenPipeError:
231                 # one extra attempt, then give up
232                 sock = self._socket.reset()
233                 sock.sendall(request)
234 
235             raw_response = b""
236             with Timeout(self.timeout) as timeout:
237                 while True:
238                     try:
239                         raw_response += sock.recv(4096)
240                     except socket.timeout:
241                         timeout.sleep(0)
242                         continue
243                     if raw_response == b"":
244                         timeout.sleep(0)
245                     elif has_valid_json_rpc_ending(raw_response):
246                         try:
247                             response = self.decode_rpc_response(raw_response)
248                         except JSONDecodeError:
249                             timeout.sleep(0)
250                             continue
251                         else:
252                             return response
253                     else:
254                         timeout.sleep(0)
255                         continue
256 
257 
258 # A valid JSON RPC response can only end in } or ] http://www.jsonrpc.org/specification
259 def has_valid_json_rpc_ending(raw_response: bytes) -> bool:
260     stripped_raw_response = raw_response.rstrip()
261     for valid_ending in [b"}", b"]"]:
262         if stripped_raw_response.endswith(valid_ending):
263             return True
264     else:
265         return False
266
```
--- 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/web3/providers/ipc.py b/web3/providers/ipc.py
--- a/web3/providers/ipc.py
+++ b/web3/providers/ipc.py
@@ -15,6 +15,7 @@
 from typing import (
     Any,
     Type,
+    Union,
 )
 
 from web3._utils.threads import (
@@ -203,7 +204,13 @@
     logger = logging.getLogger("web3.providers.IPCProvider")
     _socket = None
 
-    def __init__(self, ipc_path: str=None, timeout: int=10, *args: Any, **kwargs: Any) -> None:
+    def __init__(
+        self,
+        ipc_path: Union[str, Path] = None,
+        timeout: int = 10,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
         if ipc_path is None:
             self.ipc_path = get_default_ipc_path()
         elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):

{"golden_diff": "diff --git a/web3/providers/ipc.py b/web3/providers/ipc.py\n--- a/web3/providers/ipc.py\n+++ b/web3/providers/ipc.py\n@@ -15,6 +15,7 @@\n from typing import (\n     Any,\n     Type,\n+    Union,\n )\n \n from web3._utils.threads import (\n@@ -203,7 +204,13 @@\n     logger = logging.getLogger(\"web3.providers.IPCProvider\")\n     _socket = None\n \n-    def __init__(self, ipc_path: str=None, timeout: int=10, *args: Any, **kwargs: Any) -> None:\n+    def __init__(\n+        self,\n+        ipc_path: Union[str, Path] = None,\n+        timeout: int = 10,\n+        *args: Any,\n+        **kwargs: Any,\n+    ) -> None:\n         if ipc_path is None:\n             self.ipc_path = get_default_ipc_path()\n         elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):\n", "issue": "IPCProvider argument type bug\n\r\n### What was wrong?\r\n\r\nIPCProvider correctly handles a `pathlib.Path` as its `ipc_path` argument, but the type annotation says it only accepts `str`.\r\n\r\n### How can it be fixed?\r\n\r\nChange it to `Union[str, Path]`\r\n\n", "before_files": [{"content": "from json import (\n    JSONDecodeError,\n)\nimport logging\nimport os\nfrom pathlib import (\n    Path,\n)\nimport socket\nimport sys\nimport threading\nfrom types import (\n    TracebackType,\n)\nfrom typing import (\n    Any,\n    Type,\n)\n\nfrom web3._utils.threads import (\n    Timeout,\n)\nfrom web3.types import (\n    RPCEndpoint,\n    RPCResponse,\n)\n\nfrom .base import (\n    JSONBaseProvider,\n)\n\n\ndef get_ipc_socket(ipc_path: str, timeout: float=0.1) -> socket.socket:\n    if sys.platform == 'win32':\n        # On Windows named pipe is used. Simulate socket with it.\n        from web3._utils.windows import NamedPipe\n\n        return NamedPipe(ipc_path)\n    else:\n        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)\n        sock.connect(ipc_path)\n        sock.settimeout(timeout)\n        return sock\n\n\nclass PersistantSocket:\n    sock = None\n\n    def __init__(self, ipc_path: str) -> None:\n        self.ipc_path = ipc_path\n\n    def __enter__(self) -> socket.socket:\n        if not self.ipc_path:\n            raise FileNotFoundError(\"cannot connect to IPC socket at path: %r\" % self.ipc_path)\n\n        if not self.sock:\n            self.sock = self._open()\n        return self.sock\n\n    def __exit__(\n        self, exc_type: Type[BaseException], exc_value: BaseException, traceback: TracebackType\n    ) -> None:\n        # only close the socket if there was an error\n        if exc_value is not None:\n            try:\n                self.sock.close()\n            except Exception:\n                pass\n            self.sock = None\n\n    def _open(self) -> socket.socket:\n        return get_ipc_socket(self.ipc_path)\n\n    def reset(self) -> socket.socket:\n        self.sock.close()\n        self.sock = self._open()\n        return self.sock\n\n\n# type ignored b/c missing return statement is by design here\ndef get_default_ipc_path() -> str:  # type: ignore\n    if sys.platform == 'darwin':\n        ipc_path = os.path.expanduser(os.path.join(\n            \"~\",\n            \"Library\",\n            \"Ethereum\",\n            \"geth.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.expanduser(os.path.join(\n            \"~\",\n            \"Library\",\n            \"Application Support\",\n            \"io.parity.ethereum\",\n            \"jsonrpc.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        base_trinity_path = Path('~').expanduser() / '.local' / 'share' / 'trinity'\n        ipc_path = str(base_trinity_path / 'mainnet' / 'ipcs-eth1' / 'jsonrpc.ipc')\n        if Path(ipc_path).exists():\n            return str(ipc_path)\n\n    elif sys.platform.startswith('linux') or sys.platform.startswith('freebsd'):\n        ipc_path = os.path.expanduser(os.path.join(\n            \"~\",\n            \".ethereum\",\n            \"geth.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.expanduser(os.path.join(\n            \"~\",\n            \".local\",\n            \"share\",\n            \"io.parity.ethereum\",\n            \"jsonrpc.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        base_trinity_path = Path('~').expanduser() / '.local' / 'share' / 'trinity'\n        ipc_path = str(base_trinity_path / 'mainnet' / 'ipcs-eth1' / 'jsonrpc.ipc')\n        if Path(ipc_path).exists():\n            return str(ipc_path)\n\n    elif sys.platform == 'win32':\n        ipc_path = os.path.join(\n            \"\\\\\\\\\",\n            \".\",\n            \"pipe\",\n            \"geth.ipc\"\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.join(\n            \"\\\\\\\\\",\n            \".\",\n            \"pipe\",\n            \"jsonrpc.ipc\"\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    else:\n        raise ValueError(\n            \"Unsupported platform '{0}'.  Only darwin/linux/win32/freebsd are \"\n            \"supported.  You must specify the ipc_path\".format(sys.platform)\n        )\n\n\n# type ignored b/c missing return statement is by design here\ndef get_dev_ipc_path() -> str:  # type: ignore\n    if sys.platform == 'darwin':\n        tmpdir = os.environ.get('TMPDIR', '')\n        ipc_path = os.path.expanduser(os.path.join(\n            tmpdir,\n            \"geth.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    elif sys.platform.startswith('linux') or sys.platform.startswith('freebsd'):\n        ipc_path = os.path.expanduser(os.path.join(\n            \"/tmp\",\n            \"geth.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    elif sys.platform == 'win32':\n        ipc_path = os.path.join(\n            \"\\\\\\\\\",\n            \".\",\n            \"pipe\",\n            \"geth.ipc\"\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.join(\n            \"\\\\\\\\\",\n            \".\",\n            \"pipe\",\n            \"jsonrpc.ipc\"\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    else:\n        raise ValueError(\n            \"Unsupported platform '{0}'.  Only darwin/linux/win32/freebsd are \"\n            \"supported.  You must specify the ipc_path\".format(sys.platform)\n        )\n\n\nclass IPCProvider(JSONBaseProvider):\n    logger = logging.getLogger(\"web3.providers.IPCProvider\")\n    _socket = None\n\n    def __init__(self, ipc_path: str=None, timeout: int=10, *args: Any, **kwargs: Any) -> None:\n        if ipc_path is None:\n            self.ipc_path = get_default_ipc_path()\n        elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):\n            self.ipc_path = str(Path(ipc_path).expanduser().resolve())\n        else:\n            raise TypeError(\"ipc_path must be of type string or pathlib.Path\")\n\n        self.timeout = timeout\n        self._lock = threading.Lock()\n        self._socket = PersistantSocket(self.ipc_path)\n        super().__init__()\n\n    def __str__(self) -> str:\n        return f\"<{self.__class__.__name__} {self.ipc_path}>\"\n\n    def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:\n        self.logger.debug(\"Making request IPC. Path: %s, Method: %s\",\n                          self.ipc_path, method)\n        request = self.encode_rpc_request(method, params)\n\n        with self._lock, self._socket as sock:\n            try:\n                sock.sendall(request)\n            except BrokenPipeError:\n                # one extra attempt, then give up\n                sock = self._socket.reset()\n                sock.sendall(request)\n\n            raw_response = b\"\"\n            with Timeout(self.timeout) as timeout:\n                while True:\n                    try:\n                        raw_response += sock.recv(4096)\n                    except socket.timeout:\n                        timeout.sleep(0)\n                        continue\n                    if raw_response == b\"\":\n                        timeout.sleep(0)\n                    elif has_valid_json_rpc_ending(raw_response):\n                        try:\n                            response = self.decode_rpc_response(raw_response)\n                        except JSONDecodeError:\n                            timeout.sleep(0)\n                            continue\n                        else:\n                            return response\n                    else:\n                        timeout.sleep(0)\n                        continue\n\n\n# A valid JSON RPC response can only end in } or ] http://www.jsonrpc.org/specification\ndef has_valid_json_rpc_ending(raw_response: bytes) -> bool:\n    stripped_raw_response = raw_response.rstrip()\n    for valid_ending in [b\"}\", b\"]\"]:\n        if stripped_raw_response.endswith(valid_ending):\n            return True\n    else:\n        return False\n", "path": "web3/providers/ipc.py"}], "after_files": [{"content": "from json import (\n    JSONDecodeError,\n)\nimport logging\nimport os\nfrom pathlib import (\n    Path,\n)\nimport socket\nimport sys\nimport threading\nfrom types import (\n    TracebackType,\n)\nfrom typing import (\n    Any,\n    Type,\n    Union,\n)\n\nfrom web3._utils.threads import (\n    Timeout,\n)\nfrom web3.types import (\n    RPCEndpoint,\n    RPCResponse,\n)\n\nfrom .base import (\n    JSONBaseProvider,\n)\n\n\ndef get_ipc_socket(ipc_path: str, timeout: float=0.1) -> socket.socket:\n    if sys.platform == 'win32':\n        # On Windows named pipe is used. Simulate socket with it.\n        from web3._utils.windows import NamedPipe\n\n        return NamedPipe(ipc_path)\n    else:\n        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)\n        sock.connect(ipc_path)\n        sock.settimeout(timeout)\n        return sock\n\n\nclass PersistantSocket:\n    sock = None\n\n    def __init__(self, ipc_path: str) -> None:\n        self.ipc_path = ipc_path\n\n    def __enter__(self) -> socket.socket:\n        if not self.ipc_path:\n            raise FileNotFoundError(\"cannot connect to IPC socket at path: %r\" % self.ipc_path)\n\n        if not self.sock:\n            self.sock = self._open()\n        return self.sock\n\n    def __exit__(\n        self, exc_type: Type[BaseException], exc_value: BaseException, traceback: TracebackType\n    ) -> None:\n        # only close the socket if there was an error\n        if exc_value is not None:\n            try:\n                self.sock.close()\n            except Exception:\n                pass\n            self.sock = None\n\n    def _open(self) -> socket.socket:\n        return get_ipc_socket(self.ipc_path)\n\n    def reset(self) -> socket.socket:\n        self.sock.close()\n        self.sock = self._open()\n        return self.sock\n\n\n# type ignored b/c missing return statement is by design here\ndef get_default_ipc_path() -> str:  # type: ignore\n    if sys.platform == 'darwin':\n        ipc_path = os.path.expanduser(os.path.join(\n            \"~\",\n            \"Library\",\n            \"Ethereum\",\n            \"geth.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.expanduser(os.path.join(\n            \"~\",\n            \"Library\",\n            \"Application Support\",\n            \"io.parity.ethereum\",\n            \"jsonrpc.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        base_trinity_path = Path('~').expanduser() / '.local' / 'share' / 'trinity'\n        ipc_path = str(base_trinity_path / 'mainnet' / 'ipcs-eth1' / 'jsonrpc.ipc')\n        if Path(ipc_path).exists():\n            return str(ipc_path)\n\n    elif sys.platform.startswith('linux') or sys.platform.startswith('freebsd'):\n        ipc_path = os.path.expanduser(os.path.join(\n            \"~\",\n            \".ethereum\",\n            \"geth.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.expanduser(os.path.join(\n            \"~\",\n            \".local\",\n            \"share\",\n            \"io.parity.ethereum\",\n            \"jsonrpc.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        base_trinity_path = Path('~').expanduser() / '.local' / 'share' / 'trinity'\n        ipc_path = str(base_trinity_path / 'mainnet' / 'ipcs-eth1' / 'jsonrpc.ipc')\n        if Path(ipc_path).exists():\n            return str(ipc_path)\n\n    elif sys.platform == 'win32':\n        ipc_path = os.path.join(\n            \"\\\\\\\\\",\n            \".\",\n            \"pipe\",\n            \"geth.ipc\"\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.join(\n            \"\\\\\\\\\",\n            \".\",\n            \"pipe\",\n            \"jsonrpc.ipc\"\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    else:\n        raise ValueError(\n            \"Unsupported platform '{0}'.  Only darwin/linux/win32/freebsd are \"\n            \"supported.  You must specify the ipc_path\".format(sys.platform)\n        )\n\n\n# type ignored b/c missing return statement is by design here\ndef get_dev_ipc_path() -> str:  # type: ignore\n    if sys.platform == 'darwin':\n        tmpdir = os.environ.get('TMPDIR', '')\n        ipc_path = os.path.expanduser(os.path.join(\n            tmpdir,\n            \"geth.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    elif sys.platform.startswith('linux') or sys.platform.startswith('freebsd'):\n        ipc_path = os.path.expanduser(os.path.join(\n            \"/tmp\",\n            \"geth.ipc\"\n        ))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    elif sys.platform == 'win32':\n        ipc_path = os.path.join(\n            \"\\\\\\\\\",\n            \".\",\n            \"pipe\",\n            \"geth.ipc\"\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.join(\n            \"\\\\\\\\\",\n            \".\",\n            \"pipe\",\n            \"jsonrpc.ipc\"\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    else:\n        raise ValueError(\n            \"Unsupported platform '{0}'.  Only darwin/linux/win32/freebsd are \"\n            \"supported.  You must specify the ipc_path\".format(sys.platform)\n        )\n\n\nclass IPCProvider(JSONBaseProvider):\n    logger = logging.getLogger(\"web3.providers.IPCProvider\")\n    _socket = None\n\n    def __init__(\n        self,\n        ipc_path: Union[str, Path] = None,\n        timeout: int = 10,\n        *args: Any,\n        **kwargs: Any,\n    ) -> None:\n        if ipc_path is None:\n            self.ipc_path = get_default_ipc_path()\n        elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):\n            self.ipc_path = str(Path(ipc_path).expanduser().resolve())\n        else:\n            raise TypeError(\"ipc_path must be of type string or pathlib.Path\")\n\n        self.timeout = timeout\n        self._lock = threading.Lock()\n        self._socket = PersistantSocket(self.ipc_path)\n        super().__init__()\n\n    def __str__(self) -> str:\n        return f\"<{self.__class__.__name__} {self.ipc_path}>\"\n\n    def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:\n        self.logger.debug(\"Making request IPC. Path: %s, Method: %s\",\n                          self.ipc_path, method)\n        request = self.encode_rpc_request(method, params)\n\n        with self._lock, self._socket as sock:\n            try:\n                sock.sendall(request)\n            except BrokenPipeError:\n                # one extra attempt, then give up\n                sock = self._socket.reset()\n                sock.sendall(request)\n\n            raw_response = b\"\"\n            with Timeout(self.timeout) as timeout:\n                while True:\n                    try:\n                        raw_response += sock.recv(4096)\n                    except socket.timeout:\n                        timeout.sleep(0)\n                        continue\n                    if raw_response == b\"\":\n                        timeout.sleep(0)\n                    elif has_valid_json_rpc_ending(raw_response):\n                        try:\n                            response = self.decode_rpc_response(raw_response)\n                        except JSONDecodeError:\n                            timeout.sleep(0)\n                            continue\n                        else:\n                            return response\n                    else:\n                        timeout.sleep(0)\n                        continue\n\n\n# A valid JSON RPC response can only end in } or ] http://www.jsonrpc.org/specification\ndef has_valid_json_rpc_ending(raw_response: bytes) -> bool:\n    stripped_raw_response = raw_response.rstrip()\n    for valid_ending in [b\"}\", b\"]\"]:\n        if stripped_raw_response.endswith(valid_ending):\n            return True\n    else:\n        return False\n", "path": "web3/providers/ipc.py"}]}