Make audio, motion, regular procedures

Author: WyattBlue

auto_editor/analyze.py

@@ -21,13 +21,10 @@
 from auto_editor.lib.data_structs import Sym
 from auto_editor.render.subtitle import SubtitleParser
 from auto_editor.utils.cmdkw import (
-    ParserError,
     Required,
-    parse_with_palet,
     pAttr,
     pAttrs,
 )
-from auto_editor.utils.func import boolop
 from auto_editor.wavfile import read
 
 if TYPE_CHECKING:
@@ -38,7 +35,6 @@
     from numpy.typing import NDArray
 
     from auto_editor.ffwrapper import FileInfo
-    from auto_editor.lib.data_structs import Env
     from auto_editor.output import Ensure
     from auto_editor.utils.bar import Bar
     from auto_editor.utils.log import Log
@@ -412,81 +408,3 @@ def motion(self, s: int, blur: int, width: int) -> NDArray[np.float64]:
 
         self.bar.end()
         return self.cache("motion", mobj, threshold_list[:index])
-
-
-def edit_method(val: str, filesetup: FileSetup, env: Env) -> NDArray[np.bool_]:
-    assert isinstance(filesetup, FileSetup)
-    src = filesetup.src
-    tb = filesetup.tb
-    ensure = filesetup.ensure
-    strict = filesetup.strict
-    bar = filesetup.bar
-    temp = filesetup.temp
-    log = filesetup.log
-
-    if ":" in val:
-        method, attrs = val.split(":", 1)
-    else:
-        method, attrs = val, ""
-
-    levels = Levels(ensure, src, tb, bar, temp, log)
-
-    if method == "none":
-        return levels.none()
-    if method == "all/e":
-        return levels.all()
-
-    try:
-        obj = parse_with_palet(attrs, builder_map[method], env)
-    except ParserError as e:
-        log.error(e)
-
-    try:
-        if method == "audio":
-            s = obj["stream"]
-            if s == "all" or s == Sym("all"):
-                total_list: NDArray[np.bool_] | None = None
-                for s in range(len(src.audios)):
-                    audio_list = to_threshold(levels.audio(s), obj["threshold"])
-                    if total_list is None:
-                        total_list = audio_list
-                    else:
-                        total_list = boolop(total_list, audio_list, np.logical_or)
-
-                if total_list is None:
-                    if strict:
-                        log.error("Input has no audio streams.")
-                    stream_data = levels.all()
-                else:
-                    stream_data = total_list
-            else:
-                assert isinstance(s, int)
-                stream_data = to_threshold(levels.audio(s), obj["threshold"])
-
-            assert isinstance(obj["minclip"], int)
-            assert isinstance(obj["mincut"], int)
-
-            mut_remove_small(stream_data, obj["minclip"], replace=1, with_=0)
-            mut_remove_small(stream_data, obj["mincut"], replace=0, with_=1)
-
-            return stream_data
-
-        if method == "motion":
-            return to_threshold(
-                levels.motion(obj["stream"], obj["blur"], obj["width"]),
-                obj["threshold"],
-            )
-
-        if method == "subtitle":
-            return levels.subtitle(
-                obj["pattern"],
-                obj["stream"],
-                obj["ignore_case"],
-                obj["max_count"],
-            )
-    except LevelError as e:
-        if strict:
-            log.error(e)
-
-        return levels.all()
-    raise ValueError("Unreachable")

auto_editor/lang/palet.py

@@ -18,7 +18,12 @@
 import numpy as np
 from numpy import logical_and, logical_not, logical_or, logical_xor
 
-from auto_editor.analyze import edit_method, mut_remove_large, mut_remove_small
+from auto_editor.analyze import (
+    LevelError,
+    mut_remove_large,
+    mut_remove_small,
+    to_threshold,
+)
 from auto_editor.lib.contracts import *
 from auto_editor.lib.data_structs import *
 from auto_editor.lib.err import MyError
@@ -49,7 +54,6 @@ class ClosingError(MyError):
 LPAREN, RPAREN, LBRAC, RBRAC, LCUR, RCUR, EOF = "(", ")", "[", "]", "{", "}", "EOF"
 VAL, QUOTE, SEC, DB, DOT, VLIT = "VAL", "QUOTE", "SEC", "DB", "DOT", "VLIT"
 SEC_UNITS = ("s", "sec", "secs", "second", "seconds")
-METHODS = ("audio:", "motion:", "subtitle:")
 brac_pairs = {LPAREN: RPAREN, LBRAC: RBRAC, LCUR: RCUR}
 
 str_escape = {
@@ -315,7 +319,6 @@ def get_next_token(self) -> Token:
 
             result = ""
             has_illegal = False
-            is_method = False
 
             def normal() -> bool:
                 return (
@@ -334,21 +337,14 @@ def handle_strings() -> bool:
 
             while normal():
                 result += self.char
-                if (result + ":") in METHODS:
-                    is_method = True
-                    normal = handle_strings
+                # if (result + ":") in METHODS:
+                #     is_method = True
+                #     normal = handle_strings
 
                 if self.char in "'`|\\":
                     has_illegal = True
                 self.advance()
 
-            if is_method:
-                return Token(VAL, Method(result))
-
-            for method in METHODS:
-                if result == method[:-1]:
-                    return Token(VAL, Method(result))
-
             if self.char == ".":  # handle `object.method` syntax
                 self.advance()
                 return Token(DOT, (Sym(result), self.get_next_token()))
@@ -368,16 +364,6 @@ def handle_strings() -> bool:
 ###############################################################################
 
 
-@dataclass(slots=True)
-class Method:
-    val: str
-
-    def __str__(self) -> str:
-        return f"#<method:{self.val}>"
-
-    __repr__ = __str__
-
-
 class Parser:
     def __init__(self, lexer: Lexer):
         self.lexer = lexer
@@ -807,7 +793,7 @@ def __call__(self, *args: Any) -> Any:
 
 
 @dataclass(slots=True)
-class KeywordProc:
+class KeywordUserProc:
     env: Env
     name: str
     parms: list[str]
@@ -952,7 +938,7 @@ def syn_define(env: Env, node: Node) -> None:
                     raise MyError(f"{node[0]}: must be an identifier")
 
         if kw_only:
-            env[n] = KeywordProc(env, n, parms, kparms, body, (len(parms), None))
+            env[n] = KeywordUserProc(env, n, parms, kparms, body, (len(parms), None))
         else:
             env[n] = UserProc(env, n, parms, (), body)
         return None
@@ -1481,6 +1467,60 @@ def edit_all() -> np.ndarray:
     return env["@levels"].all()
 
 
+def edit_audio(
+    threshold: float = 0.04, stream: object = "all", mincut: int = 6 , minclip: int = 3
+) -> np.ndarray:
+    if "@levels" not in env or "@filesetup" not in env:
+        raise MyError("Can't use `audio` if there's no input media")
+
+    levels = env["@levels"]
+    src = env["@filesetup"].src
+
+    stream_data: NDArray[np.bool_] | None = None
+    if stream == "all" or stream == Sym("all"):
+        stream_range = range(0, len(src.audios))
+    else:
+        assert isinstance(stream, int)
+        stream_range = range(stream, stream + 1)
+
+    try:
+        for s in stream_range:
+            audio_list = to_threshold(levels.audio(s), threshold)
+            if stream_data is None:
+                stream_data = audio_list
+            else:
+                stream_data = boolop(stream_data, audio_list, np.logical_or)
+    except LevelError as e:
+        raise MyError(e)
+
+    if stream_data is not None:
+        mut_remove_small(stream_data, minclip, replace=1, with_=0)
+        mut_remove_small(stream_data, mincut, replace=0, with_=1)
+
+        return stream_data
+
+    return levels.all()
+
+
+def edit_motion(
+    threshold: float = 0.02,
+    stream: int = 0,
+    blur: int = 9,
+    width: int = 400,
+) -> np.ndarray:
+    if "@levels" not in env:
+        raise MyError("Can't use `motion` if there's no input media")
+
+    return to_threshold(env["@levels"].motion(stream, blur, width), threshold)
+
+
+def edit_subtitle(pattern, stream=0, ignore_case=False, max_count=None):
+    if "@levels" not in env:
+        raise MyError("Can't use `subtitle` if there's no input media")
+
+    return env["@levels"].subtitle(pattern, stream, ignore_case, max_count)
+
+
 def my_eval(env: Env, node: object) -> Any:
     if type(node) is Sym:
         val = env.get(node.val)
@@ -1494,11 +1534,6 @@ def my_eval(env: Env, node: object) -> Any:
             )
         return val
 
-    if isinstance(node, Method):
-        if "@filesetup" not in env:
-            raise MyError("Can't use edit methods if there's no input files")
-        return edit_method(node.val, env["@filesetup"], env)
-
     if type(node) is list:
         return [my_eval(env, item) for item in node]
 
@@ -1530,7 +1565,21 @@ def my_eval(env: Env, node: object) -> Any:
         if type(oper) is Syntax:
             return oper(env, node)
 
-        return oper(*(my_eval(env, c) for c in node[1:]))
+        i = 1
+        args: list[Any] = []
+        kwargs: dict[str, Any] = {}
+        while i < len(node):
+            result = my_eval(env, node[i])
+            if type(result) is Keyword:
+                i += 1
+                if i >= len(node):
+                    raise MyError("todo: write good error message")
+                kwargs[result.val] = my_eval(env, node[i])
+            else:
+                args.append(result)
+            i += 1
+
+        return oper(*args, **kwargs)
 
     return node
 
@@ -1545,6 +1594,16 @@ def my_eval(env: Env, node: object) -> Any:
     # edit procedures
     "none": Proc("none", edit_none, (0, 0)),
     "all/e": Proc("all/e", edit_all, (0, 0)),
+    "audio": Proc("audio", edit_audio, (0, 4),
+        is_threshold, orc(is_nat, Sym("all"), "all"), is_nat,
+        {"threshold": 0, "stream": 1, "minclip": 2, "mincut": 2}
+    ),
+    "motion": Proc("motion", edit_motion, (0, 4),
+        is_threshold, is_nat, is_nat, is_nat1
+    ),
+    "subtitle": Proc("subtitle", edit_subtitle, (1, 4),
+        is_str, is_nat, is_bool, orc(is_nat, is_void)
+    ),
     # syntax
     "lambda": Syntax(syn_lambda),
     "λ": Syntax(syn_lambda),