install notebook deps as optional dependencies

This can allow a user to run fluxvis inside e.g., google colab without
installing any software on their local computer.

fluxvis.ipynb

@@ -0,0 +1,60 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "touched-diana",
+   "metadata": {},
+   "source": [
+    "# Welcome to the Flux Visualization notebook!\n",
+    "\n",
+    "To use this notebook, select \"run all\". After a delay, the \"Upload\" button will appear. Click it and then select the file to upload. After the upload is complete, the visualization will be shown below. You can save an image by right clicking.\n",
+    "\n",
+    "If you need to change the visualization settings, add or modify arguments to the `go()` call (e.g., to set the `tracks=`), then hit ctrl-enter to re-evaluate the cell, and upload again. The initial pre-sets are for Apple II 5.25\" floppies stored in `.a2r` format, showing only full tracks.\n",
+    "\n",
+    "The full source for fluxvis is [on github](https://github.com/adafruit/fluxvis) and can be installed locally [from pypi](https://pypi.org/project/fluxvis/) via `pip install fluxvis`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "synthetic-freedom",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "!pip install 'fluxvis[notebook]'"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "quiet-arcade",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from fluxvis.notebook import go\n",
+    "go()"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.9.2"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

fluxvis.ipynb.license

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2022 Jeff Epler for Adafruit Industries
+#
+# SPDX-License-Identifier: MIT

fluxvis/greaseweazle/image/adf.py

@@ -6,7 +6,7 @@
 # See the file COPYING for more details, or visit <http://unlicense.org>.
 
 from . import error
-import .codec.amiga.amigados as amigados
+from . import codec.amiga.amigados as amigados
 from .img import IMG
 from .image import Image
 

fluxvis/greaseweazle/tools/util.py

@@ -7,9 +7,8 @@
 # This is free and unencumbered software released into the public domain.
 # See the file COPYING for more details, or visit <http://unlicense.org>.
 
-import argparse, os, sys, serial, struct, time, re, platform
+import argparse, os, sys, struct, time, re, platform
 import importlib
-import serial.tools.list_ports
 from collections import OrderedDict
 
 from .. import error
@@ -29,60 +28,6 @@ class CmdlineHelpFormatter(argparse.ArgumentDefaultsHelpFormatter,
         return help + ' (default: %(default)s)'
 
 
-class ArgumentParser(argparse.ArgumentParser):
-    def __init__(self, formatter_class=CmdlineHelpFormatter, *args, **kwargs):
-        return super().__init__(formatter_class=formatter_class,
-                                *args, **kwargs)
-
-speed_desc = """\
-SPEED: Track rotation time specified as:
-  <N>rpm | <N>ms | <N>us | <N>ns | <N>scp | <N>
-"""
-
-tspec_desc = """\
-TSPEC: Colon-separated list of:
-  c=SET               :: Set of cylinders to access
-  h=SET               :: Set of heads (sides) to access
-  step=[0-9]          :: # physical head steps between cylinders
-  hswap               :: Swap physical drive heads
-  h[01].off=[+-][0-9] :: Physical cylkinder offsets per head
-  SET is a comma-separated list of integers and integer ranges
-  eg. 'c=0-7,9-12:h=0-1'
-"""
-
-# Returns time period in seconds (float)
-# Accepts rpm, ms, us, ns, scp. Naked value is assumed rpm.
-def period(arg):
-    m = re.match('(\d*\.\d+|\d+)rpm', arg)
-    if m is not None:
-        return 60 / float(m.group(1))
-    m = re.match('(\d*\.\d+|\d+)ms', arg)
-    if m is not None:
-        return float(m.group(1)) / 1e3
-    m = re.match('(\d*\.\d+|\d+)us', arg)
-    if m is not None:
-        return float(m.group(1)) / 1e6
-    m = re.match('(\d*\.\d+|\d+)ns', arg)
-    if m is not None:
-        return float(m.group(1)) / 1e9
-    m = re.match('(\d*\.\d+|\d+)scp', arg)
-    if m is not None:
-        return float(m.group(1)) / 40e6 # SCP @ 40MHz
-    return 60 / float(arg)
-    
-def drive_letter(letter):
-    types = {
-        'A': (USB.BusType.IBMPC, 0),
-        'B': (USB.BusType.IBMPC, 1),
-        '0': (USB.BusType.Shugart, 0),
-        '1': (USB.BusType.Shugart, 1),
-        '2': (USB.BusType.Shugart, 2),
-        'Q': (USB.BusType.Apple, 0)
-    }
-    if not letter.upper() in types:
-        raise argparse.ArgumentTypeError("invalid drive letter: '%s'" % letter)
-    return types[letter.upper()]
-
 def range_str(l):
     if len(l) == 0:
         return '<none>'
@@ -192,20 +137,6 @@ class TrackSet:
     def __iter__(self):
         return self.TrackIter(self)
 
-def split_opts(seq):
-    """Splits a name from its list of options."""
-    parts = seq.split('::')
-    name, opts = parts[0], dict()
-    for x in map(lambda x: x.split(':'), parts[1:]):
-        for y in x:
-            try:
-                opt, val = y.split('=')
-            except ValueError:
-                opt, val = y, True
-            if opt:
-                opts[opt] = val
-    return name, opts
-
 
 image_types = OrderedDict(
     { '.adf': 'ADF',
@@ -262,141 +193,6 @@ def with_drive_selected(fn, usb, args, *_args, **_kwargs):
 def valid_ser_id(ser_id):
     return ser_id and ser_id.upper().startswith("GW")
 
-def score_port(x, old_port=None):
-    score = 0
-    if x.manufacturer == "Keir Fraser" and x.product == "Greaseweazle":
-        score = 20
-    elif x.vid == 0x1209 and x.pid == 0x4d69:
-        # Our very own properly-assigned PID. Guaranteed to be us.
-        score = 20
-    elif x.vid == 0x1209 and x.pid == 0x0001:
-        # Our old shared Test PID. It's not guaranteed to be us.
-        score = 10
-    if score > 0 and valid_ser_id(x.serial_number):
-        # A valid serial id is a good sign unless this is a reopen, and
-        # the serials don't match!
-        if not old_port or not valid_ser_id(old_port.serial_number):
-            score = 20
-        elif x.serial_number == old_port.serial_number:
-            score = 30
-        else:
-            score = 0
-    if old_port and old_port.location:
-        # If this is a reopen, location field must match. A match is not
-        # sufficient in itself however, as Windows may supply the same
-        # location for multiple USB ports (this may be an interaction with
-        # BitDefender). Hence we do not increase the port's score here.
-        if not x.location or x.location != old_port.location:
-            score = 0
-    return score
-
-def find_port(old_port=None):
-    best_score, best_port = 0, None
-    for x in serial.tools.list_ports.comports():
-        score = score_port(x, old_port)
-        if score > best_score:
-            best_score, best_port = score, x
-    if best_port:
-        return best_port.device
-    raise serial.SerialException('Cannot find the Greaseweazle device')
-
-def port_info(devname):
-    for x in serial.tools.list_ports.comports():
-        if x.device == devname:
-            return x
-    return None
-
-def usb_reopen(usb, is_update):
-    mode = { False: 1, True: 0 }
-    try:
-        usb.switch_fw_mode(mode[is_update])
-    except (serial.SerialException, struct.error):
-        # Mac and Linux raise SerialException ("... returned no data")
-        # Win10 pyserial returns a short read which fails struct.unpack
-        pass
-    usb.ser.close()
-    for i in range(10):
-        time.sleep(0.5)
-        try:
-            devicename = find_port(usb.port_info)
-            new_ser = serial.Serial(devicename)
-        except serial.SerialException:
-            # Device not found
-            pass
-        else:
-            new_usb = USB.Unit(new_ser)
-            new_usb.port_info = port_info(devicename)
-            new_usb.jumperless_update = usb.jumperless_update
-            new_usb.can_mode_switch = usb.can_mode_switch
-            return new_usb
-    raise serial.SerialException('Could not reopen port after mode switch')
-
-
-def print_update_instructions(usb):
-    print("To perform an Update:")
-    if not usb.jumperless_update:
-        print(" - Disconnect from USB")
-        print(" - Install the Update Jumper at pins %s"
-              % ("RXI-TXO" if usb.hw_model != 1 else "DCLK-GND"))
-        print(" - Reconnect to USB")
-    print(" - Run \"gw update\" to download and install latest firmware")
-
-
-def usb_mode_check(usb, is_update):
-
-    if usb.update_mode and not is_update:
-        if usb.can_mode_switch:
-            usb = usb_reopen(usb, is_update)
-            if not usb.update_mode:
-                return usb
-        print("ERROR: Device is in Firmware Update Mode")
-        print(" - The only available action is \"gw update\"")
-        if usb.update_jumpered:
-            print(" - For normal operation disconnect from USB and remove "
-                  "the Update Jumper at pins %s"
-                  % ("RXI-TXO" if usb.hw_model != 1 else "DCLK-GND"))
-        else:
-            print(" - Main firmware is erased: You *must* perform an update!")
-        sys.exit(1)
-
-    if is_update and not usb.update_mode:
-        if usb.can_mode_switch:
-            usb = usb_reopen(usb, is_update)
-            error.check(usb.update_mode, """\
-Device did not change to Firmware Update Mode as requested.
-If the problem persists, install the Update Jumper at pins RXI-TXO.""")
-            return usb
-        print("ERROR: Device is not in Firmware Update Mode")
-        print_update_instructions(usb)
-        sys.exit(1)
-
-    if not usb.update_mode and usb.update_needed:
-        print("ERROR: Device firmware v%u.%u is unsupported"
-              % (usb.major, usb.minor))
-        print_update_instructions(usb)
-        sys.exit(1)
-
-    return usb
-
-
-def usb_open(devicename, is_update=False, mode_check=True):
-
-    if devicename is None:
-        devicename = find_port()
-    
-    usb = USB.Unit(serial.Serial(devicename))
-    usb.port_info = port_info(devicename)
-    is_win7 = (platform.system() == 'Windows' and platform.release() == '7')
-    usb.jumperless_update = ((usb.hw_model, usb.hw_submodel) != (1, 0)
-                             and not is_win7)
-    usb.can_mode_switch = (usb.jumperless_update
-                           and not (usb.update_mode and usb.update_jumpered))
-
-    if mode_check:
-        usb = usb_mode_check(usb, is_update)
-
-    return usb
-    
 
 
 # Local variables:

fluxvis/notebook.py

@@ -0,0 +1,76 @@
+# SPDX-FileCopyrightText: 2022 Jeff Epler for Adafruit Industries
+#
+# SPDX-License-Identifier: MIT
+
+"""Helper function for use in Jupyter Notebook"""
+
+import io
+import os
+import tempfile
+import shutil
+from ipywidgets import FileUpload
+from IPython.display import display
+import matplotlib.pyplot as plt
+from . import open_flux, process
+
+
+def go(
+    side=0,
+    tracks=35,
+    start=0,
+    stride=4,
+    linear=False,
+    slices=800,
+    stacks=3,
+    location=None,
+    diameter=108,
+    resolution=900,
+    oversample=2,
+):  # pylint: disable=invalid-name,too-many-arguments
+    """Main helper function for operation in a notebook"""
+    uploader = FileUpload(accept=".a2r,.scp", multiple=False)
+
+    display(uploader)
+
+    def on_upload_change(_):
+        try:
+            for meta, content in zip(uploader.metadata, uploader.data):
+                print(f"processing {meta['name']} with content of {len(content)} bytes")
+                process_one_flux(meta["name"], content)
+        finally:
+            uploader.metadata.clear()
+            uploader.data.clear()
+            uploader._counter = 0  # pylint: disable=protected-access
+
+    def process_one_flux(filename, content):
+        with io.BytesIO(content) as b, tempfile.NamedTemporaryFile(
+            suffix=os.path.splitext(filename)[1]
+        ) as t:
+            shutil.copyfileobj(b, t)
+            t.flush()
+            flux = open_flux(t.name)
+        density = process(
+            flux,
+            side=side,
+            tracks=tracks,
+            start=start,
+            stride=stride,
+            linear=linear,
+            slices=slices,
+            stacks=stacks,
+            location=location,
+            diameter=diameter,
+            resolution=resolution,
+            oversample=oversample,
+        )
+
+        fig, axis = plt.subplots()
+        fig.set_dpi(96)
+        fig.set_size_inches(density.shape[0] / 96, density.shape[1] / 96)
+        fig.set_frameon(False)
+        fig.set_tight_layout(True)
+        plt.axis("off")
+        axis.imshow(density)
+        plt.show()
+
+    uploader.observe(on_upload_change, names="_counter")

setup.cfg

@@ -12,6 +12,7 @@ long_description_content_type = text/markdown
 url = https://github.com/adafruit/fluxvis
 classifiers =
     Programming Language :: Python :: 3
+    Programming Language :: Python :: 3.7
     Programming Language :: Python :: 3.9
     Programming Language :: Python :: 3.10
     Programming Language :: Python :: Implementation :: CPython
@@ -27,7 +28,7 @@ packages =
     fluxvis.greaseweazle.image
     fluxvis.greaseweazle.codec
     fluxvis.greaseweazle.tools
-python_requires = >=3.9
+python_requires = >=3.7
 install_requires =
     numpy
     scikit_image
@@ -35,3 +36,6 @@ install_requires =
 [options.entry_points]
 console_scripts =
     fluxvis = fluxvis.__main__:main
+
+[options.extras_require]
+    notebook = ipywidgets>=7.4,<8