style(all): apply normal formatting

hplt-project · May 26, 2024 · 0b04407 · 0b04407
1 parent 191d29f
commit 0b04407
Show file tree

Hide file tree

Showing 44 changed files with 2,788 additions and 2,003 deletions.
diff --git a/dump-parameter-schema.py b/dump-parameter-schema.py
@@ -11,86 +11,88 @@
 
 
 def _argparse_parse_args(self: argparse.ArgumentParser, args=None, namespace=None):
-	"""Creates a json string from the argparser instance"""
-	json_out: Dict[str,Any] = {}
-	json_out["type"] = "bilingual/monolingual" #TODO "monolingual" or "bilingual" but no idea how to determine this automatically
-	json_out["description"] = self.description
-
-	# non-simple type so it doesn't end up in json
-	SWITCH = object()
-	SUBSTITUTE = object()
-
-	# We need to skip [0], as this is the prepended `--help`
-	param_dict: Dict[str,Dict] = {}
-	for argument in self._actions[1:]:
-		# Skip --help and --version
-		if isinstance(argument, (argparse._HelpAction, argparse._VersionAction)):
-			continue
-
-		current_str = {
-			"help": argument.help,
-			"required": argument.required
-		}
-
-		if isinstance(argument, argparse._StoreConstAction):
-			current_str |= {
-				"type": "bool",
-				"default": False
-			}
-		elif type(argument.type) == type:
-			current_str |= {
-				"type": cast(type, argument.type).__name__,
-				"default": argument.default
-			}
-		elif type(argument.type) == argparse.FileType:
-			current_str |= {
-				"type": "str",
-				"default": "-"
-			}
-		elif argument.default is not None:
-			current_str|= {
-				"type": type(argument.default).__name__,
-				"default": argument.default
-			}
-		else:
-			print(f"Unknown type for \"{argument.dest}\": skipped\n{argument!r}\n", file=sys.stderr)
-			continue
-
-
-		# If it is an `--option` type argument
-		if argument.option_strings:
-			current_str |= {
-				SWITCH: argument.option_strings[0], #TODO prefer long names?
-				SUBSTITUTE: argument.option_strings[0].replace('-','').upper()
-			}
-		# or a positional one
-		else:
-			current_str |= {
-				SUBSTITUTE: argument.dest.upper()
-			}
-
-		if argument.choices is not None:
-			current_str |= {
-				"allowed_values": argument.choices
-			}
-
-		# Add to the parameter dict
-		param_dict[current_str[SUBSTITUTE]] = current_str
-
-	json_out["parameters"] = param_dict
-
-	json_out["command"] = script_path
-	for _, value in param_dict.items():
-		if not value["required"]:
-			json_out["command"] += " ${" + value[SUBSTITUTE] + ":+" + value[SWITCH] + (" $" + value[SUBSTITUTE] if value["type"] != "bool" else "") + "}"
-		else:
-			json_out["command"] += (" " + value[SWITCH] if SWITCH in value else "") + " $" + value[SUBSTITUTE]
-
-	json.dump(json_out, sys.stdout, indent=4, skipkeys=True)
-	sys.exit(0)
+    """Creates a json string from the argparser instance"""
+    json_out: Dict[str, Any] = {}
+    json_out["type"] = (
+        "bilingual/monolingual"  # TODO "monolingual" or "bilingual" but no idea how to determine this automatically
+    )
+    json_out["description"] = self.description
+
+    # non-simple type so it doesn't end up in json
+    SWITCH = object()
+    SUBSTITUTE = object()
+
+    # We need to skip [0], as this is the prepended `--help`
+    param_dict: Dict[str, Dict] = {}
+    for argument in self._actions[1:]:
+        # Skip --help and --version
+        if isinstance(argument, (argparse._HelpAction, argparse._VersionAction)):
+            continue
+
+        current_str = {"help": argument.help, "required": argument.required}
+
+        if isinstance(argument, argparse._StoreConstAction):
+            current_str |= {"type": "bool", "default": False}
+        elif type(argument.type) == type:
+            current_str |= {
+                "type": cast(type, argument.type).__name__,
+                "default": argument.default,
+            }
+        elif type(argument.type) == argparse.FileType:
+            current_str |= {"type": "str", "default": "-"}
+        elif argument.default is not None:
+            current_str |= {
+                "type": type(argument.default).__name__,
+                "default": argument.default,
+            }
+        else:
+            print(
+                f'Unknown type for "{argument.dest}": skipped\n{argument!r}\n',
+                file=sys.stderr,
+            )
+            continue
+
+        # If it is an `--option` type argument
+        if argument.option_strings:
+            current_str |= {
+                SWITCH: argument.option_strings[0],  # TODO prefer long names?
+                SUBSTITUTE: argument.option_strings[0].replace("-", "").upper(),
+            }
+        # or a positional one
+        else:
+            current_str |= {SUBSTITUTE: argument.dest.upper()}
+
+        if argument.choices is not None:
+            current_str |= {"allowed_values": argument.choices}
+
+        # Add to the parameter dict
+        param_dict[current_str[SUBSTITUTE]] = current_str
+
+    json_out["parameters"] = param_dict
+
+    json_out["command"] = script_path
+    for _, value in param_dict.items():
+        if not value["required"]:
+            json_out["command"] += (
+                " ${"
+                + value[SUBSTITUTE]
+                + ":+"
+                + value[SWITCH]
+                + (" $" + value[SUBSTITUTE] if value["type"] != "bool" else "")
+                + "}"
+            )
+        else:
+            json_out["command"] += (
+                (" " + value[SWITCH] if SWITCH in value else "")
+                + " $"
+                + value[SUBSTITUTE]
+            )
+
+    json.dump(json_out, sys.stdout, indent=4, skipkeys=True)
+    sys.exit(0)
 
 
 argparse.ArgumentParser.parse_args = _argparse_parse_args
 
 with open(script_path) as fh:
-	exec(fh.read())
+    exec(fh.read())
diff --git a/opuscleaner/_util.py b/opuscleaner/_util.py
@@ -6,13 +6,14 @@
 
 T = TypeVar("T")
 
+
 def none_throws(optional: Optional[T], message: str = "Unexpected `None`") -> T:
     if optional is None:
         raise AssertionError(message)
     return optional
 
 
-def _thread_pool_worker(id:int, exc_queue:SimpleQueue, target, args, kwargs):
+def _thread_pool_worker(id: int, exc_queue: SimpleQueue, target, args, kwargs):
     try:
         target(*args, **kwargs)
         exc_queue.put((id, None))
@@ -26,6 +27,7 @@ class ThreadPool:
     exception as soon as it is received. It is then up to you to stop any other
     threads. This pool will wait for them when it exits the context.
     """
+
     def __init__(self):
         self.threads = {}
         self.queue = SimpleQueue()
@@ -41,7 +43,8 @@ def start(self, func, *args, **kwargs):
                 "args": args,
                 "kwargs": kwargs,
             },
-            name=func.__name__)
+            name=func.__name__,
+        )
         self.threads[thread_id].start()
 
     def join(self):
@@ -54,7 +57,7 @@ def join(self):
 
     def __enter__(self):
         return self
-    
+
     def __exit__(self, *args, **kwargs):
         for thread in self.threads.values():
             thread.join()
@@ -65,17 +68,20 @@ class Cancelled(Exception):
     """Error raised by CancelableQueue's `put()` or `get()` when `cancel()` was
     called.
     """
+
     pass
 
 
-T = TypeVar('T')
+T = TypeVar("T")
+
 
 class CancelableQueue(Generic[T]):
     """SimpleQueue, but when cancel() is called it will release all blocking
     put() and get() calls and raise `Cancelled`. Also much worse performance
     than SimpleQueue so don't use for heavy workloads plz.
     """
-    def __init__(self, capacity:Optional[int]=None):
+
+    def __init__(self, capacity: Optional[int] = None):
         self.capacity = capacity
         self.size = 0
         self.queue = deque()
@@ -87,7 +93,11 @@ def put(self, item: T):
         when `cancel()` was called.
         """
         with self.cv:
-            self.cv.wait_for(lambda: self.cancelled or self.capacity is None or self.size < self.capacity)
+            self.cv.wait_for(
+                lambda: self.cancelled
+                or self.capacity is None
+                or self.size < self.capacity
+            )
             if self.cancelled:
                 raise Cancelled()
             self.queue.append(item)
@@ -106,9 +116,9 @@ def get(self) -> T:
             item = self.queue.popleft()
             self.cv.notify()
         return item
-    
+
     def cancel(self):
         """Makes all calls to `get()` and `put()` raise `Cancelled()`."""
         with self.cv:
             self.cancelled = True
-            self.cv.notify_all()    
+            self.cv.notify_all()
diff --git a/opuscleaner/categories.py b/opuscleaner/categories.py
@@ -8,47 +8,62 @@
 
 
 class Category(BaseModel):
-	name: str
+    name: str
 
-	@validator('name')
-	def name_must_not_be_empty(cls, value:str) -> str:
-		assert len(value.strip()) > 0, 'must not be empty'
-		return value.strip()
+    @validator("name")
+    def name_must_not_be_empty(cls, value: str) -> str:
+        assert len(value.strip()) > 0, "must not be empty"
+        return value.strip()
 
 
 class CategoryMapping(BaseModel):
-	categories: List[Category]
-	mapping: Dict[str,List[str]]
+    categories: List[Category]
+    mapping: Dict[str, List[str]]
 
-	@validator('categories')
-	def categories_must_be_unique(cls, value:List[Category]) -> List[Category]:
-		assert len(set(category.name.strip() for category in value)) == len(value), 'categories must have unique names'
-		return value
+    @validator("categories")
+    def categories_must_be_unique(cls, value: List[Category]) -> List[Category]:
+        assert len(set(category.name.strip() for category in value)) == len(
+            value
+        ), "categories must have unique names"
+        return value
 
-	@validator('mapping')
-	def mapping_must_only_contain_categories(cls, value:Dict[str,List[str]], values: Dict[str,Any], **kwargs) -> Dict[str,List[str]]:
-		assert len(set(value.keys()) - set(category.name.strip() for category in values.get('categories', ''))) == 0, 'mapping must only contain keys that are defined in `categories`'
-		return value
+    @validator("mapping")
+    def mapping_must_only_contain_categories(
+        cls, value: Dict[str, List[str]], values: Dict[str, Any], **kwargs
+    ) -> Dict[str, List[str]]:
+        assert (
+            len(
+                set(value.keys())
+                - set(
+                    category.name.strip() for category in values.get("categories", "")
+                )
+            )
+            == 0
+        ), "mapping must only contain keys that are defined in `categories`"
+        return value
 
 
-def read_categories(fh:TextIO) -> CategoryMapping:
-	return parse_obj_as(CategoryMapping, json.load(fh))
+def read_categories(fh: TextIO) -> CategoryMapping:
+    return parse_obj_as(CategoryMapping, json.load(fh))
 
-def write_categories(mapping:CategoryMapping, fh:TextIO) -> None:
-	json.dump(mapping.dict(), fh, indent=2)
+
+def write_categories(mapping: CategoryMapping, fh: TextIO) -> None:
+    json.dump(mapping.dict(), fh, indent=2)
 
 
 app = FastAPI()
 
-@app.get('/')
+
+@app.get("/")
 def get_mapping() -> CategoryMapping:
-	if os.path.exists(CATEGORIES_PATH):
-		with open(CATEGORIES_PATH, 'r') as fh:
-			return read_categories(fh)
-	else:
-		return CategoryMapping(categories=DEFAULT_CATEGORIES, mapping=dict())
-
-@app.put('/')
-def update_categories(body:CategoryMapping) -> None:
-	with open(CATEGORIES_PATH, 'w') as fh:
-		write_categories(body, fh)
+    if os.path.exists(CATEGORIES_PATH):
+        with open(CATEGORIES_PATH, "r") as fh:
+            return read_categories(fh)
+    else:
+        return CategoryMapping(categories=DEFAULT_CATEGORIES, mapping=dict())
+
+
+@app.put("/")
+def update_categories(body: CategoryMapping) -> None:
+    with open(CATEGORIES_PATH, "w") as fh:
+        write_categories(body, fh)