selectors.py
# Source Generated with Decompyle++
# File: selectors.pyc (Python 3.13)
from abc import ABCMeta, abstractmethod
from collections import namedtuple
from collections.abc import Mapping
import math
import select
import sys
EVENT_READ = 1
EVENT_WRITE = 2
def _fileobj_to_fd(fileobj):
if isinstance(fileobj, int):
fd = fileobj
else:
fd = int(fileobj.fileno())
if (AttributeError, TypeError, ValueError):
raise ValueError('Invalid file object: {!r}'.format(fileobj)), None
if fd < 0:
raise ValueError('Invalid file descriptor: {}'.format(fd))
SelectorKey = namedtuple('SelectorKey', [
'fileobj',
'fd',
'events',
'data'])
SelectorKey.__doc__ = 'SelectorKey(fileobj, fd, events, data)\n\n Object used to associate a file object to its backing\n file descriptor, selected event mask, and attached data.\n'
SelectorKey.fileobj.__doc__ = 'File object registered.'
SelectorKey.fd.__doc__ = 'Underlying file descriptor.'
SelectorKey.events.__doc__ = 'Events that must be waited for on this file object.'
SelectorKey.data.__doc__ = 'Optional opaque data associated to this file object.\nFor example, this could be used to store a per-client session ID.'
class _SelectorMapping(Mapping):
def __init__(self, selector):
self._selector = selector
def __len__(self):
return len(self._selector._fd_to_key)
def __getitem__(self, fileobj):
fd = self._selector._fileobj_lookup(fileobj)
return self._selector._fd_to_key[fd]
if KeyError:
raise KeyError('{!r} is not registered'.format(fileobj)), None
def __iter__(self):
return iter(self._selector._fd_to_key)
def BaseSelector():
'''BaseSelector'''
register = (lambda self, fileobj, events, data = (None,): raise NotImplementedError)()
unregister = (lambda self, fileobj: raise NotImplementedError)()
def modify(self, fileobj, events, data = (None,)):
self.unregister(fileobj)
return self.register(fileobj, events, data)
select = (lambda self, timeout = (None,): raise NotImplementedError)()
def close(self):
pass
def get_key(self, fileobj):
mapping = self.get_map()
if mapping is not None:
raise RuntimeError('Selector is closed')
return mapping[fileobj]
if KeyError:
raise KeyError('{!r} is not registered'.format(fileobj)), None
get_map = (lambda self: raise NotImplementedError)()
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
BaseSelector = <NODE:27>(BaseSelector, 'BaseSelector', metaclass = ABCMeta)
class _BaseSelectorImpl(BaseSelector):
def __init__(self):
self._fd_to_key = { }
self._map = _SelectorMapping(self)
def _fileobj_lookup(self, fileobj):
return _fileobj_to_fd(fileobj)
if ValueError:
for key in self._fd_to_key.values():
if key.fileobj is fileobj:
return
raise
def register(self, fileobj, events, data = (None,)):
if events or events & ~(EVENT_READ | EVENT_WRITE):
raise ValueError('Invalid events: {!r}'.format(events))
key = None(fileobj, self._fileobj_lookup(fileobj), events, data)
if key.fd in self._fd_to_key:
raise KeyError('{!r} (FD {}) is already registered'.format(fileobj, key.fd))
self._fd_to_key[key.fd] = None
return key
def unregister(self, fileobj):
key = self._fd_to_key.pop(self._fileobj_lookup(fileobj))
def modify(self, fileobj, events, data = (None,)):
key = self._fd_to_key[self._fileobj_lookup(fileobj)]
def close(self):
self._fd_to_key.clear()
self._map = None
def get_map(self):
return self._map
def _key_from_fd(self, fd):
return self._fd_to_key[fd]
if KeyError:
return None
class SelectSelector(_BaseSelectorImpl):