Home-IoT/craftbeerpi4-pione
1
0
Fork 0
mirror of https://github.com/PiBrewing/craftbeerpi4.git synced 2024-11-24 16:08:11 +01:00
Code Issues Projects Releases Packages Wiki Activity

repairs log_file_controller.py.

Browse source 
I seem to have exidentally search and replaced some
needed glob.glob calls when adding dinamic log location,
which should now be repaired.
There also was a mysterious cbpi_dashboard_1.json appearing
in the wrong folder which now inside the dashboard folder.
I also figured out how to run tests locally,
they should now respect the dynamic folder paths.
This commit is contained in:
prash3r 2022-09-24 15:19:34 +02:00
parent e0d809c3a3
commit e7aa0a64c7
7 changed files with 25 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
.vscode/launch.json vendored
View file

@ -28,6 +28,14 @@
"request": "launch", "request": "launch",
"module": "run", "module": "run",
"args": ["--config-folder-path=./.devcontainer/cbpi-dev-config", "setup"] "args": ["--config-folder-path=./.devcontainer/cbpi-dev-config", "setup"]
},
{
"name": "run tests",
"type": "python",
"request": "launch",
"module": "pytest",
"args": ["tests"]
} }
] ]
} }

3
cbpi/configFolder.py
View file

@ -23,6 +23,9 @@ class ConfigFolder:
def get_file_path(self, file): def get_file_path(self, file):
return os.path.join(self.configFolderPath, file) return os.path.join(self.configFolderPath, file)
def get_dashboard_path(self, file):
return os.path.join(self.configFolderPath, "dashboard", file)
def get_upload_file(self, file): def get_upload_file(self, file):
return os.path.join(self.configFolderPath, 'upload', file) return os.path.join(self.configFolderPath, 'upload', file)

10
cbpi/controller/dashboard_controller.py
View file

@ -18,14 +18,14 @@ class DashboardController:
self.logger = logging.getLogger(__name__) self.logger = logging.getLogger(__name__)
self.cbpi.register(self) self.cbpi.register(self)
self.path = cbpi.config_folder.get_file_path("cbpi_dashboard_1.json") self.path = cbpi.config_folder.get_dashboard_path("cbpi_dashboard_1.json")
async def init(self): async def init(self):
pass pass
async def get_content(self, dashboard_id): async def get_content(self, dashboard_id):
try: try:
self.path = self.cbpi.config_folder.get_file_path("cbpi_dashboard_"+ str(dashboard_id) +".json") self.path = self.cbpi.config_folder.get_dashboard_path("cbpi_dashboard_"+ str(dashboard_id) +".json")
logging.info(self.path) logging.info(self.path)
with open(self.path) as json_file: with open(self.path) as json_file:
data = json.load(json_file) data = json.load(json_file)
@ -35,21 +35,21 @@ class DashboardController:
async def add_content(self, dashboard_id, data): async def add_content(self, dashboard_id, data):
print(data) print(data)
self.path = self.cbpi.config_folder.get_file_path("cbpi_dashboard_" + str(dashboard_id)+ ".json") self.path = self.cbpi.config_folder.get_dashboard_path("cbpi_dashboard_" + str(dashboard_id)+ ".json")
with open(self.path, 'w') as outfile: with open(self.path, 'w') as outfile:
json.dump(data, outfile, indent=4, sort_keys=True) json.dump(data, outfile, indent=4, sort_keys=True)
self.cbpi.notify(title="Dashboard {}".format(dashboard_id), message="Saved Successfully", type=NotificationType.SUCCESS) self.cbpi.notify(title="Dashboard {}".format(dashboard_id), message="Saved Successfully", type=NotificationType.SUCCESS)
return {"status": "OK"} return {"status": "OK"}
async def delete_content(self, dashboard_id): async def delete_content(self, dashboard_id):
self.path = self.cbpi.config_folder.get_file_path("cbpi_dashboard_"+ str(dashboard_id)+ ".json") self.path = self.cbpi.config_folder.get_dashboard_path("cbpi_dashboard_"+ str(dashboard_id)+ ".json")
if os.path.exists(self.path): if os.path.exists(self.path):
os.remove(self.path) os.remove(self.path)
self.cbpi.notify(title="Dashboard {}".format(dashboard_id), message="Deleted Successfully", type=NotificationType.SUCCESS) self.cbpi.notify(title="Dashboard {}".format(dashboard_id), message="Deleted Successfully", type=NotificationType.SUCCESS)
async def get_custom_widgets(self): async def get_custom_widgets(self):
path = os.path.join(self.cbpi.config_folder.get_file_path("dashboard"), "widgets") path = self.cbpi.config_folder.get_dashboard_path("widgets")
onlyfiles = [os.path.splitext(f)[0] for f in sorted(listdir(path)) if isfile(join(path, f)) and f.endswith(".svg")] onlyfiles = [os.path.splitext(f)[0] for f in sorted(listdir(path)) if isfile(join(path, f)) and f.endswith(".svg")]
return onlyfiles return onlyfiles

8
cbpi/controller/log_file_controller.py
View file

@ -118,7 +118,7 @@ class LogController:
for name in names: for name in names:
# get all log names # get all log names
all_filenames = os.path.join(self.logsFolderPath, f"sensor_{name}.log*") all_filenames = glob.glob(os.path.join(self.logsFolderPath, f"sensor_{name}.log*"))
# concat all logs # concat all logs
df = pd.concat([pd.read_csv(f, parse_dates=True, date_parser=dateparse, index_col='DateTime', names=['DateTime', name], header=None) for f in all_filenames]) df = pd.concat([pd.read_csv(f, parse_dates=True, date_parser=dateparse, index_col='DateTime', names=['DateTime', name], header=None) for f in all_filenames])
logging.info("Read all files for {}".format(names)) logging.info("Read all files for {}".format(names))
@ -176,7 +176,7 @@ class LogController:
:return: list of log file names :return: list of log file names
''' '''
return [os.path.basename(x) for x in os.path.join(self.logsFolderPath, f"sensor_{name}.log*")] return [os.path.basename(x) for x in glob.glob(os.path.join(self.logsFolderPath, f"sensor_{name}.log*"))]
def clear_log(self, name:str ) -> str: def clear_log(self, name:str ) -> str:
all_filenames = glob.glob(os.path.join(self.logsFolderPath, f"sensor_{name}.log*")) all_filenames = glob.glob(os.path.join(self.logsFolderPath, f"sensor_{name}.log*"))
@ -215,9 +215,9 @@ class LogController:
""" """
formatted_time = strftime("%Y-%m-%d-%H_%M_%S", localtime()) formatted_time = strftime("%Y-%m-%d-%H_%M_%S", localtime())
file_name = os.path.join(self.logsFolderPath, f"{formatted_time}-sensor-{name}.zip" % (formatted_time, name)) file_name = os.path.join(self.logsFolderPath, f"{formatted_time}-sensor-{name}.zip")
zip = zipfile.ZipFile(file_name, 'w', zipfile.ZIP_DEFLATED) zip = zipfile.ZipFile(file_name, 'w', zipfile.ZIP_DEFLATED)
all_filenames = os.path.join(self.logsFolderPath, f"sensor_{name}.log*") all_filenames = glob.glob(os.path.join(self.logsFolderPath, f"sensor_{name}.log*"))
for f in all_filenames: for f in all_filenames:
zip.write(os.path.join(f)) zip.write(os.path.join(f))
zip.close() zip.close()

3
tests/cbpi_config_fixture.py
View file

@ -19,5 +19,6 @@ class CraftBeerPiTestCase(AioHTTPTestCase):
def configuration(self): def configuration(self):
test_directory = os.path.dirname(__file__) test_directory = os.path.dirname(__file__)
test_config_directory = os.path.join(test_directory, 'cbpi-test-config') test_config_directory = os.path.join(test_directory, 'cbpi-test-config')
configFolder = ConfigFolder(test_config_directory) test_logs_directory = os.path.join(test_directory, 'logs')
configFolder = ConfigFolder(test_config_directory, test_logs_directory)
return configFolder return configFolder

2
tests/test_cli.py
View file

@ -9,7 +9,7 @@ logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(
class CLITest(unittest.TestCase): class CLITest(unittest.TestCase):
def test_list(self): def test_list(self):
cli = CraftBeerPiCli(ConfigFolder("./cbpi-test-config")) cli = CraftBeerPiCli(ConfigFolder("./cbpi-test-config", './logs')) # inside tests folder
cli.plugins_list() cli.plugins_list()
if __name__ == '__main__': if __name__ == '__main__':

4
tests/test_logger.py
View file

@ -10,11 +10,11 @@ class LoggerTestCase(CraftBeerPiTestCase):
@unittest_run_loop @unittest_run_loop
async def test_log_data(self): async def test_log_data(self):
os.makedirs("./logs", exist_ok=True) os.makedirs(os.path.join(".", "tests", "logs"), exist_ok=True)
log_name = "test" log_name = "test"
#clear all logs #clear all logs
self.cbpi.log.clear_log(log_name) self.cbpi.log.clear_log(log_name)
assert len(glob.glob('./logs/sensor_%s.log*' % log_name)) == 0 assert len(glob.glob(os.path.join(".", "tests", "logs", f"sensor_{log_name}.log*"))) == 0
# write log entries # write log entries
for i in range(5): for i in range(5):