in-progress peewee rewrite

.gitignore

@@ -0,0 +1,4 @@
+*.bak
+*.dat
+*.dir
+*.pyc

smoketest.py

@@ -1,10 +1,10 @@
 import argparse
 import os
 import pprint
-import shelve
 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
@@ -13,122 +13,118 @@ 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}"
+        return f'{self.context}: {self.traceback}'
 
-
-def check_simfile(filename: str) -> List[SimfileError]:
+def check_simfile(self, filename: str) -> List[SimfileError]:
     errors = []
     try:
         sim = simfile.open(filename)
     except Exception:
-        errors.append(SimfileError("parse"))
+        errors.append(SimfileError('parse'))
         return errors
-
+    
     timing_data = None
     try:
         timing_data = TimingData.from_simfile(sim)
     except Exception:
-        errors.append(SimfileError("timing"))
-
+        errors.append(SimfileError('timing'))
+    
     try:
         displaybpm(sim)
     except Exception:
-        errors.append(SimfileError("displaybpm"))
-
+        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"))
+            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"))
-
+            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"))
-
+                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"))
-
+                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"))
-
+            errors.append(SimfileError(f'chart {c} timed_notes'))
+    
     return errors
 
 
-def scan_directory(path: str, simfile_errors: Dict[str, List[SimfileError]]):
+def scan_songs_dir(self, path: str):
     stats = Counter()
     for entry in os.scandir(path):
         if simfile_errors.get(entry.path) == []:
-            stats["skipped"] += 1
+            stats['skipped'] += 1
             continue
         if entry.is_dir():
-            stats.update(scan_directory(entry.path, simfile_errors))
-        elif any(entry.name.endswith(ext) for ext in (".sm", ".ssc")):
+            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
+            stats['checked'] += 1
             simfile_errors[entry.path] = errors
             if errors:
-                stats["error"] += 1
-                pprint.pprint({"path": entry.path, "errors": errors})
+                stats['error'] += 1
+                pprint.pprint({'path': entry.path, 'errors': errors})
             else:
-                stats["success"] += 1
-
+                stats['success'] += 1
+    
     return stats
 
 
+
 def dir_path(string):
     if os.path.isdir(string):
         return string
     else:
-        raise ValueError("dir_path: not a directory")
-
+        raise ValueError('dir_path: not a directory')
+  
 
 def main():
     argparser = argparse.ArgumentParser()
     argparser.add_argument(
-        "directory",
-        help="directory of simfiles to scan",
+        'd',
+        '--songs-dir',
+        help='directory of packs to scan',
         type=dir_path,
     )
-    argparser.add_argument(
-        "-s",
-        "--shelf",
-        help="shelf file to use",
-        default=f"shelf-{simfile.__version__}",
-    )
     args = argparser.parse_args()
 
-    with shelve.open(args.shelf) as simfile_errors:
-        stats = scan_directory(args.directory, simfile_errors)
+    scan_songs_dir(args.songs_dir)
         for k, v in stats.items():
             print(k, v)
 
 
-if __name__ == "__main__":
-    main()
+if __name__ == '__main__':
+    main()

storage.py

@@ -0,0 +1,43 @@
+from peewee import *
+
+
+DB = SqliteDatabase("smoketest.db")
+
+
+class Run(Model):
+    created = DateTimeField()
+    simfile_version = CharField()
+    msdparser_version = CharField()
+    smoketest_version = CharField()
+    success = IntegerField()
+    error = IntegerField()
+    checked = IntegerField()
+    skipped = IntegerField()
+
+    class Meta:
+        database = DB
+
+
+class SimfileObject(Model):
+    kind = CharField()
+    path = CharField()
+    simfile_title = CharField(null=True)
+    chart_stepstype = CharField(null=True)
+    chart_meter = CharField(null=True)
+    chart_index = IntegerField(null=True)
+
+    class Meta:
+        database = DB
+
+
+class SimfileError(Model):
+    context = CharField()
+    traceback = CharField(max_length=10000)
+    simfile_object = ForeignKeyField(model=SimfileObject)
+    run = ForeignKeyField(model=Run)
+
+    class Meta:
+        database = DB
+
+
+DB.connect()