python: Improve error handling. Stop abusing the assert statement.

scripts/greaseweazle/error.py

@@ -0,0 +1,19 @@
+# greaseweazle/error.py
+#
+# Error management and reporting.
+#
+# Written & released by Keir Fraser <keir.xen@gmail.com>
+#
+# This is free and unencumbered software released into the public domain.
+# See the file COPYING for more details, or visit <http://unlicense.org>.
+
+class Fatal(Exception):
+    pass
+
+def check(pred, desc):
+    if not pred:
+        raise Fatal(desc)
+    
+# Local variables:
+# python-indent: 4
+# End:

scripts/greaseweazle/image/hfe.py

@@ -7,6 +7,7 @@
 
 import struct
 
+from greaseweazle import error
 from greaseweazle.flux import Flux
 from greaseweazle.bitcell import Bitcell
 from bitarray import bitarray
@@ -33,11 +34,11 @@ class HFE:
 
         (sig, f_rev, nr_cyls, nr_sides, t_enc, bitrate,
          _, _, _, tlut_base) = struct.unpack("<8s4B2H2BH", dat[:20])
-        assert sig != b"HXCHFEV3", "HFEv3 is not supported"
-        assert sig == b"HXCPICFE" and f_rev <= 1, "Not a valid HFE file"
-        assert 0 < nr_cyls
-        assert 0 < nr_sides < 3
-        assert bitrate != 0
+        error.check(sig != b"HXCHFEV3", "HFEv3 is not supported")
+        error.check(sig == b"HXCPICFE" and f_rev <= 1, "Not a valid HFE file")
+        error.check(0 < nr_cyls, "HFE: Invalid #cyls")
+        error.check(0 < nr_sides < 3, "HFE: Invalid #sides")
+        error.check(bitrate != 0, "HFE: Invalid bitrate")
         
         hfe = cls(0, nr_sides)
         hfe.bitrate = bitrate

scripts/greaseweazle/image/ipf.py

@@ -10,6 +10,7 @@ import platform
 import ctypes as ct
 from bitarray import bitarray
 from greaseweazle.flux import Flux
+from greaseweazle import error
 
 class CapsDateTimeExt(ct.Structure):
     _pack_ = 1
@@ -139,15 +140,15 @@ class IPF:
         ipf = cls(0, 0)
 
         ipf.iid = ipf.lib.CAPSAddImage()
-        assert ipf.iid >= 0, "Could not create IPF image container"
+        error.check(ipf.iid >= 0, "Could not create IPF image container")
         cname = ct.c_char_p(name.encode())
         res = ipf.lib.CAPSLockImage(ipf.iid, cname)
-        assert res == 0, "Could not open IPF image '%s'" % name
+        error.check(res == 0, "Could not open IPF image '%s'" % name)
         res = ipf.lib.CAPSLoadImage(ipf.iid, DI_LOCK.def_flags)
-        assert res == 0, "Could not load IPF image '%s'" % name
+        error.check(res == 0, "Could not load IPF image '%s'" % name)
         ipf.pi = CapsImageInfo()
         res = ipf.lib.CAPSGetImageInfo(ct.byref(ipf.pi), ipf.iid)
-        assert res == 0
+        error.check(res == 0, "Could not get info for IPF '%s'" % name)
         print(ipf)
 
         return ipf
@@ -163,7 +164,7 @@ class IPF:
         ti = CapsTrackInfoT2(2)
         res = self.lib.CAPSLockTrack(ct.byref(ti), self.iid,
                                      cyl, head, DI_LOCK.def_flags)
-        assert res == 0
+        error.check(res == 0, "Could not lock IPF track %d.%d" % (cyl, head))
 
         if not ti.trackbuf:
             return None # unformatted/empty
@@ -179,16 +180,16 @@ class IPF:
         #    si = CapsSectorInfo()
         #    res = self.lib.CAPSGetInfo(ct.byref(si), self.iid,
         #                               cyl, head, 1, i)
-        #    assert res == 0
+        #    error.check(res == 0, "Couldn't get sector info")
         #    # Data is at trackbuf[si.datastart:si.datastart + si.datasize]
         #for i in range(ti.weakcnt):
         #    wi = CapsDataInfo()
         #    res = self.lib.CAPSGetInfo(ct.byref(wi), self.iid,
         #                               cyl, head, 2, i)
-        #    assert res == 0
+        #    error.check(res == 0, "Couldn't get weak data info")
         #    # Weak data at trackbuf[wi.start:wi.start + wi.size]
 
-        assert ti.weakcnt == 0, "Can't yet handle weak data"
+        error.check(ti.weakcnt == 0, "Can't yet handle weak data")
 
         # We don't really have access to the bitrate. It depends on RPM.
         # So we assume a rotation rate of 300 RPM (5 rev/sec).
@@ -252,16 +253,15 @@ def open_libcaps():
             break
         except:
             pass
-    if "lib" not in locals():
-        print("""\
-** Could not find SPS/CAPS IPF decode library
+    
+    error.check("lib" in locals(), """\
+Could not find SPS/CAPS IPF decode library
 For installation instructions please read the wiki:
 <https://github.com/keirf/Greaseweazle/wiki/IPF-Images>""")
-        sys.exit(1)
     
     # We have opened the library. Now initialise it.
     res = lib.CAPSInit()
-    assert res == 0, "Failure initialising %s" % libname
+    error.check(res == 0, "Failure initialising IPF library '%s'" % name)
 
     return lib
 

scripts/greaseweazle/image/scp.py

@@ -7,6 +7,7 @@
 
 import struct
 
+from greaseweazle import error
 from greaseweazle.flux import Flux
 
 class SCP:
@@ -32,7 +33,7 @@ class SCP:
 
         header = struct.unpack("<3s9BI", dat[0:16])
         (sig, _, _, nr_revs, s_trk, e_trk, flags, _, ss, _, _) = header
-        assert sig == b"SCP"
+        error.check(sig == b"SCP", "SCP: Bad signature")
         nr_sides = 1 if ss else 2
         
         trk_offs = struct.unpack("<168I", dat[16:0x2b0])
@@ -48,8 +49,8 @@ class SCP:
             # Parse the SCP track header and extract the flux data.
             thdr = dat[trk_off:trk_off+4+12*nr_revs]
             sig, tnr, _, _, s_off = struct.unpack("<3sB3I", thdr[:16])
-            assert sig == b"TRK"
-            assert tnr == trknr
+            error.check(sig == b"TRK", "SCP: Missing track signature")
+            error.check(tnr == trknr, "SCP: Wrong track number in header")
             _, e_nr, e_off = struct.unpack("<3I", thdr[-12:])
             tdat = dat[trk_off+s_off:trk_off+e_off+e_nr*2]
 

scripts/greaseweazle/usb.py

@@ -7,6 +7,7 @@
 
 import struct
 from greaseweazle import version
+from greaseweazle import error
 from greaseweazle.flux import Flux
 
 ## Control-Path command set
@@ -167,7 +168,8 @@ class Unit:
     def _send_cmd(self, cmd):
         self.ser.write(cmd)
         (c,r) = struct.unpack("2B", self.ser.read(2))
-        assert c == cmd[0]
+        error.check(c == cmd[0], "Command returned garbage (%02x != %02x)"
+                    % (c, cmd[0]))
         if r != 0:
             raise CmdError(c, r)
 
@@ -260,7 +262,7 @@ class Unit:
                     flux.append(val)
         except StopIteration:
             pass
-        assert flux[-1] == 0
+        error.check(flux[-1] == 0, "Missing terminator on flux read stream")
         return flux[:-1]
 
 

scripts/gw.py

@@ -11,22 +11,23 @@
 
 import sys
 import importlib
+from greaseweazle import error
 
 missing_modules = []
 
 try:
     import bitarray
-except ModuleNotFoundError:
+except ImportError:
     missing_modules.append("bitarray")
     
 try:
     import crcmod
-except ModuleNotFoundError:
+except ImportError:
     missing_modules.append("crcmod")
     
 try:
     import serial.tools.list_ports
-except ModuleNotFoundError:
+except ImportError:
     missing_modules.append("pyserial")
 
 if missing_modules:
@@ -48,9 +49,14 @@ if len(argv) < 2 or argv[1] not in actions:
 
 mod = importlib.import_module('greaseweazle.tools.' + argv[1])
 main = mod.__dict__['main']
-res = main(argv)
-if res is None:
-    res = 0
+try:
+    res = main(argv)
+    if res is None:
+        res = 0
+except error.Fatal as err:
+    print("** FATAL ERROR:")
+    print(err)
+    res = 1
 sys.exit(res)
     
 # Local variables: