Skip to content

Commit

Permalink
fix: fix UP031 ruff rule (#16)
Browse files Browse the repository at this point in the history
  • Loading branch information
joostlek authored Aug 6, 2024
1 parent a39946b commit fc27a54
Show file tree
Hide file tree
Showing 4 changed files with 34 additions and 38 deletions.
1 change: 0 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,6 @@ ignore = [
"B008", # Do not perform function call
"S110", # `try`-`except`-`pass` detected, consider logging the exception
"D106", # Missing docstring in public nested class
"UP031",
"B904",
"UP007", # typer needs Optional syntax
"UP038", # Use `X | Y` in `isinstance` is slower
Expand Down
20 changes: 9 additions & 11 deletions src/uvcclient/camera.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ def __init__(self, host, username, password, port=80):
self._username = username
self._password = password
self._cookie = ""
self._log = logging.getLogger("UVCCamera(%s)" % self._host)
self._log = logging.getLogger(f"UVCCamera({self._host})")

def _safe_request(self, *args, **kwargs):
try:
Expand All @@ -54,7 +54,7 @@ def _safe_request(self, *args, **kwargs):
except OSError:
raise CameraConnectError("Unable to contact camera")
except httplib.HTTPException as ex:
raise CameraConnectError("Error connecting to camera: %s" % (str(ex)))
raise CameraConnectError(f"Error connecting to camera: {ex!s}")

def login(self):
resp = self._safe_request("GET", "/")
Expand Down Expand Up @@ -84,16 +84,14 @@ def login(self):
}
resp = self._safe_request("POST", "/login.cgi", data, headers)
if resp.status != 200:
raise CameraAuthError("Failed to login: %s" % resp.reason)
raise CameraAuthError(f"Failed to login: {resp.reason}")

def _cfgwrite(self, setting, value):
headers = {"Cookie": self._cookie}
resp = self._safe_request(
"GET", "/cfgwrite.cgi?%s=%s" % (setting, value), headers=headers
)
self._log.debug(
"Setting %s=%s: %s %s" % (setting, value, resp.status, resp.reason)
"GET", f"/cfgwrite.cgi?{setting}={value}", headers=headers
)
self._log.debug(f"Setting {setting}={value}: {resp.status} {resp.reason}")
return resp.status == 200

def set_led(self, enabled):
Expand All @@ -117,7 +115,7 @@ def get_snapshot(self):
if resp.status in (401, 403, 302):
raise CameraAuthError("Not logged in")
elif resp.status != 200:
raise CameraConnectError("Snapshot failed: %s" % resp.status)
raise CameraConnectError(f"Snapshot failed: {resp.status}")
return resp.read()

def reboot(self):
Expand All @@ -126,15 +124,15 @@ def reboot(self):
if resp.status in (401, 403, 302):
raise CameraAuthError("Not logged in")
elif resp.status != 200:
raise CameraConnectError("Reboot failed: %s" % resp.status)
raise CameraConnectError(f"Reboot failed: {resp.status}")

def get_status(self):
headers = {"Cookie": self._cookie}
resp = self._safe_request("GET", self.status_url, headers=headers)
if resp.status in (401, 403, 302):
raise CameraAuthError("Not logged in")
elif resp.status != 200:
raise CameraConnectError("Status failed: %s" % resp.status)
raise CameraConnectError(f"Status failed: {resp.status}")
return json.loads(resp.read().decode())


Expand All @@ -148,7 +146,7 @@ def login(self):
data = json.dumps({"username": self._username, "password": self._password})
resp = self._safe_request("POST", "/api/1.1/login", data, headers=headers)
if resp.status != 200:
raise CameraAuthError("Failed to login: %s" % resp.reason)
raise CameraAuthError(f"Failed to login: {resp.reason}")
headers = dict(resp.getheaders())
try:
self._cookie = headers["Set-Cookie"]
Expand Down
12 changes: 6 additions & 6 deletions src/uvcclient/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ def do_reboot(client, camera_info):
except camera.CameraConnectError:
print("Failed to connect to camera")
except Exception as e:
print("Failed to reboot: %s" % e)
print(f"Failed to reboot: {e}")


def do_set_password(opts):
Expand Down Expand Up @@ -173,7 +173,7 @@ def main():
if opts.name:
opts.uuid = client.name_to_uuid(opts.name)
if not opts.uuid:
print("`%s' is not a valid name" % opts.name)
print(f"`{opts.name}' is not a valid name")
return

if opts.dump:
Expand All @@ -193,7 +193,7 @@ def main():
elif cam["state"] == "CONNECTED":
status = "online"
else:
status = "unknown:%s" % cam["state"]
status = "unknown:{}".format(cam["state"])
print(
"%s: %-24.24s [%10s] %s" % (cam["uuid"], cam["name"], status, recmode)
)
Expand All @@ -216,7 +216,7 @@ def main():
return r == "none"
elif opts.get_picture_settings:
settings = client.get_picture_settings(opts.uuid)
print(",".join(["%s=%s" % (k, v) for k, v in settings.items()]))
print(",".join([f"{k}={v}" for k, v in settings.items()]))
return 0
elif opts.set_picture_settings:
settings = {}
Expand All @@ -230,11 +230,11 @@ def main():
try:
result = client.set_picture_settings(opts.uuid, settings)
except Invalid as e:
print("Invalid value: %s" % e)
print(f"Invalid value: {e}")
return 1
for k in settings:
if type(result[k])(settings[k]) != result[k]:
print("Rejected: %s" % k)
print(f"Rejected: {k}")
return 0
elif opts.set_led is not None:
camera = client.get_camera(opts.uuid)
Expand Down
39 changes: 19 additions & 20 deletions src/uvcclient/nvr.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,10 +62,10 @@ def __init__(self, host, port, apikey, path="/", ssl=False):
if path != "/":
raise Invalid("Path not supported yet")
self._apikey = apikey
self._log = logging.getLogger("UVC(%s:%s)" % (host, port))
self._log = logging.getLogger(f"UVC({host}:{port})")
self._bootstrap = self._get_bootstrap()
version = ".".join(str(x) for x in self.server_version)
self._log.debug("Server version is %s" % version)
self._log.debug(f"Server version is {version}")

@property
def server_version(self):
Expand Down Expand Up @@ -99,39 +99,39 @@ def _safe_request(self, *args, **kwargs):
except OSError:
raise CameraConnectionError("Unable to contact camera")
except httplib.HTTPException as ex:
raise CameraConnectionError("Error connecting to camera: %s" % (str(ex)))
raise CameraConnectionError(f"Error connecting to camera: {ex!s}")

def _uvc_request(self, *args, **kwargs):
try:
return self._uvc_request_safe(*args, **kwargs)
except OSError:
raise NvrError("Failed to contact NVR")
except httplib.HTTPException as ex:
raise NvrError("Error connecting to camera: %s" % str(ex))
raise NvrError(f"Error connecting to camera: {ex!s}")

def _uvc_request_safe(
self, path, method="GET", data=None, mimetype="application/json"
):
conn = self._get_http_connection()
if "?" in path:
url = "%s&apiKey=%s" % (path, self._apikey)
url = f"{path}&apiKey={self._apikey}"
else:
url = "%s?apiKey=%s" % (path, self._apikey)
url = f"{path}?apiKey={self._apikey}"

headers = {
"Content-Type": mimetype,
"Accept": "application/json, text/javascript, */*; q=0.01",
"Accept-Encoding": "gzip, deflate, sdch",
}
self._log.debug("%s %s headers=%s data=%s" % (method, url, headers, repr(data)))
self._log.debug(f"{method} {url} headers={headers} data={data!r}")
conn.request(method, url, data, headers)
resp = conn.getresponse()
headers = dict(resp.getheaders())
self._log.debug("%s %s Result: %s %s" % (method, url, resp.status, resp.reason))
self._log.debug(f"{method} {url} Result: {resp.status} {resp.reason}")
if resp.status in (401, 403):
raise NotAuthorized("NVR reported authorization failure")
if resp.status / 100 != 2:
raise NvrError("Request failed: %s" % resp.status)
raise NvrError(f"Request failed: {resp.status}")

data = resp.read()
if (
Expand All @@ -146,7 +146,7 @@ def _get_bootstrap(self):

def dump(self, uuid):
"""Dump information for a camera by UUID."""
data = self._uvc_request("/api/2.0/camera/%s" % uuid)
data = self._uvc_request(f"/api/2.0/camera/{uuid}")
pprint.pprint(data)

def set_recordmode(self, uuid, mode, chan=None):
Expand All @@ -158,7 +158,7 @@ def set_recordmode(self, uuid, mode, chan=None):
:param chan: One of the values from CHANNEL_NAMES
:returns: True if successful, False or None otherwise
"""
url = "/api/2.0/camera/%s" % uuid
url = f"/api/2.0/camera/{uuid}"
data = self._uvc_request(url)
settings = data["data"][0]["recordingSettings"]
mode = mode.lower()
Expand All @@ -182,7 +182,7 @@ def set_recordmode(self, uuid, mode, chan=None):
return settings == updated

def get_recordmode(self, uuid):
url = "/api/2.0/camera/%s" % uuid
url = f"/api/2.0/camera/{uuid}"
data = self._uvc_request(url)
recmodes = data["data"][0]["recordingSettings"]
if recmodes["fullTimeRecordEnabled"]:
Expand All @@ -193,33 +193,32 @@ def get_recordmode(self, uuid):
return "none"

def get_picture_settings(self, uuid):
url = "/api/2.0/camera/%s" % uuid
url = f"/api/2.0/camera/{uuid}"
data = self._uvc_request(url)
return data["data"][0]["ispSettings"]

def set_picture_settings(self, uuid, settings):
url = "/api/2.0/camera/%s" % uuid
url = f"/api/2.0/camera/{uuid}"
data = self._uvc_request(url)
for key in settings:
dtype = type(data["data"][0]["ispSettings"][key])
try:
data["data"][0]["ispSettings"][key] = dtype(settings[key])
except ValueError:
raise Invalid(
"Setting `%s' requires %s not %s"
% (key, dtype.__name__, type(settings[key]).__name__)
f"Setting `{key}' requires {dtype.__name__} not {type(settings[key]).__name__}"
)
data = self._uvc_request(url, "PUT", json.dumps(data["data"][0]))
return data["data"][0]["ispSettings"]

def prune_zones(self, uuid):
url = "/api/2.0/camera/%s" % uuid
url = f"/api/2.0/camera/{uuid}"
data = self._uvc_request(url)
data["data"][0]["zones"] = [data["data"][0]["zones"][0]]
self._uvc_request(url, "PUT", json.dumps(data["data"][0]))

def list_zones(self, uuid):
url = "/api/2.0/camera/%s" % uuid
url = f"/api/2.0/camera/{uuid}"
data = self._uvc_request(url)
return data["data"][0]["zones"]

Expand Down Expand Up @@ -258,10 +257,10 @@ def name_to_uuid(self, name):
return cams_by_name.get(name)

def get_camera(self, uuid):
return self._uvc_request("/api/2.0/camera/%s" % uuid)["data"][0]
return self._uvc_request(f"/api/2.0/camera/{uuid}")["data"][0]

def get_snapshot(self, uuid):
url = "/api/2.0/snapshot/camera/%s?force=true&apiKey=%s" % (uuid, self._apikey)
url = f"/api/2.0/snapshot/camera/{uuid}?force=true&apiKey={self._apikey}"
print(url)
resp = self._safe_request("GET", url)
if resp.status != 200:
Expand Down

0 comments on commit fc27a54

Please sign in to comment.