132 lines
4.4 KiB (Stored with Git LFS)
Python
132 lines
4.4 KiB (Stored with Git LFS)
Python
from gobot import GobotHardware, GobotAddresses
|
|
import logging
|
|
import json
|
|
import os
|
|
|
|
class ScanResult:
|
|
corexy_available: bool
|
|
head_available: bool
|
|
vacuum_available: bool
|
|
|
|
corexy_port: int
|
|
head_port: int
|
|
vacuum_port: int
|
|
|
|
def __init__(self):
|
|
self.corexy_available = False
|
|
self.head_available = False
|
|
self.vacuum_available = False
|
|
|
|
self.corexy_port = 0
|
|
self.head_port = 0
|
|
self.vacuum_port = 0
|
|
|
|
def logOut(self, logger):
|
|
logger.info("")
|
|
logger.info("============ SCAN SUMMARY ============")
|
|
logger.info(f"CoreXY : {str(self.corexy_available).ljust(5)} Port: {self.corexy_port}")
|
|
logger.info(f"Head : {str(self.head_available).ljust(5)} Port: {self.head_port}")
|
|
logger.info(f"Vacum : {str(self.vacuum_available).ljust(5)} Port: {self.vacuum_port}")
|
|
logger.info("Scan completed")
|
|
logger.info("")
|
|
|
|
def applyToInterface(self, gobot: GobotHardware):
|
|
if self.corexy_available:
|
|
gobot.setPortMap(GobotAddresses.COREXY, self.corexy_port)
|
|
|
|
if self.head_available:
|
|
gobot.setPortMap(GobotAddresses.HEAD, self.head_port)
|
|
|
|
if self.vacuum_available:
|
|
gobot.setPortMap(GobotAddresses.VACUUM, self.vacuum_port)
|
|
|
|
def __dict__(self):
|
|
return {
|
|
"corexy_available": self.corexy_available,
|
|
"head_available": self.head_available,
|
|
"vacuum_available": self.vacuum_available,
|
|
"corexy_port": self.corexy_port,
|
|
"head_port": self.head_port,
|
|
"vacuum_port": self.vacuum_port
|
|
}
|
|
|
|
def save_to_cache(self, folder: str = "./cache"):
|
|
with open(f"{folder}/scan_result.json", "w") as f:
|
|
f.write(json.dumps(self.__dict__()))
|
|
|
|
def load_from_cache(self, folder: str = "./cache") -> bool:
|
|
if not os.path.exists(f"{folder}/scan_result.json"):
|
|
return False
|
|
|
|
with open(f"{folder}/scan_result.json", "r") as f:
|
|
data = json.load(f)
|
|
self.corexy_available = data["corexy_available"]
|
|
self.head_available = data["head_available"]
|
|
self.vacuum_available = data["vacuum_available"]
|
|
|
|
self.corexy_port = data["corexy_port"]
|
|
self.head_port = data["head_port"]
|
|
self.vacuum_port = data["vacuum_port"]
|
|
return True
|
|
|
|
def perform_scan(gobot: GobotHardware, mainLogger: logging.Logger):
|
|
scan_result = ScanResult()
|
|
|
|
mainLogger.info("============ SCANNING ============")
|
|
user_input = input("Load from cache? y/n: ")
|
|
if user_input.lower() == "y" or user_input == "":
|
|
if scan_result.load_from_cache():
|
|
mainLogger.info("Loaded from cache")
|
|
return scan_result
|
|
else:
|
|
mainLogger.error("No cache found")
|
|
|
|
mainLogger.info("Scanning for nodes")
|
|
|
|
gobot.setPortMap(GobotAddresses.COREXY, 0)
|
|
gobot.setPortMap(GobotAddresses.COREXY, 1)
|
|
gobot.setPortMap(GobotAddresses.COREXY, 2)
|
|
gobot.setPortMap(GobotAddresses.COREXY, 3)
|
|
pkg = gobot.get_info(GobotAddresses.COREXY)
|
|
|
|
if pkg is None:
|
|
mainLogger.error("CoreXY not found")
|
|
scan_result.corexy_available = False
|
|
else:
|
|
scan_result.corexy_available = True
|
|
scan_result.corexy_port = pkg.port
|
|
mainLogger.info("CoreXY Found")
|
|
|
|
gobot.setPortMap(GobotAddresses.HEAD, 0)
|
|
gobot.setPortMap(GobotAddresses.HEAD, 1)
|
|
gobot.setPortMap(GobotAddresses.HEAD, 2)
|
|
gobot.setPortMap(GobotAddresses.HEAD, 3)
|
|
pkg = gobot.get_info(GobotAddresses.HEAD)
|
|
|
|
if pkg is None:
|
|
mainLogger.error("Head not found")
|
|
scan_result.head_available = False
|
|
else:
|
|
scan_result.head_available = True
|
|
scan_result.head_port = pkg.port
|
|
mainLogger.info("Head Found")
|
|
|
|
gobot.setPortMap(GobotAddresses.VACUUM, 0)
|
|
gobot.setPortMap(GobotAddresses.VACUUM, 1)
|
|
gobot.setPortMap(GobotAddresses.VACUUM, 2)
|
|
gobot.setPortMap(GobotAddresses.VACUUM, 3)
|
|
pkg = gobot.get_info(GobotAddresses.VACUUM)
|
|
|
|
if pkg is None:
|
|
mainLogger.error("Vacum not found")
|
|
scan_result.vacuum_available = False
|
|
else:
|
|
scan_result.vacuum_available = True
|
|
scan_result.vacuum_port = pkg.port
|
|
mainLogger.info("Vacum Found")
|
|
|
|
user_input = input("Save to cache? y/n: ")
|
|
if user_input.lower() == "y":
|
|
scan_result.save_to_cache()
|
|
|
|
return scan_result |