Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_status(host, services, zconf, path):
"""
:param host: Hostname or ip to fetch status from
:type host: str
:return: The device status as a named tuple.
:rtype: pychromecast.dial.DeviceStatus or None
"""
if not host:
for service in services.copy():
service_info = get_info_from_service(service, zconf)
host, _ = get_host_from_service_info(service_info)
if host:
_LOGGER.debug("Resolved service %s to %s", service, host)
break
headers = {"content-type": "application/json"}
req = requests.get(FORMAT_BASE_URL.format(host) + path, headers=headers, timeout=10)
req.raise_for_status()
# The Requests library will fall back to guessing the encoding in case
# no encoding is specified in the response headers - which is the case
# for the Chromecast.
# The standard mandates utf-8 encoding, let's fall back to that instead
# if no encoding is provided, since the autodetection does not always
# provide correct results.
try:
self.socket = new_socket()
self.socket.settimeout(self.timeout)
self._report_connection_status(
ConnectionStatus(
CONNECTION_STATUS_CONNECTING,
NetworkAddress(self.host, self.port),
)
)
# Resolve the service name. If service is None, we're
# connecting directly to a host name or IP-address
if service:
host = None
port = None
service_info = get_info_from_service(service, self.zconf)
host, port = get_host_from_service_info(service_info)
if host and port:
try:
self.fn = service_info.properties[b"fn"].decode("utf-8")
except (AttributeError, KeyError, UnicodeError):
pass
self.logger.debug(
"[%s:%s] Resolved service %s to %s:%s",
self.fn or self.host,
self.port,
service,
host,
port,
)
self.host = host
self.port = port
else: