Compare commits
	
		
			5 Commits
		
	
	
		
			8376e26f9c
			...
			4beba1ab24
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 4beba1ab24 | |||
| a16d472028 | |||
| 1605920d65 | |||
| 08d0e70e26 | |||
| c5caa109cd | 
| @@ -12,3 +12,4 @@ from .inkycal_webshot import Webshot | |||||||
| from .inkycal_xkcd import Xkcd | from .inkycal_xkcd import Xkcd | ||||||
| from .inkycal_fullweather import Fullweather | from .inkycal_fullweather import Fullweather | ||||||
| from .inkycal_tindie import Tindie | from .inkycal_tindie import Tindie | ||||||
|  | from .inkycal_vikunja import Vikunja | ||||||
|   | |||||||
| @@ -82,10 +82,11 @@ class Inkyimage: | |||||||
|     @staticmethod |     @staticmethod | ||||||
|     def preview(image): |     def preview(image): | ||||||
|         """Previews an image on gpicview (only works on Rapsbian with Desktop).""" |         """Previews an image on gpicview (only works on Rapsbian with Desktop).""" | ||||||
|         path = "~/temp" |         path = "/root/repos/Inkycal/temp" | ||||||
|         image.save(path + "/temp.png") |         image.save(path + "/temp.png") | ||||||
|         os.system("gpicview " + path + "/temp.png") |         print(f"previewing image at {path}/temp.png") | ||||||
|         os.system("rm " + path + "/temp.png") |         # os.system("gpicview " + path + "/temp.png") | ||||||
|  |         # os.system("rm " + path + "/temp.png") | ||||||
|  |  | ||||||
|     def _image_loaded(self): |     def _image_loaded(self): | ||||||
|         """returns True if image was loaded""" |         """returns True if image was loaded""" | ||||||
|   | |||||||
| @@ -103,6 +103,9 @@ class Todoist(inkycal_module): | |||||||
|         all_active_tasks = self._api.get_tasks() |         all_active_tasks = self._api.get_tasks() | ||||||
|  |  | ||||||
|         logger.debug(f"all_projects: {all_projects}") |         logger.debug(f"all_projects: {all_projects}") | ||||||
|  |         print(f"all_projects: {all_projects}")  | ||||||
|  |         logger.debug(f"all_active_tasks: {all_active_tasks}") | ||||||
|  |         print(f"all_active_tasks: {all_active_tasks}") | ||||||
|  |  | ||||||
|         # Filter entries in all_projects if filter was given |         # Filter entries in all_projects if filter was given | ||||||
|         if self.project_filter: |         if self.project_filter: | ||||||
|   | |||||||
							
								
								
									
										318
									
								
								inkycal/modules/inkycal_vikunja.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										318
									
								
								inkycal/modules/inkycal_vikunja.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,318 @@ | |||||||
|  | """ | ||||||
|  | Inkycal Todoist Module | ||||||
|  | Copyright by aceinnolab | ||||||
|  | """ | ||||||
|  | import arrow | ||||||
|  | import json | ||||||
|  | import logging | ||||||
|  | import requests | ||||||
|  |  | ||||||
|  | from inkycal.modules.template import inkycal_module | ||||||
|  | from inkycal.custom import * | ||||||
|  |  | ||||||
|  | from todoist_api_python.api import TodoistAPI | ||||||
|  |  | ||||||
|  | logger = logging.getLogger(__name__) | ||||||
|  |  | ||||||
|  | class LoginVikunja(): | ||||||
|  |     def __init__(self, username, password, totp_passcode=None, token=None, api_url='http://192.168.50.10:3456/api/v1/'): | ||||||
|  |         self.username = username | ||||||
|  |         self.password = password | ||||||
|  |         self.totp_passcode = totp_passcode | ||||||
|  |         self.token = None | ||||||
|  |         self.api_url = api_url | ||||||
|  |         self._access_token = token | ||||||
|  |         if self._access_token is None: | ||||||
|  |             self._access_token = self.get_token() | ||||||
|  |      | ||||||
|  |     def _create_url(self, path): | ||||||
|  |         return self.api_url + path | ||||||
|  |  | ||||||
|  |     """returns the token from the login request""" | ||||||
|  |     def _post_login_request(self, username, password, totp_passcode): | ||||||
|  |         login_url = self._create_url('login') | ||||||
|  |         payload = { | ||||||
|  |             'long_token': True, | ||||||
|  |             'username': username, | ||||||
|  |             'password': password, | ||||||
|  |             'totp_passcode': totp_passcode | ||||||
|  |         } | ||||||
|  |         return requests.post(login_url, json=payload, timeout=5) | ||||||
|  |      | ||||||
|  |     def _get_access_token(self): | ||||||
|  |         if not self._access_token: | ||||||
|  |             token_json = self._post_login_request(self.username, self.password, self.totp_passcode) | ||||||
|  |             if token_json.status_code == 200: | ||||||
|  |                 token = json.loads(token_json.text) | ||||||
|  |                 self._access_token = token['token'] | ||||||
|  |             else: | ||||||
|  |                 raise Exception('Login failed') | ||||||
|  |         return self._access_token | ||||||
|  |      | ||||||
|  |     def get_token(self): | ||||||
|  |         return self._get_access_token() | ||||||
|  |      | ||||||
|  |     def get_headers(self): | ||||||
|  |         return {'Authorization': 'Bearer ' + self._get_access_token()} | ||||||
|  |  | ||||||
|  | class ApiVikunja(): | ||||||
|  |     def __init__(self, username, password, totp_passcode=None, token=None, api_url='http://192.168.50.10:3456/api/v1/'): | ||||||
|  |         self.username = username | ||||||
|  |         self.password = password | ||||||
|  |         self.totp_passcode = totp_passcode | ||||||
|  |         self.token = None | ||||||
|  |         self.api_url = api_url | ||||||
|  |         self._cache = {'projects': None, 'tasks': None, 'labels': None} | ||||||
|  |         self._login = LoginVikunja(username, password, totp_passcode, token, api_url) | ||||||
|  |     def _create_url(self, path): | ||||||
|  |         return self.api_url + path | ||||||
|  |  | ||||||
|  |     def _to_json(self, response): | ||||||
|  |         try: | ||||||
|  |             return response.json() | ||||||
|  |         except Exception as e: | ||||||
|  |             logger.error(f'Error parsing json: {e}') | ||||||
|  |             raise e | ||||||
|  |  | ||||||
|  |     def _get_json(self, url, params=None, headers=None): | ||||||
|  |         if params is None: | ||||||
|  |             params = {} | ||||||
|  |         response = requests.get(url, params=params, headers=headers, timeout=5) | ||||||
|  |         response.raise_for_status() | ||||||
|  |         json_result = self._to_json(response) | ||||||
|  |         total_pages = int(response.headers.get('x-pagination-total-pages', 1)) | ||||||
|  |         if total_pages > 1: | ||||||
|  |             logger.debug('Trying to get all pages') | ||||||
|  |             for page in range(2, total_pages + 1): | ||||||
|  |                 logger.debug(f'Getting page {page}') | ||||||
|  |                 params.update({'page': page}) | ||||||
|  |                 response = requests.get(url, params=params, headers=headers, timeout=5) | ||||||
|  |                 response.raise_for_status() | ||||||
|  |                 json_result = json_result + self._to_json(response) | ||||||
|  |         return json_result | ||||||
|  |  | ||||||
|  |     def get_projects(self): | ||||||
|  |         if self._cache['projects'] is None: | ||||||
|  |             self._cache['projects'] = self._get_json(self._create_url('projects'), headers=self._login.get_headers()) | ||||||
|  |         return self._cache['projects'] | ||||||
|  |      | ||||||
|  |     def get_tasks(self, exclude_completed=True): | ||||||
|  |         if self._cache['tasks'] is None: | ||||||
|  |             url = self._create_url('tasks/all') | ||||||
|  |             params = {'filter': 'done=false'} if exclude_completed else {} | ||||||
|  |             self._cache['tasks'] = self._get_json(url, params, headers=self._login.get_headers()) or [] | ||||||
|  |         return self._cache['tasks'] | ||||||
|  |  | ||||||
|  |      | ||||||
|  |  | ||||||
|  | class Vikunja(inkycal_module): | ||||||
|  |     """Todoist api class | ||||||
|  |     parses todos from the todoist api. | ||||||
|  |     """ | ||||||
|  |  | ||||||
|  |     name = "Vikunja API - show your todos from Vikunja" | ||||||
|  |  | ||||||
|  |     requires = { | ||||||
|  |         'url-frontend': { | ||||||
|  |             "label": "Please enter your Vikunja URL", | ||||||
|  |         }, | ||||||
|  |         'url-backend': { | ||||||
|  |             "label": "Please enter your Vikunja URL", | ||||||
|  |         }, | ||||||
|  |         'username': { | ||||||
|  |             "label": "Please enter your Vikunja username", | ||||||
|  |         }, | ||||||
|  |         'password': { | ||||||
|  |             "label": "Please enter your Vikunja password", | ||||||
|  |         }, | ||||||
|  |     } | ||||||
|  |  | ||||||
|  |     optional = { | ||||||
|  |         'project_filter': { | ||||||
|  |             "label": "Show Todos only from following project (separated by a comma). Leave empty to show " + | ||||||
|  |                      "todos from all projects", | ||||||
|  |         } | ||||||
|  |     } | ||||||
|  |  | ||||||
|  |     def __init__(self, config): | ||||||
|  |         """Initialize inkycal_rss module""" | ||||||
|  |  | ||||||
|  |         super().__init__(config) | ||||||
|  |  | ||||||
|  |         config = config['config'] | ||||||
|  |  | ||||||
|  |         # Check if all required parameters are present | ||||||
|  |         for param in self.requires: | ||||||
|  |             if param not in config: | ||||||
|  |                 raise Exception(f'config is missing {param}') | ||||||
|  |  | ||||||
|  |         # module specific parameters | ||||||
|  |         self.frontend_url = config['url-frontend'] | ||||||
|  |         self.backend_url = config['url-backend'] | ||||||
|  |  | ||||||
|  |         # if project filter is set, initialize it | ||||||
|  |         if config['project_filter'] and isinstance(config['project_filter'], str): | ||||||
|  |             self.project_filter = config['project_filter'].split(',') | ||||||
|  |         else: | ||||||
|  |             self.project_filter = config['project_filter'] | ||||||
|  |  | ||||||
|  |         # self._api = TodoistAPI(config['api_key']) | ||||||
|  |         self._vikunja_api = ApiVikunja(config['username'], config['password'], None, None, config['url-backend']) | ||||||
|  |  | ||||||
|  |         # give an OK message | ||||||
|  |         logger.debug(f'{__name__} loaded') | ||||||
|  |  | ||||||
|  |     def _validate(self): | ||||||
|  |         """Validate module-specific parameters""" | ||||||
|  |         if not isinstance(self.api_key, str): | ||||||
|  |             print('api_key has to be a string: "Yourtopsecretkey123" ') | ||||||
|  |      | ||||||
|  |     def get_projects(): | ||||||
|  |  | ||||||
|  |         pass | ||||||
|  |  | ||||||
|  |     def generate_image(self): | ||||||
|  |         """Generate image for this module""" | ||||||
|  |  | ||||||
|  |         # Define new image size with respect to padding | ||||||
|  |         im_width = int(self.width - (2 * self.padding_left)) | ||||||
|  |         im_height = int(self.height - (2 * self.padding_top)) | ||||||
|  |         im_size = im_width, im_height | ||||||
|  |         logger.debug(f'Image size: {im_size}') | ||||||
|  |  | ||||||
|  |         # Create an image for black pixels and one for coloured pixels | ||||||
|  |         im_black = Image.new('RGB', size=im_size, color='white') | ||||||
|  |         im_colour = Image.new('RGB', size=im_size, color='white') | ||||||
|  |  | ||||||
|  |         # Check if internet is available | ||||||
|  |         if internet_available(): | ||||||
|  |             logger.info('Connection test passed') | ||||||
|  |         else: | ||||||
|  |             logger.error("Network not reachable. Please check your connection.") | ||||||
|  |             raise NetworkNotReachableError | ||||||
|  |  | ||||||
|  |         # Set some parameters for formatting todos | ||||||
|  |         line_spacing = 1 | ||||||
|  |         text_bbox_height = self.font.getbbox("hg") | ||||||
|  |         line_height = text_bbox_height[3] + line_spacing | ||||||
|  |         line_width = im_width | ||||||
|  |         max_lines = im_height // line_height | ||||||
|  |  | ||||||
|  |         # Calculate padding from top so the lines look centralised | ||||||
|  |         spacing_top = int(im_height % line_height / 2) | ||||||
|  |  | ||||||
|  |         # Calculate line_positions | ||||||
|  |         line_positions = [ | ||||||
|  |             (0, spacing_top + _ * line_height) for _ in range(max_lines)] | ||||||
|  |  | ||||||
|  |         # Get all projects by name and id | ||||||
|  |         # all_projects = self._api.get_projects() | ||||||
|  |         # filtered_project_ids_and_names = {project.id: project.name for project in all_projects} | ||||||
|  |         # all_active_tasks = self._api.get_tasks() | ||||||
|  |         all_projects = self._vikunja_api.get_projects() | ||||||
|  |         all_active_tasks = self._vikunja_api.get_tasks() | ||||||
|  |         all_active_tasks = [task for task in all_active_tasks if task['done'] == False] | ||||||
|  |  | ||||||
|  |         logger.debug(f"all_projects: {all_projects}") | ||||||
|  |         logger.debug(f"all_active_tasks: {all_active_tasks}") | ||||||
|  |         print(f"all_projects: {all_projects}") | ||||||
|  |         print(f"all_active_tasks: {all_active_tasks}") | ||||||
|  |  | ||||||
|  |         # Filter entries in all_projects if filter was given | ||||||
|  |         if self.project_filter: | ||||||
|  |             # filtered_projects = [project for project in all_projects if project.name in self.project_filter] | ||||||
|  |             filtered_projects = [project for project in all_projects if project['title'] in self.project_filter] | ||||||
|  |             filtered_project_ids_and_names = {project['id']: project['title'] for project in filtered_projects} | ||||||
|  |             filtered_project_ids = [project for project in filtered_project_ids_and_names] | ||||||
|  |             logger.debug(f"filtered projects: {filtered_projects}") | ||||||
|  |             print(f"filtered projects: {filtered_projects}") | ||||||
|  |             print(f"filtered_project_ids_and_names: {filtered_project_ids_and_names}") | ||||||
|  |             print(f"filtered_project_ids: {filtered_project_ids}") | ||||||
|  |  | ||||||
|  |             # If filter was activated and no project was found with that name, | ||||||
|  |             # raise an exception to avoid showing a blank image | ||||||
|  |             if not filtered_projects: | ||||||
|  |                 logger.error('No project found from project filter!') | ||||||
|  |                 logger.error('Please double check spellings in project_filter') | ||||||
|  |                 raise Exception('No matching project found in filter. Please ' | ||||||
|  |                                 'double check spellings in project_filter or leave' | ||||||
|  |                                 'empty') | ||||||
|  |             # filtered version of all active tasks | ||||||
|  |             all_active_tasks = [task for task in all_active_tasks if task['project_id'] in filtered_project_ids] | ||||||
|  |  | ||||||
|  |         # Simplify the tasks for faster processing | ||||||
|  |         simplified = [ | ||||||
|  |             { | ||||||
|  |                 'name': task['title'], | ||||||
|  |                 'due': arrow.get(task['due_date']).format("D-MMM-YY") if 'due_date' in task and task['due_date'][:2] != '00' else "", | ||||||
|  |                 'priority': task['priority'], | ||||||
|  |                 'project': filtered_project_ids_and_names[task['project_id']] | ||||||
|  |             } | ||||||
|  |             for task in all_active_tasks | ||||||
|  |         ] | ||||||
|  |  | ||||||
|  |         logger.debug(f'simplified: {simplified}') | ||||||
|  |         print(f'simplified: {simplified}') | ||||||
|  |  | ||||||
|  |         project_lengths = [] | ||||||
|  |         due_lengths = [] | ||||||
|  |  | ||||||
|  |         for task in simplified: | ||||||
|  |             if task["project"]: | ||||||
|  |                 project_lengths.append(int(self.font.getlength(task['project']) * 1.1)) | ||||||
|  |             if task["due"]: | ||||||
|  |                 due_lengths.append(int(self.font.getlength(task['due']) * 1.1)) | ||||||
|  |  | ||||||
|  |         # Get maximum width of project names for selected font | ||||||
|  |         project_offset = int(max(project_lengths)) if project_lengths else 0 | ||||||
|  |  | ||||||
|  |         # Get maximum width of project dues for selected font | ||||||
|  |         due_offset = int(max(due_lengths)) if due_lengths else 0 | ||||||
|  |  | ||||||
|  |         # create a dict with names of filtered groups | ||||||
|  |         groups = {group_name:[] for group_name in filtered_project_ids_and_names.values()} | ||||||
|  |         for task in simplified: | ||||||
|  |             group_of_current_task = task["project"] | ||||||
|  |             if group_of_current_task in groups: | ||||||
|  |                 groups[group_of_current_task].append(task) | ||||||
|  |  | ||||||
|  |         logger.debug(f"grouped: {groups}") | ||||||
|  |  | ||||||
|  |         # Add the parsed todos on the image | ||||||
|  |         cursor = 0 | ||||||
|  |         for name, todos in groups.items(): | ||||||
|  |             if todos: | ||||||
|  |                 for todo in todos: | ||||||
|  |                     if cursor < max_lines: | ||||||
|  |                         line_x, line_y = line_positions[cursor] | ||||||
|  |  | ||||||
|  |                         if todo['project']: | ||||||
|  |                             # Add todos project name | ||||||
|  |                             write( | ||||||
|  |                                 im_colour, line_positions[cursor], | ||||||
|  |                                 (project_offset, line_height), | ||||||
|  |                                 todo['project'], font=self.font, alignment='left') | ||||||
|  |  | ||||||
|  |                         # Add todos due if not empty | ||||||
|  |                         if todo['due']: | ||||||
|  |                             write( | ||||||
|  |                                 im_black, | ||||||
|  |                                 (line_x + project_offset, line_y), | ||||||
|  |                                 (due_offset, line_height), | ||||||
|  |                                 todo['due'], font=self.font, alignment='left') | ||||||
|  |  | ||||||
|  |                         if todo['name']: | ||||||
|  |                             # Add todos name | ||||||
|  |                             write( | ||||||
|  |                                 im_black, | ||||||
|  |                                 (line_x + project_offset + due_offset, line_y), | ||||||
|  |                                 (im_width - project_offset - due_offset, line_height), | ||||||
|  |                                 todo['name'], font=self.font, alignment='left') | ||||||
|  |  | ||||||
|  |                         cursor += 1 | ||||||
|  |                     else: | ||||||
|  |                         logger.error('More todos than available lines') | ||||||
|  |                         break | ||||||
|  |  | ||||||
|  |         # return the images ready for the display | ||||||
|  |         return im_black, im_colour | ||||||
							
								
								
									
										67
									
								
								tests/test_inkycal_vikunja.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										67
									
								
								tests/test_inkycal_vikunja.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,67 @@ | |||||||
|  | import requests | ||||||
|  | import json | ||||||
|  | from inkycal.modules.inkycal_vikunja import LoginVikunja | ||||||
|  | from inkycal.modules.inkycal_vikunja import ApiVikunja | ||||||
|  | from inkycal.modules.inkycal_vikunja import Vikunja | ||||||
|  | from inkycal.modules.inky_image import Inkyimage | ||||||
|  | import unittest | ||||||
|  | from tests import Config | ||||||
|  | preview = Inkyimage.preview | ||||||
|  | merge = Inkyimage.merge | ||||||
|  |  | ||||||
|  | class TestLoginVikunja(unittest.TestCase): | ||||||
|  |     def setUp(self) -> None: | ||||||
|  |         self.api_url = 'http://192.168.50.10:3456/api/v1/' | ||||||
|  |         self.username = 'iicd' | ||||||
|  |         self.password = '9297519Mhz.' | ||||||
|  |         self.totp_passcode = None | ||||||
|  |     def test_post_login_request(self): | ||||||
|  |         login = LoginVikunja(self.username, self.password, self.totp_passcode, self.api_url) | ||||||
|  |         token_json = login._post_login_request(self.username, self.password, self.totp_passcode) | ||||||
|  |         self.assertTrue(token_json.status_code == 200) | ||||||
|  |  | ||||||
|  | class TestApiVikunja(unittest.TestCase): | ||||||
|  |     def setUp(self) -> None: | ||||||
|  |         self.api_url = 'http://192.168.50.10:3456/api/v1/' | ||||||
|  |         self.username = 'iicd' | ||||||
|  |         self.password = '9297519Mhz.' | ||||||
|  |         self.totp_passcode = None | ||||||
|  |         self.api = ApiVikunja(self.username, self.password, self.totp_passcode,  None, self.api_url) | ||||||
|  |          | ||||||
|  |     def test_get_projects(self): | ||||||
|  |         json_projects = self.api.get_projects() | ||||||
|  |         # print(json.dumps(json_projects, indent=4)) | ||||||
|  |         self.assertTrue(json_projects) | ||||||
|  |      | ||||||
|  |     def test_get_tasks(self): | ||||||
|  |         json_tasks = self.api.get_tasks(exclude_completed=True)   | ||||||
|  |         print(json.dumps(json_tasks, indent=4)) | ||||||
|  |         self.assertTrue(json_tasks) | ||||||
|  |      | ||||||
|  | tests = [ | ||||||
|  |     { | ||||||
|  |         "name": "Vikunja", | ||||||
|  |         "config": { | ||||||
|  |             "size": [400, 1000], | ||||||
|  |             "url-frontend": "http://ff.mhrooz.xyz:8077/", | ||||||
|  |             "url-backend": "http://192.168.50.10:3456/api/v1/", | ||||||
|  |             "username": "iicd", | ||||||
|  |             "password": "9297519Mhz.", | ||||||
|  |             "project_filter": ["LMU", "Master Thesis"], | ||||||
|  |             "padding_x": 10, | ||||||
|  |             "padding_y": 10, | ||||||
|  |             "fontsize": 12, | ||||||
|  |             "language": "en" | ||||||
|  |         } | ||||||
|  |     }, | ||||||
|  | ] | ||||||
|  |  | ||||||
|  | class TestVikunja(unittest.TestCase): | ||||||
|  |     def test_generate_image(self): | ||||||
|  |         for test in tests: | ||||||
|  |             print(f'test {tests.index(test) + 1} generating image..') | ||||||
|  |             module = Vikunja(test) | ||||||
|  |             im_black, im_colour = module.generate_image() | ||||||
|  |             print('OK') | ||||||
|  |             if Config.USE_PREVIEW: | ||||||
|  |                 preview(merge(im_black, im_colour)) | ||||||
		Reference in New Issue
	
	Block a user