rewrite

.gitignore

@@ -2,3 +2,4 @@
 *.dat
 *.dir
 *.pyc
+*.db

analyze.py

@@ -1,69 +0,0 @@
-import argparse
-from collections import Counter, defaultdict
-import shelve
-from typing import Counter, Dict, List, Optional
-
-import simfile
-
-from smoketest import SimfileError
-
-
-def analyze(
-    simfile_errors: Dict[str, List[SimfileError]],
-    *,
-    traceback_substring: str,
-    verbose: bool,
-    number: Optional[int]
-) -> None:
-    error_counter = Counter()
-    error_occurrences = defaultdict(list)
-    for filename, errors in simfile_errors.items():
-        for error in errors:
-            line = error.traceback.splitlines()[-1]
-            if not traceback_substring or traceback_substring in line:
-                error_counter[line] += 1
-                error_occurrences[line].append((filename, error.context))
-    for line, count in error_counter.most_common(number):
-        print(f'=== {line} ({count} occurrences) ===')
-        if verbose:
-            for filename, context in error_occurrences[line]:
-                print(f'  * "{filename}" (during {repr(context)})')
-            print()
-
-
-def main():
-    argparser = argparse.ArgumentParser()
-    argparser.add_argument(
-        '-s', '--shelf',
-        help='shelf file to use',
-        default=f'shelf-{simfile.__version__}',
-    )
-    argparser.add_argument(
-        '-t', '--traceback-substring',
-        help='filter tracebacks by substring',
-    )
-    argparser.add_argument(
-        '-n', '--number',
-        type=int,
-        default=10,
-        help='number of most common errors to print',
-    )
-    argparser.add_argument(
-        '-v', '--verbose',
-        help='enable verbose output',
-        default=False,
-        action='store_true',
-    )
-    args = argparser.parse_args()
-
-    with shelve.open(args.shelf) as simfile_errors:
-        analyze(
-            simfile_errors,
-            traceback_substring=args.traceback_substring,
-            verbose=args.verbose,
-            number=args.number,
-        )
-
-
-if __name__ == '__main__':
-    main()

main.py

@@ -0,0 +1,19 @@
+from smoketest.args import SmoketestArgs
+from smoketest.runner import SmoketestRun
+from smoketest.storage import DB, MODELS
+
+
+def main():
+    args = SmoketestArgs().parse_args()
+
+    DB.connect()
+    if args.init_db:
+        DB.create_tables(MODELS)
+    if args.songs_dir:
+        SmoketestRun(args).process_songs_dir(args.songs_dir)
+    elif args.pack_dir:
+        SmoketestRun(args).process_pack(args.pack_dir)
+
+
+if __name__ == "__main__":
+    main()

poetry.lock

@@ -93,7 +93,7 @@ python-versions = ">=3.6"
 name = "mypy-extensions"
 version = "0.4.3"
 description = "Experimental type system extensions for programs checked with the mypy typechecker."
-category = "dev"
+category = "main"
 optional = false
 python-versions = "*"
 
@@ -161,6 +161,18 @@ category = "dev"
 optional = false
 python-versions = ">=3.7"
 
+[[package]]
+name = "typed-argument-parser"
+version = "1.7.2"
+description = "Typed Argument Parser"
+category = "main"
+optional = false
+python-versions = "*"
+
+[package.dependencies]
+typing_extensions = ">=3.7.4"
+typing-inspect = ">=0.7.1"
+
 [[package]]
 name = "typed-ast"
 version = "1.5.4"
@@ -173,10 +185,22 @@ python-versions = ">=3.6"
 name = "typing-extensions"
 version = "4.3.0"
 description = "Backported and Experimental Type Hints for Python 3.7+"
-category = "dev"
+category = "main"
 optional = false
 python-versions = ">=3.7"
 
+[[package]]
+name = "typing-inspect"
+version = "0.7.1"
+description = "Runtime inspection utilities for typing module."
+category = "main"
+optional = false
+python-versions = "*"
+
+[package.dependencies]
+mypy-extensions = ">=0.3.0"
+typing-extensions = ">=3.7.4"
+
 [[package]]
 name = "zipp"
 version = "3.8.1"
@@ -192,7 +216,7 @@ testing = ["pytest (>=6)", "pytest-checkdocs (>=2.4)", "pytest-flake8", "pytest-
 [metadata]
 lock-version = "1.1"
 python-versions = "^3.7"
-content-hash = "4fdaaca625a1242b7a0012eb7261349b6f25459688543e12b9d4c37eed4f8e3a"
+content-hash = "86790b53efa0ea295271084a2e0fc85fd169fee4b848238c5ac8d7b71ded07d7"
 
 [metadata.files]
 appdirs = []
@@ -210,6 +234,8 @@ semver = []
 simfile = []
 six = []
 tomli = []
+typed-argument-parser = []
 typed-ast = []
 typing-extensions = []
+typing-inspect = []
 zipp = []

pyproject.toml

@@ -10,6 +10,7 @@ python = "^3.7"
 simfile = {version = "^2.1.0b3", allow-prereleases = true}
 peewee = "^3.15.1"
 semver = "^2.13.0"
+typed-argument-parser = "^1.7.2"
 
 [tool.poetry.dev-dependencies]
 black = "^22.6.0"

smoketest.py

@@ -1,130 +0,0 @@
-import argparse
-import os
-import pprint
-import traceback
-from typing import Counter, Dict, List, Optional
-
-from peewee import SqliteDatabase
-import simfile
-from simfile.notes import NoteData
-from simfile.notes.group import group_notes
-from simfile.notes.timed import time_notes
-from simfile.ssc import SSCChart
-from simfile.timing.engine import TimingData
-from simfile.timing.displaybpm import displaybpm
-
-from .storage import *
-
-
-class SimfileError:
-    context: str
-    traceback: Optional[str]
-    
-    def __init__(self, context: str):
-        self.context = context
-        self.traceback = traceback.format_exc()
-    
-    def __repr__(self):
-        return f'{self.context}: {self.traceback}'
-
-def check_simfile(self, filename: str) -> List[SimfileError]:
-    errors = []
-    try:
-        sim = simfile.open(filename)
-    except Exception:
-        errors.append(SimfileError('parse'))
-        return errors
-    
-    timing_data = None
-    try:
-        timing_data = TimingData.from_simfile(sim)
-    except Exception:
-        errors.append(SimfileError('timing'))
-    
-    try:
-        displaybpm(sim)
-    except Exception:
-        errors.append(SimfileError('displaybpm'))
-    
-    for c, chart in enumerate(sim.charts):
-        try:
-            note_data = NoteData.from_chart(chart)
-        except Exception:
-            errors.append(SimfileError(f'chart {c} note_data'))
-        try:
-            for _ in group_notes(note_data, join_heads_to_tails=True):
-                pass
-        except Exception:
-            errors.append(SimfileError(f'chart {c} group_notes'))
-        
-        if isinstance(chart, SSCChart):
-            try:
-                displaybpm(sim, chart)
-            except Exception:
-                errors.append(SimfileError(f'chart {c} displaybpm'))
-        
-        # skip the remaining chart tests if timing data parsing failed
-        if timing_data is None:
-            continue
-        
-        if isinstance(chart, SSCChart):
-            try:
-                timing_data = TimingData.from_simfile(sim, chart)
-            except Exception:
-                errors.append(SimfileError(f'chart {c} timing'))
-        
-        try:
-            for _ in time_notes(note_data, timing_data):
-                pass
-        except Exception:
-            errors.append(SimfileError(f'chart {c} timed_notes'))
-    
-    return errors
-
-
-def scan_songs_dir(self, path: str):
-    stats = Counter()
-    for entry in os.scandir(path):
-        if simfile_errors.get(entry.path) == []:
-            stats['skipped'] += 1
-            continue
-        if entry.is_dir():
-            stats.update(scan_pack(entry.path))
-        elif any(entry.name.endswith(ext) for ext in ('.sm', '.ssc')):
-            errors = check_simfile(entry.path)
-            stats['checked'] += 1
-            simfile_errors[entry.path] = errors
-            if errors:
-                stats['error'] += 1
-                pprint.pprint({'path': entry.path, 'errors': errors})
-            else:
-                stats['success'] += 1
-    
-    return stats
-
-
-
-def dir_path(string):
-    if os.path.isdir(string):
-        return string
-    else:
-        raise ValueError('dir_path: not a directory')
-  
-
-def main():
-    argparser = argparse.ArgumentParser()
-    argparser.add_argument(
-        'd',
-        '--songs-dir',
-        help='directory of packs to scan',
-        type=dir_path,
-    )
-    args = argparser.parse_args()
-
-    scan_songs_dir(args.songs_dir)
-        for k, v in stats.items():
-            print(k, v)
-
-
-if __name__ == '__main__':
-    main()

smoketest/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.1.0"

smoketest/args.py

@@ -0,0 +1,24 @@
+import os
+from typing import Optional
+
+from tap import Tap
+
+
+__all__ = ["SmoketestArgs"]
+
+
+class SmoketestArgs(Tap):
+    songs_dir: Optional[str] = None
+    """directory of packs to scan"""
+
+    pack_dir: Optional[str] = None
+    """single pack directory to scan"""
+
+    new_only: bool = False
+    """only scan newly discovered simfiles"""
+
+    error_only: bool = False
+    """only scan simfiles that have previously errored"""
+
+    init_db: bool = False
+    """initialize the database with tables"""

smoketest/runner.py

@@ -0,0 +1,177 @@
+from contextlib import contextmanager
+import dataclasses
+from datetime import datetime
+import os
+import sys
+import traceback
+from typing import Iterator, Optional
+
+from . import __version__
+import msdparser
+import simfile
+from simfile.dir import SimfilePack
+from simfile.notes import NoteData
+from simfile.notes.group import group_notes, SameBeatNotes
+from simfile.notes.timed import time_notes
+from simfile.ssc import SSCChart
+from simfile.timing.engine import TimingData
+from simfile.timing.displaybpm import displaybpm
+from simfile.types import Simfile, Chart
+
+from .args import SmoketestArgs
+from .storage import *
+
+
+@dataclasses.dataclass(frozen=True)
+class SimfileContext:
+    run: Run
+    path: str
+    kind: Optional[str] = None
+    simfile_title: Optional[str] = None
+    chart_stepstype: Optional[str] = None
+    chart_meter: Optional[str] = None
+    chart_index: Optional[int] = None
+
+    def report_error(self, action: str):
+        print(f"{self.path} ({self.kind}): {traceback.format_exc()}")
+        simfile_object, _ = SimfileObject.get_or_create(
+            path=self.path,
+            kind=self.kind,
+            simfile_title=self.simfile_title,
+            chart_stepstype=self.chart_stepstype,
+            chart_meter=self.chart_meter,
+            chart_index=self.chart_index,
+        )
+        SimfileError.create(
+            action=action,
+            traceback=traceback.format_exc(),
+            simfile_object=simfile_object,
+            run=self.run,
+        )
+
+    @contextmanager
+    def _update(self, **kwargs) -> Iterator["SimfileContext"]:
+        yield dataclasses.replace(self, **kwargs)
+
+    @contextmanager
+    def object(self, kind, **kwargs) -> Iterator["SimfileContext"]:
+        with self._update(kind=kind, **kwargs) as context:
+            SimfileObject.get_or_create(
+                kind=kind,
+                path=context.path,
+                simfile_title=context.simfile_title,
+                chart_stepstype=context.chart_stepstype,
+                chart_meter=context.chart_meter,
+                chart_index=context.chart_index,
+            )
+            yield context
+
+    @contextmanager
+    def perform(self, action: str, **kwargs) -> Iterator["SimfileContext"]:
+        with self._update(**kwargs) as context:
+            try:
+                yield context
+            except Exception:
+                context.report_error(action)
+
+
+class SmoketestRun:
+    args: SmoketestArgs
+    run: Run
+
+    def __init__(self, args: SmoketestArgs):
+        self.args = args
+        self.run = Run.create(
+            created=datetime.now(),
+            simfile_version=simfile.__version__,
+            msdparser_version=msdparser.__version__,
+            smoketest_version=__version__,
+        )
+
+    def process_songs_dir(self, songs_dir: str):
+        for entry in os.scandir(songs_dir):
+            if entry.is_dir():
+                self.process_pack(entry.path)
+
+    def process_pack(self, pack_dir: str):
+        root_context = SimfileContext(run=self.run, path=pack_dir)
+        with root_context.object("SimfilePack", path=pack_dir) as context:
+            print(f"Processing pack {pack_dir}")
+            pack = SimfilePack(pack_dir)
+
+            with context.perform("simfile_dirs") as context:
+                for simfile_dir in pack.simfile_dirs():
+                    if simfile_dir.sm_path:
+                        self._open_simfile(context, simfile_dir.sm_path)
+                    if simfile_dir.ssc_path:
+                        self._open_simfile(context, simfile_dir.ssc_path)
+
+    def _open_simfile(self, context: SimfileContext, simfile_path: str):
+        if self.args.new_only:
+            if SimfileObject.get_or_none(SimfileObject.path == simfile_path):
+                return
+        elif self.args.error_only:
+            if not (
+                SimfileError.select()
+                .join(SimfileObject)
+                .where(SimfileObject.path == simfile_path)  # type: ignore
+                .get_or_none()
+            ):
+                return
+
+        with context.perform("simfile.open", path=simfile_path) as context:
+            sim = simfile.open(simfile_path)
+
+            with context.object(
+                kind=sim.__class__.__name__,
+                simfile_title=sim.title,
+            ) as context:
+                self._test_simfile(context, sim)
+
+    def _test_simfile(self, context: SimfileContext, sim: Simfile):
+        with context.perform("TimingData") as context:
+            TimingData(sim)
+
+        with context.perform("displaybpm") as context:
+            displaybpm(sim)
+
+        with context.perform("charts") as context:
+            for c, chart_ in enumerate(sim.charts):
+                chart: Chart = chart_  # typing workaround
+                with context.object(
+                    chart.__class__.__name__,
+                    chart_stepstype=chart.stepstype,
+                    chart_meter=chart.meter,
+                    chart_index=c,
+                ) as context:
+                    self._test_chart(context, sim, chart)
+
+    def _test_chart(self, context: SimfileContext, sim: Simfile, chart: Chart):
+        td = None
+        if isinstance(chart, SSCChart):
+            with context.perform("TimingData+ssc") as context:
+                td = TimingData(sim, chart)
+            with context.perform("displaybpm+ssc") as context:
+                displaybpm(sim)
+        else:
+            # We've already performed this action. Don't double-report any error
+            try:
+                td = TimingData(sim)
+            except Exception:
+                pass
+
+        with context.perform("NoteData") as context:
+            nd = NoteData(chart)
+
+            with context.perform("group_notes") as context:
+                for _ in group_notes(
+                    iter(nd),
+                    join_heads_to_tails=True,
+                    same_beat_notes=SameBeatNotes.JOIN_ALL,
+                ):
+                    pass
+
+            if td:
+                with context.perform("time_notes") as context:
+                    for _ in time_notes(nd, td):
+                        pass

smoketest/storage.py

@@ -9,10 +9,6 @@ class Run(Model):
     simfile_version = CharField()
     msdparser_version = CharField()
     smoketest_version = CharField()
-    success = IntegerField()
-    error = IntegerField()
-    checked = IntegerField()
-    skipped = IntegerField()
 
     class Meta:
         database = DB
@@ -31,7 +27,7 @@ class SimfileObject(Model):
 
 
 class SimfileError(Model):
-    context = CharField()
+    action = CharField()
     traceback = CharField(max_length=10000)
     simfile_object = ForeignKeyField(model=SimfileObject)
     run = ForeignKeyField(model=Run)
@@ -40,4 +36,4 @@ class SimfileError(Model):
         database = DB
 
 
-DB.connect()
+MODELS = [Run, SimfileObject, SimfileError]