0.6.1 (#8)

Bug fixes
- A small adjustment to OCR to remove trailing "." characters which occasionally occur
- Better error handling for log deletion/rename
- Added errors for invalid settings
- Update resources to be versioned
- Updated mods resource to include which variant to display

TODO.md

@@ -8,7 +8,8 @@
 - Convert enchants to use resource manager
 - Selected statistics for results
 - Convert resources to a package w/ manager.py, mods.py, trade.py, enchants.py, bases.py
-
+- Make OCR insensitive to c/t using a special character to represent "c or t"
+  - This change requires putting the actual enchant strings in the trie as values
 
 # Major features
 ## Buy list

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "labbie"
-version = "0.6.0"
+version = "0.6.1"
 description = "Lab enchant assistant"
 authors = ["Brandon Norick <b.norick@gmail.com>"]
 

src/labbie/__main__.py

@@ -44,9 +44,16 @@ def main():
     log_path = utils.logs_dir() / 'current_run.log'
     if log_path.is_file():
         prev_log_path = log_path.with_name('prev_run.log')
-        if prev_log_path.is_file():
+        # delete prev log if it exists
+        try:
             prev_log_path.unlink()
-        log_path.rename(prev_log_path)
+        except FileNotFoundError:
+            pass
+        # move log from last run to prev log
+        try:
+            log_path.rename(prev_log_path)
+        except FileExistsError:
+            pass
     log_filter = utils.LogFilter('INFO')
     logger.add(log_path, filter=log_filter, rotation="100 MB", retention=1, mode='w', encoding='utf8')
 

src/labbie/mods.py

@@ -1,6 +1,4 @@
-import collections
 import dataclasses
-from logging import log
 import string
 import functools
 from typing import Dict, List, Optional, Tuple, Union
@@ -46,42 +44,38 @@ def helm_display_mods(self):
 
     def _build_helm_mod_info(
         self,
-        mods: List[List[Tuple[str, List[str], List[Union[int, str]]]]]
+        mods: List[List[Tuple[str, List[str], List[Union[float, int, str]], bool]]]
     ) -> Dict[str, HelmModInfo]:
         result = {}
         for index, mod_variants in enumerate(mods):
             trade_text = None
-            matched_index = None
 
-            for variant_index, (mod_format, slot_patterns, _) in enumerate(mod_variants):
+            for mod_format, slot_patterns, _, _ in mod_variants:
                 if '{' not in mod_format:
                     candidate = mod_format.lower()
                     if candidate in self._trade.text_to_stat_id:
                         trade_text = candidate
-                        matched_index = variant_index
                         break
 
                 slotted = mod_format.format(*slot_patterns).lower()
                 hash_only_slot_patterns = ['#'] * mod_format.count('{')
                 hash_only_slotted = mod_format.format(*hash_only_slot_patterns).lower()
                 if slotted in self._trade.text_to_stat_id:
                     trade_text = slotted
-                    matched_index = variant_index
                     break
                 elif hash_only_slotted in self._trade.text_to_stat_id:
                     trade_text = hash_only_slotted
-                    matched_index = variant_index
                     break
                 else:
                     continue
             else:
                 logger.warning(f'No trade text found for {index=} {mod_variants=}')
 
             trade_stat_id = self._trade.text_to_stat_id.get(trade_text)
-            for variant_index, (mod_format, slot_patterns, values) in enumerate(mod_variants):
+            for mod_format, slot_patterns, values, display in mod_variants:
                 if '{' not in mod_format:
                     result[mod_format] = HelmModInfo(
-                        display=(variant_index == matched_index),
+                        display=display,
                         mod=mod_format,
                         trade_text=trade_text,
                         trade_stat_id=trade_stat_id,
@@ -104,15 +98,13 @@ def _build_helm_mod_info(
                                for pattern, value in zip(slot_patterns, values)]
                 mod = mod_format.format(*slot_values)
                 result[mod] = HelmModInfo(
-                    display=(variant_index == matched_index),
+                    display=display,
                     mod=mod,
                     trade_text=trade_text,
                     trade_stat_id=trade_stat_id,
                     trade_stat_value=trade_stat_value
                 )
 
-        self._fix_krangled_helm_mods(result)
-
         return result
 
     @functools.cached_property
@@ -161,9 +153,3 @@ def get_mod_list_from_ocr_results(self, enchant_list):
                 enchants.append(keys[0])
 
         return enchants
-
-    def _fix_krangled_helm_mods(self, helm_mod_info: Dict[str, HelmModInfo]):
-        helm_mod_info['Fireball Always Ignites'].display = False
-        helm_mod_info['Fireball has +30% chance to Ignite'].display = True
-        helm_mod_info['Burning Arrow Always Ignites'].display = False
-        helm_mod_info['Burning Arrow has +30% chance to Ignite'].display = True

src/labbie/ocr.py

@@ -68,7 +68,7 @@ def parse_image(image, save_path, dilate):
     if save_path:
         Image.fromarray(im_bw).save(save_path / 'full_processed.png')
     enchants = pytesseract.image_to_string(im_bw, config='--psm 12').replace('\x0c', '').replace('’', "'")
-    enchants = [e.strip() for e in enchants.split('\n') if e]
+    enchants = [e.strip().rstrip('.') for e in enchants.split('\n') if e]
     return _fix_krangled_ocr(enchants)
 
 

src/labbie/resources.py

@@ -1,7 +1,9 @@
 import asyncio
+import dataclasses
+import functools
 import gzip
 import pathlib
-from typing import Dict, List, Optional, Tuple, Union
+from typing import ClassVar, Dict, List, Optional, Tuple, Union
 
 import aiohttp
 import injector
@@ -14,8 +16,6 @@
 from labbie import utils
 
 _Constants = constants.Constants
-_PathLike = Union[str, pathlib.Path]
-_CONTAINER_URL = 'https://labbie.blob.core.windows.net/enchants'
 _LOADERS = {}
 
 
@@ -50,70 +50,70 @@ def get_loader(path: pathlib.Path):
     raise ValueError(f'No loader specified for file with suffixes {suffixes}')
 
 
-@injector.singleton
-class ResourceManager:
+@dataclasses.dataclass
+class Resource:
+    _CONTAINER_URL: ClassVar[str] = 'https://labbie.blob.core.windows.net/resources'
 
-    @injector.inject
-    def __init__(self, constants: _Constants, app_state: state.AppState):
-        self._constants = constants
-        self._app_state = app_state
-        self.trade_stats: Dict[str, str] = None
-        self.items: Dict[str, List[Tuple[bool, str, str]]] = None
-        self.mods: Dict[str, Dict[str, List[List[Union[str, int]]]]] = None
-        self._init_task = None
+    version: int
+    path_format: str
 
-    def initialize(self):
-        self._constants.resources_dir.mkdir(parents=True, exist_ok=True)
-        self._init_task = asyncio.create_task(self._get_all_resources())
+    @functools.cached_property
+    def path(self) -> pathlib.Path:
+        return pathlib.Path(self.path_format.format(version=self.version))
 
-    async def _get_all_resources(self):
-        # TODO(bnorick): handle PermissionError / OSError from disk write failures in a reasonable way
+    @functools.cached_property
+    def url(self) -> pathlib.Path:
+        return f'{self._CONTAINER_URL}/{self.path}'
 
-        async with aiohttp.ClientSession() as session:
-            self.trade_stats = await self.load_or_download_resource('trade_stats.json.gz', session=session)
-            self.items = await self.load_or_download_resource('trade_items.json.gz', session=session)
-            self.mods = await self.load_or_download_resource('mods.json.gz', session=session)
-        self._app_state.resources_ready = True
+    @functools.cached_property
+    def name(self) -> pathlib.Path:
+        return self.path.name
 
-    async def resource_needs_update(self, resource_path: _PathLike, session: Optional[aiohttp.ClientSession] = None):
+    def local_path(self, resources_dir: pathlib.Path):
+        return resources_dir / self.name
+
+    def cached_hash_path(self, resources_dir: pathlib.Path):
+        local_path = self.local_path(resources_dir)
+        return local_path.parent / f'{local_path.name}.md5'
+
+    async def needs_update(self, resources_dir: pathlib.Path,
+                           session: Optional[aiohttp.ClientSession] = None):
         async with utils.client_session(session) as session:
-            async with session.head(f'{_CONTAINER_URL}/{resource_path}') as resp:
+            async with session.head(self.url) as resp:
                 if resp.status == 404:
                     return True
-                cached_hash = self.cached_resource_hash(resource_path)
+                cached_hash = self.cached_hash(resources_dir)
                 remote_hash = resp.headers['Content-MD5']
                 needs_update = cached_hash != remote_hash
-                logger.debug(f'{resource_path=} {needs_update=} {cached_hash=}{"==" if not needs_update else "!="}{remote_hash=}')
-
+                logger.debug(f'{self.local_path(resources_dir)} {needs_update=} '
+                             f'{cached_hash=}{"==" if not needs_update else "!="}{remote_hash=}')
                 return needs_update
 
-    def cached_resource_hash(self, resource_path: _PathLike):
-        local_resource_path = self.local_resource_path(resource_path)
-        hash_path = local_resource_path.parent / f'{local_resource_path.name}.md5'
+    def cached_hash(self, resources_dir: pathlib.Path):
+        hash_path = self.cached_hash_path(resources_dir)
 
         if not hash_path.exists():
             return None
 
         with hash_path.open(encoding='utf8') as f:
             return f.read()
 
-    async def load_or_download_resource(self, resource_path: _PathLike, force: bool = False,
-                                        session: Optional[aiohttp.ClientSession] = None):
-        needs_update = await self.resource_needs_update(resource_path, session)
-        if needs_update:
+    async def load_or_download(self, resources_dir: pathlib.Path, force: bool = False,
+                               session: Optional[aiohttp.ClientSession] = None):
+        needs_update = await self.needs_update(resources_dir, session=session)
+        if force or needs_update:
             async with utils.client_session(session) as session:
-                remote_resource = self.remote_resource_path(resource_path)
-                async with session.get(remote_resource) as resp:
+                async with session.get(self.url) as resp:
                     if resp.status != 200:
-                        raise errors.FailedToDownloadResource(remote_resource)
+                        raise errors.FailedToDownloadResource(self.url)
 
                     content = await resp.content.read()
                     hash = resp.headers['Content-MD5']
-                    self.save(resource_path, content, hash)
+                    self.save(resources_dir, content, hash)
 
-        return self.load(resource_path)
+        return self.load(resources_dir)
 
-    def save(self, resource_path: _PathLike, content: Union[str, bytes], hash: str):
+    def save(self, resources_dir: pathlib.Path, content: Union[str, bytes], hash: str):
         kwargs = {}
         if isinstance(content, str):
             mode = 'w'
@@ -123,23 +123,51 @@ def save(self, resource_path: _PathLike, content: Union[str, bytes], hash: str):
         else:
             raise ValueError(f'Invalid content type, expected str or bytes but got {type(content).__name__}')
 
-        local_resource_path = self.local_resource_path(resource_path)
-        with local_resource_path.open(mode, **kwargs) as f:
+        local_path = self.local_path(resources_dir)
+        with local_path.open(mode, **kwargs) as f:
             f.write(content)
 
-        hash_path = local_resource_path.parent / f'{local_resource_path.name}.md5'
+        hash_path = self.cached_hash_path(resources_dir)
         with hash_path.open('w', encoding='utf8') as f:
             f.write(hash)
 
+    def load(self, resources_dir: pathlib.Path):
+        local_path = self.local_path(resources_dir)
+        loader = get_loader(local_path)
+        return loader(local_path)
+
 
-    def load(self, resource_path: _PathLike):
-        local_resource_path = self.local_resource_path(resource_path)
-        loader = get_loader(local_resource_path)
-        return loader(local_resource_path)
+@injector.singleton
+class ResourceManager:
+
+    # NOTE: file names must not collide, as the local path only includes the name and not the full path
+    _RESOURCES = {
+        'trade_stats': Resource(version=1, path_format='pathofexile/{version}/stats.json.gz'),
+        'items': Resource(version=1, path_format='pathofexile/{version}/items.json.gz'),
+        'mods': Resource(version=1, path_format='repoe/{version}/mods.json.gz'),
+    }
+
+    @injector.inject
+    def __init__(self, constants: _Constants, app_state: state.AppState):
+        self._constants = constants
+        self._app_state = app_state
 
-    def remote_resource_path(self, resource_path: _PathLike):
-        return f'{_CONTAINER_URL}/{resource_path}'
+        self._init_task = None
 
-    def local_resource_path(self, resource_path: _PathLike):
-        resource_path = pathlib.Path(resource_path)
-        return self._constants.resources_dir / resource_path
+        # NOTE: the following attributes are set by _get_all_resources
+        self.trade_stats: Dict[str, str] = None
+        self.items: Dict[str, List[Tuple[bool, str, str]]] = None
+        self.mods: Dict[str, List[List[str, List[str], List[Union[float, int, str]], bool]]] = None
+
+    def initialize(self):
+        self._constants.resources_dir.mkdir(parents=True, exist_ok=True)
+        self._init_task = asyncio.create_task(self._get_all_resources())
+
+    async def _get_all_resources(self):
+        # TODO(bnorick): handle potential PermissionError / OSError from disk write failures in a reasonable way
+        async with aiohttp.ClientSession() as session:
+            for name, resource in self._RESOURCES.items():
+                value = await resource.load_or_download(
+                    resources_dir=self._constants.resources_dir, session=session)
+                setattr(self, name, value)
+        self._app_state.resources_ready = True

src/labbie/ui/settings/widget/presenter.py

@@ -6,6 +6,7 @@
 from labbie import config
 from labbie import constants
 from labbie import state
+from labbie.ui import keys
 from labbie.ui.app import presenter as app
 from labbie.ui.settings.widget import view
 from labbie.ui.screen_selection.widget import view as screen_selection
@@ -81,6 +82,25 @@ def on_screen_selection_done(self):
         self._view.bottom = str(bottom + 1)
 
     async def on_save(self, checked):
+        try:
+            left = int(self._view.left)
+            top = int(self._view.top)
+            right = int(self._view.right)
+            bottom = int(self._view.bottom)
+
+            if right <= left:
+                raise ValueError('Left must be less than right')
+
+            if bottom <= top:
+                raise ValueError('Top must be less than bottom')
+        except ValueError as e:
+            err = str(e)
+            if err.startswith('invalid literal'):
+                e = ValueError('Bounds values (left, top, right, bottom) must be integers')
+
+            self._app_presenter.show(keys.ErrorWindowKey(e))
+            return
+
         if self._view.league != self._config.league:
             self._config.league = self._view.league
             if self._config.league:

src/labbie/version.py

@@ -1 +1 @@
-__version__ = '0.6.0'
+__version__ = '0.6.1'