Python: various sourcery fixes and suggestions
This commit is contained in:
parent
31c91a246a
commit
8e43977ed7
|
@ -31,6 +31,7 @@ class Display(object):
|
|||
logger.success("planktoscope.display is ready!")
|
||||
except Exception as e:
|
||||
logger.error("Could not detect the display")
|
||||
logger.error(f"Exception was {e}")
|
||||
self.display_available = False
|
||||
|
||||
def display_machine_name(self):
|
||||
|
|
|
@ -181,7 +181,7 @@ class raspimjpeg(object):
|
|||
return status
|
||||
except FileNotFoundError as e:
|
||||
logger.error(
|
||||
f"The status file was not found, make sure the filesystem has not been corrupted"
|
||||
"The status file was not found, make sure the filesystem has not been corrupted"
|
||||
)
|
||||
return ""
|
||||
|
||||
|
@ -489,7 +489,7 @@ class raspimjpeg(object):
|
|||
"""
|
||||
logger.debug(f"Capturing an image to {path}")
|
||||
if path == "":
|
||||
self.__send_command(f"im")
|
||||
self.__send_command("im")
|
||||
else:
|
||||
self.__send_command(f"im {path}")
|
||||
time.sleep(0.1)
|
||||
|
@ -500,7 +500,7 @@ class raspimjpeg(object):
|
|||
def stop(self):
|
||||
"""Halt and release the camera."""
|
||||
logger.debug("Releasing the camera now")
|
||||
self.__send_command(f"ru 0")
|
||||
self.__send_command("ru 0")
|
||||
|
||||
def close(self):
|
||||
"""Kill the process."""
|
||||
|
|
|
@ -22,7 +22,7 @@ class ImagerState(object):
|
|||
allowed = []
|
||||
|
||||
def switch(self, state):
|
||||
""" Switch to new state """
|
||||
"""Switch to new state"""
|
||||
if state.name in self.allowed:
|
||||
logger.info(f"Current:{self} => switched to new state {state.name}")
|
||||
self.__class__ = state
|
||||
|
@ -39,33 +39,33 @@ class Stop(ImagerState):
|
|||
|
||||
|
||||
class Imaging(ImagerState):
|
||||
""" State of getting ready to start """
|
||||
"""State of getting ready to start"""
|
||||
|
||||
name = "imaging"
|
||||
allowed = ["waiting", "stop"]
|
||||
|
||||
|
||||
class Waiting(ImagerState):
|
||||
""" State of waiting for the pump to finish """
|
||||
"""State of waiting for the pump to finish"""
|
||||
|
||||
name = "waiting"
|
||||
allowed = ["stop", "capture"]
|
||||
|
||||
|
||||
class Capture(ImagerState):
|
||||
""" State of capturing image """
|
||||
"""State of capturing image"""
|
||||
|
||||
name = "capture"
|
||||
allowed = ["stop", "waiting"]
|
||||
|
||||
|
||||
class Imager(object):
|
||||
""" A class representing the imager """
|
||||
"""A class representing the imager"""
|
||||
|
||||
def __init__(self):
|
||||
# State of the imager - default is stop.
|
||||
self.state = Stop()
|
||||
|
||||
def change(self, state):
|
||||
""" Change state """
|
||||
"""Change state"""
|
||||
self.state.switch(state)
|
||||
|
|
|
@ -35,7 +35,7 @@ class StreamingHandler(http.server.BaseHTTPRequestHandler):
|
|||
with open("/dev/shm/mjpeg/cam.jpg", "rb") as jpeg: # nosec
|
||||
frame = jpeg.read()
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"Camera has not been started yet")
|
||||
logger.error("Camera has not been started yet")
|
||||
time.sleep(5)
|
||||
except Exception as e:
|
||||
logger.exception(f"An exception occured {e}")
|
||||
|
|
|
@ -202,4 +202,4 @@ if __name__ == "__main__":
|
|||
exit(error)
|
||||
print("All the files are valid")
|
||||
|
||||
exit(0)
|
||||
exit(0)
|
||||
|
|
|
@ -117,20 +117,20 @@ class MQTT_Client:
|
|||
# When connected, run subscribe()
|
||||
self.client.subscribe(self.topic)
|
||||
|
||||
@logger.catch
|
||||
# Run this function in order to subscribe to all the topics begining by actuator
|
||||
@logger.catch
|
||||
def on_subscribe(self, client, obj, mid, granted_qos):
|
||||
# Print when subscribed
|
||||
# TODO Fix bug when this is called outside of this init function (for example when the imager subscribe to status/pump)
|
||||
logger.success(
|
||||
f"{self.name} subscribed to {self.topic}! - mid:{str(mid)} qos:{str(granted_qos)}"
|
||||
f"{self.name} subscribed to {self.topic}! - mid:{mid} qos:{granted_qos}"
|
||||
)
|
||||
|
||||
# Run this command when Node-RED is sending a message on the subscribed topic
|
||||
@logger.catch
|
||||
def on_message(self, client, userdata, msg):
|
||||
# Print the topic and the message
|
||||
logger.info(f"{self.name}: {msg.topic} {str(msg.qos)} {str(msg.payload)}")
|
||||
logger.info(f"{self.name}: {msg.topic} {msg.qos} {msg.payload}")
|
||||
# Parse the topic to find the command. ex : actuator/pump -> pump
|
||||
# This only removes the top-level topic!
|
||||
self.command = msg.topic.split("/", 1)[1]
|
||||
|
@ -157,7 +157,7 @@ class MQTT_Client:
|
|||
return self.__new_message
|
||||
|
||||
def read_message(self):
|
||||
logger.debug(f"clearing the __new_message flag")
|
||||
logger.debug("clearing the __new_message flag")
|
||||
self.__new_message = False
|
||||
|
||||
@logger.catch
|
||||
|
@ -170,4 +170,4 @@ class MQTT_Client:
|
|||
# This is called if this script is launched directly
|
||||
if __name__ == "__main__":
|
||||
# TODO This should be a test suite for this library
|
||||
pass
|
||||
pass
|
||||
|
|
|
@ -36,7 +36,7 @@ def adaptative_threshold(img):
|
|||
# logger.debug(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss)
|
||||
# logger.debug(time.monotonic() - start)
|
||||
# logger.success(f"Threshold used was {ret}")
|
||||
logger.success(f"Adaptative threshold is done")
|
||||
logger.success("Adaptative threshold is done")
|
||||
return mask
|
||||
|
||||
|
||||
|
@ -62,7 +62,7 @@ def simple_threshold(img):
|
|||
# logger.debug(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss)
|
||||
# logger.debug(time.monotonic() - start)
|
||||
logger.info(f"Threshold value used was {ret}")
|
||||
logger.success(f"Simple threshold is done")
|
||||
logger.success("Simple threshold is done")
|
||||
return mask
|
||||
|
||||
|
||||
|
|
|
@ -35,11 +35,10 @@ def uuidMachineName(machine="", type=1):
|
|||
"""
|
||||
if type == 4:
|
||||
id = str(uuid.uuid4())
|
||||
elif machine == "":
|
||||
id = str(uuid.uuid1())
|
||||
else:
|
||||
if machine == "":
|
||||
id = str(uuid.uuid1())
|
||||
else:
|
||||
id = str(uuid.uuid1(node=int(str(machine)[-12:], 16)))
|
||||
id = str(uuid.uuid1(node=int(str(machine)[-12:], 16)))
|
||||
name = ""
|
||||
x = id.rsplit("-", 1)
|
||||
machine = x[1]
|
||||
|
@ -84,13 +83,13 @@ def uuidMachine(machine="", type=1):
|
|||
str: universally unique id
|
||||
"""
|
||||
if type == 4:
|
||||
id = str(uuid.uuid4())
|
||||
return str(uuid.uuid4())
|
||||
else:
|
||||
if machine == "":
|
||||
id = str(uuid.uuid1())
|
||||
else:
|
||||
id = str(uuid.uuid1(node=int(str(machine)[-12:], 16)))
|
||||
return id
|
||||
return (
|
||||
str(uuid.uuid1())
|
||||
if machine == ""
|
||||
else str(uuid.uuid1(node=int(str(machine)[-12:], 16)))
|
||||
)
|
||||
|
||||
|
||||
def uuidName():
|
||||
|
@ -151,12 +150,12 @@ def getSerial():
|
|||
Returns:
|
||||
str: serial number or MAC address
|
||||
"""
|
||||
if os.path.exists("/sys/firmware/devicetree/base/serial-number"):
|
||||
with open("/sys/firmware/devicetree/base/serial-number", "r") as serial_file:
|
||||
return serial_file.readline().strip("\x00")
|
||||
else:
|
||||
if not os.path.exists("/sys/firmware/devicetree/base/serial-number"):
|
||||
return str(uuid.getnode())
|
||||
|
||||
with open("/sys/firmware/devicetree/base/serial-number", "r") as serial_file:
|
||||
return serial_file.readline().strip("\x00")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Type 4:")
|
||||
|
|
Loading…
Reference in a new issue