Repository: opendatacube/eo-datasets (language: python)

Issue:
Add Public API for providing destination paths.
Implement a public API to provide destination paths (currently this function is not exposed publicly, but does exist).

This is required for [Statistician](https://github.com/opendatacube/odc-tools/tree/develop/libs/stats) to start using *eo-datasets* for metadata generation.

@Kirill888 
Network Read/Write + USGS Collection 2
The headline features of this PR are:
 - Support for both reading and writing from S3 and other network "filesystems", using [fsspec](https://pypi.org/project/fsspec/).
 - A new prepare command for converting USGS Collection 2 Datasets into COGs with ODC eo3 metadata
 - A Dockerfile for publishing eo-datasets as a Docker image

At the moment there is also a little CLI for pulling messages from AWS SQS and running a command with them, but that should have a better home found for it.


Discussion / hints:

Fix patch:
diff --git a/Makefile b/Makefile
index af11d88d..38dc2d56 100644
--- a/Makefile
+++ b/Makefile
@@ -16,7 +16,14 @@ check:
 
 # Run tests in Docker
 test:
-	docker run --rm --volume "${PWD}":/tests -w /home/runner eodatasets:test pytest --cov eodatasets3 --durations=5 /tests
+	docker run --rm --volume "${PWD}":/tests -w /tests eodatasets:test /tests/run-tests.sh
+
+# Run tests in Docker as container root user
+#
+# (Eg. Github Actions has file write permission issues to external volumes when not using root,
+#      and we want to write test coverage etc. to the volume.)
+test-root:
+	docker run --rm --volume "${PWD}":/tests --user root -w /tests eodatasets:test /tests/run-tests.sh
 
 # Run linters in Docker
 lint:
diff --git a/check-code.sh b/check-code.sh
index 816459ac..bf262c99 100755
--- a/check-code.sh
+++ b/check-code.sh
@@ -1,12 +1,13 @@
 #!/usr/bin/env bash
-# Convenience script for running CI-like checks
+# Convenience script for running all CI-like checks
 
 set -eu
 set -x
 
-pre-commit run -a
+script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )"
+
+cd script_dir
 
-# Run tests, taking coverage.
-# Users can specify extra folders as arguments.
-pytest --cov eodatasets3 --durations=5 . $@
+pre-commit run -a
 
+./run-tests.sh $@
diff --git a/docs/conf.py b/docs/conf.py
index 8926f1bd..4312e347 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -15,6 +15,10 @@
 
 # Show __init__ docstring as the class docstring.
 autosummary_generate = True
+# Warn about broken references
+nitpicky = True
+# Show warnings on page so people want to fix them more.
+keep_warnings = True
 
 extensions = [
     "sphinx.ext.autodoc",
@@ -24,13 +28,16 @@
     "sphinx.ext.todo",
     "sphinx.ext.coverage",
     "sphinx.ext.viewcode",
+    "sphinx.ext.doctest",
 ]
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3/", None),
-    "odc": ("https://datacube-core.readthedocs.io/en/stable/", None),
+    "datacube": ("https://datacube-core.readthedocs.io/en/stable/", None),
     "pandas": ("https://pandas.pydata.org/pandas-docs/stable/", None),
     "numpy": ("https://docs.scipy.org/doc/numpy/", None),
     "xarray": ("https://xarray.pydata.org/en/stable/", None),
+    "rasterio": ("https://rasterio.readthedocs.io/en/latest/", None),
+    "ruamel.yaml": ("https://yaml.readthedocs.io/en/latest/", None),
 }
 templates_path = ["_templates"]
 exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
@@ -38,7 +45,7 @@
 # Custom static files (such as style sheets)
 # They are copied after the builtin static files, so a file named "default.css" will
 # overwrite the builtin "default.css".
-html_static_path = ["_static"]
+# html_static_path = ["_static"]
 
 
 if is_on_readthedocs:
diff --git a/docs/index.rst b/docs/index.rst
index fb1ef90c..96f6b5ee 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -1,24 +1,44 @@
-.. eodatasets3 documentation master file, created by
-   sphinx-quickstart on Mon Aug 26 14:33:29 2019.
-   You can adapt this file completely to your liking, but it should at least
-   contain the root `toctree` directive.
+
+..
+  We avoid using bigger heading types as readthedocs annoyingly collapses all table-of-contents
+  otherwise, which is extremely annoying for small projects.
+  (there's a mountain of empty vertical space on almost all projects! why collapse?)
 
 EO Datasets 3
-=============
+-------------
+
 
 EO Datasets aims to be the easiest way to write, validate and convert dataset imagery
 and metadata for the `Open Data Cube`_
 
 .. _Open Data Cube: https://github.com/opendatacube/datacube-core
 
+There are two major tools for creating datasets:
+
+ #. :ref:`DatasetAssembler<assembling_metadata>`, for writing a package: including writing the metadata
+    document, `COG`_ imagery, thumbnails, checksum files etc.
+
+ #. :ref:`DatasetPrepare<preparing_metadata>`, for preparing a metadata document,
+    referencing existing imagery and files.
+
+Their APIs are the same, except the assembler adds methods named ``write_*`` for writing new files.
 
-Write a Dataset
-================
+.. note ::
 
-Here's a simple example of creating a dataset with one measurement (called "blue") from an existing image::
+   methods named ``note_*`` will note an existing file in the metadata, while ``write_*`` will write
+   the file into the package too.
+
+.. _assembling_metadata:
+
+Assemble a Dataset Package
+--------------------------
+
+Here's a simple example of creating a dataset package with one measurement (called "blue") from an existing image.
+
+The measurement is converted to a `COG`_ image when written to the package::
 
    collection = Path('/some/output/collection/path')
-   with DatasetAssembler(collection) as p:
+   with DatasetAssembler(collection, naming_conventions='default') as p:
       p.product_family = "blues"
 
       # Date of acquisition (UTC if no timezone).
@@ -36,10 +56,75 @@ Here's a simple example of creating a dataset with one measurement (called "blue
       # Complete the dataset.
       p.done()
 
-Note that until you call `done()`, nothing will exist in the dataset's final output location. It is stored in a hidden temporary
-folder in the output directory, and renamed by `done()` once complete and valid.
+.. note ::
+
+   Note that until you call ``done()``, nothing will exist in the dataset's final output location. It is stored in a hidden temporary
+   folder in the output directory, and renamed by ``done()`` once complete and valid.
+
+
+.. _preparing_metadata:
+
+Writing only a metadata doc
+---------------------------
+
+(ie. "I already have appropriate imagery files!")
+
+Example of generating a metadata document with :class:`DatasetPrepare <eodatasets3.DatasetPrepare>`:
+
+.. testsetup ::
+
+   from pathlib import Path
+   import tempfile
+   collection_path = Path('../tests/integration/data').resolve()
+   from datetime import datetime
+
+.. testcode ::
+
+   from eodatasets3 import DatasetPrepare
 
-Custom stac-like properties can also be set directly on ``.properties``::
+   usgs_level1 = collection_path / 'LC08_L1TP_090084_20160121_20170405_01_T1'
+   metadata_path = usgs_level1 / 'odc-metadata.yaml'
+
+   with DatasetPrepare(
+       metadata_path=metadata_path,
+   ) as p:
+       p.product_family = "level1"
+       p.datetime = datetime(2019, 7, 4, 13, 7, 5)
+       p.processed_now()
+
+       # Note the measurement in the metadata. (instead of ``write``)
+       p.note_measurement('red',
+          usgs_level1 / 'LC08_L1TP_090084_20160121_20170405_01_T1_B4.TIF'
+       )
+
+       # Or give the path relative to the dataset location
+       # (eg. This will work unchanged on non-filesystem locations, such as ``s3://`` or tar files)
+       p.note_measurement('blue',
+          'LC08_L1TP_090084_20160121_20170405_01_T1_B2.TIF',
+          relative_to_dataset_location=True
+       )
+
+       # Add links to other files included in the package ("accessories"), such as
+       # alternative metadata files.
+       [mtl_path] = usgs_level1.glob('*_MTL.txt')
+       p.note_accessory_file('metadata:mtl', mtl_path)
+
+       # Add whatever else you want.
+       ...
+
+       # Validate and write our metadata document!
+       p.done()
+
+   # We created a metadata file!
+   assert metadata_path.exists()
+
+..  testcleanup ::
+
+   if metadata_path.exists():
+       metadata_path.unlink()
+
+
+Custom properties can also be set directly on ``.properties``::
 
       p.properties['fmask:cloud_cover'] = 34.0
 
@@ -51,11 +136,28 @@ And known properties are automatically normalised::
       p.properties["eo:off_nadir"] = "34"  # into a number
 
 
+.. note ::
+
+   By default, a validation error will be thrown if a referenced file lives outside the dataset (location),
+   as this will require absolute paths. Relative paths are considered best-practice for Open Data Cube
+   metadata.
+
+   You can allow absolute paths in your metadata document using a field on construction
+   (:meth:`DatasetPrepare() <eodatasets3.DatasetPrepare.__init__>`)::
+
+       with DatasetPrepare(
+          dataset_location=usgs_level1,
+          allow_absolute_paths=True,
+        ):
+            ...
+
+.. _COG: https://www.cogeo.org/
+
+
 Including provenance
-====================
-Most of our datasets are processed from an existing (input) dataset and
-have the same spatial information. We can add them as source datasets, to record
-the provenance, and the assembler can optionally copy any common metadata automatically::
+--------------------
+Most datasets are processed from an existing input dataset and have the same spatial information as the input.
+We can record them as source datasets, and the assembler can optionally copy any common metadata automatically::
 
    collection = Path('/some/output/collection/path')
    with DatasetAssembler(collection) as p:
@@ -71,68 +173,355 @@ the provenance, and the assembler can optionally copy any common metadata automa
 
       ...
 
-In these situations, we often write our new pixels as a numpy array, inheriting the existing grid spatial information (*gridspec*)
-from our input dataset::
+In these situations, we often write our new pixels as a numpy array, inheriting the existing
+:class:`grid spatial information <eodatasets3.GridSpec>` of our input dataset::
 
       # Write a measurement from a numpy array, using the source dataset's grid spec.
       p.write_measurement_numpy(
-         "ones",
-         numpy.ones((60, 60), numpy.int16),
-         GridSpec.from_dataset_doc(l1_ls8_dataset),
+         "water",
+         my_computed_numpy_array,
+         GridSpec.from_dataset_doc(source_dataset),
          nodata=-999,
       )
 
-Writing only metadata
-=====================
+Creating documents in-memory
+----------------------------
 
-The above examples copy the imagery, converting them to valid COG_ files in a new location. But sometimes you
-want to leave the imagery as-is and just generate a metadata file for Open Data Cube. We can
-do this by using :meth:`eodatasets3.DatasetAsssembler.note_measurement`
-instead of :meth:`eodatasets3.DatasetAsssembler.write_measurement`, to note the path
-of the current image::
+You may want to assemble metadata entirely in memory without touching the filesystem.
 
+To do this, prepare a dataset as normal. You still need to give a dataset location, as paths
+in the document will be relative to this location:
 
+.. testsetup:: inmem
 
-    usgs_level1 = Path('datasets/LC08_L1TP_090084_20160121_20170405_01_T1')
+   from eodatasets3 import GridSpec
+   from affine import Affine
+   from rasterio.crs import CRS
+   from pathlib import Path
+   import numpy
+   from datetime import datetime
 
-    with DatasetAssembler(
-      dataset_location=usgs_level1
-    ) as p:
-      p.product_family = "level1"
-      p.datetime = datetime(2019, 7, 4, 13, 7, 5)
+   grid_spec = GridSpec(shape=(7721, 7621),
+      transform=Affine(30.0, 0.0, 241485.0, 0.0, -30.0, -2281485.0),
+      crs=CRS.from_epsg(32656)
+   )
 
-      # Note the measurement in the metadata. (instead of ``write``)
-      p.note_measurement('red',
-         usgs_level1 / 'LC08_L1TP_090084_20160121_20170405_01_T1_B3.TIF'
-      )
+   import tempfile
+   dataset_location = Path(tempfile.mkdtemp())
+   measurement_path = dataset_location / "our_image_dont_read_it.tif"
 
-      # Or relative to the dataset
-      # (this will work unchanged on non-filesystem locations, such as ``s3://`` or tar files)
-      p.note_measurement('blue',
-         'LC08_L1TP_090084_20160121_20170405_01_T1_B3.TIF',
-         relative_to_dataset_location=True
-      )
+.. doctest:: inmem
 
-      ...
+   >>> from eodatasets3 import DatasetPrepare
+   >>>
+   >>> p = DatasetPrepare(dataset_location=dataset_location)
+   >>> p.datetime = datetime(2019, 7, 4, 13, 7, 5)
+   >>> p.product_name = "loch_ness_sightings"
+   >>> p.processed = datetime(2019, 7, 4, 13, 8, 7)
 
-Note that the assembler will throw an error if the path lives outside
-the dataset (location), as this will require absolute paths.
-Relative paths are considered best-practice for Open Data Cube.
 
-You can allow absolute paths with a field on assembler construction
-:meth:`eodatasets3.DatasetAssembler.__init__`::
+Normally when a measurement is added, the image will be opened to read
+grid and size informaation. You can avoid this by giving a :class:`GridSpec <eodatasets3.GridSpec>`
+yourself (see :class:`GridSpec doc <eodatasets3.GridSpec>` for creation):
 
-   with DatasetAssembler(
-      dataset_location=usgs_level1,
-      allow_absolute_paths=True,
-    ):
-        ...
+.. doctest:: inmem
 
-.. _COG: https://www.cogeo.org/
+   >>> p.note_measurement(
+   ...     "blue",
+   ...     measurement_path,
+   ...     # We give it grid information, so it doesn't have to read it itself.
+   ...     grid=grid_spec,
+   ...     # And the image pixels, since we are letting it calculate our geometry.
+   ...     pixels=numpy.ones((60, 60), numpy.int16),
+   ...     nodata=-1,
+   ... )
+
+.. note::
+
+   If you're writing your own image files manually, you may still want to use eodataset's
+   name generation. You can ask for suitable paths from
+   :attr:`p.names <eodatasets3.DatasetPrepare.names>`:
+
+   See the :ref:`the naming section<names_n_paths>` for examples.
+
+Now finish it as a :class:`DatasetDoc <eodatasets3.DatasetDoc>`:
+
+.. doctest:: inmem
+
+   >>> dataset = p.to_dataset_doc()
+
+You can now use :ref:`serialise functions<serialise_explanation>` on the result yourself,
+such as conversion to a dictionary:
+
+.. doctest:: inmem
+
+   >>> from eodatasets3 import serialise
+   >>> doc: dict = serialise.to_doc(dataset)
+   >>> doc['label']
+   'loch_ness_sightings_2019-07-04'
+
+Or convert it to a formatted yaml: :meth:`serialise.to_path(path, dataset) <eodatasets3.serialise.to_path>` or
+:meth:`serialise.to_stream(stream, dataset) <eodatasets3.serialise.to_stream>`.
+
+
+Avoiding geometry calculation
+-----------------------------
+
+Datasets include a geometry field, which shows the coverage of valid data pixels of
+all measurements.
+
+By default, the assembler will create this geometry by reading the pixels from your
+measurements, and calculate a geometry vector on completion.
+
+If you want to avoid these reads and calculations, you can set the geometry manually::
+
+    p.geometry = my_shapely_polygon
+
+Or copy it from one of your source datasets when you add your provenance (if it has
+the same coverage)::
+
+    p.add_source_path(source_path, inherit_geometry=True)
+
+If you do this before you `note` measurements, it will not need to read any pixels
+from them.
+
+.. _names_n_paths :
+
+Generating names & paths alone
+------------------------------
+
+You can use the naming module alone to find file paths:
+
+..
+  Sorry future readers, but the error output of assertion failures is abysmal
+  in these tests:
+
+.. testcode::
+
+   import eodatasets3
+   from pathlib import Path
+   from eodatasets3 import DatasetDoc
+
+Create some properties.
+
+.. testcode::
+
+   d = DatasetDoc()
+   d.platform = "sentinel-2a"
+   d.product_family = "fires"
+   d.datetime = "2018-05-04T12:23:32"
+   d.processed_now()
+
+   # Arbitrarily set any properties.
+   d.properties["fmask:cloud_shadow"] = 42.0
+   d.properties.update({"odc:file_format": "GeoTIFF"})
+
+.. note::
+   You can use a plain dict if you prefer. But we use an :class:`DatasetDoc() <eodatasets3.DatasetDoc>` here, which has
+   convenience methods similar to :class:`DatasetAssembler <eodatasets3.DatasetAssembler>` for building properties.
+
+Now create a `namer` instance with our properties (and chosen naming conventions):
+
+.. testcode::
+
+   names = eodatasets3.namer(d, conventions="default")
+
+We can see some generated names:
 
-API / Class
-===========
+.. testcode::
+
+   print(names.metadata_file)
+   print(names.measurement_filename('water'))
+   print()
+   print(names.product_name)
+   print(names.dataset_folder)
+
+Output:
+
+.. testoutput::
+
+    s2a_fires_2018-05-04.odc-metadata.yaml
+    s2a_fires_2018-05-04_water.tif
+
+    s2a_fires
+    s2a_fires/2018/05/04
+
+In reality, these paths go within a location (folder, s3 bucket, etc) somewhere.
+
+This location is called the `collection_prefix`, and we can create our namer with one:
+
+.. testcode::
+
+   collection_path = Path('/datacube/collections')
+
+   names = eodatasets3.namer(d, collection_prefix=collection_path)
+
+   print("The dataset location is always a URL:")
+   print(names.dataset_location)
+
+   print()
+
+   a_file_name = names.measurement_filename('water')
+   print(f"We can resolve our previous file name to a dataset URL:")
+   print(names.resolve_file(a_file_name))
+
+   print()
+
+   print(f"Or a local path (if it's file://):")
+   print(repr(names.resolve_path(a_file_name)))
+
+.. testoutput ::
+
+    The dataset location is always a URL:
+    file:///datacube/collections/s2a_fires/2018/05/04/s2a_fires_2018-05-04.odc-metadata.yaml
+
+    We can resolve our previous file name to a dataset URL:
+    file:///datacube/collections/s2a_fires/2018/05/04/s2a_fires_2018-05-04_water.tif
+
+    Or a local path (if it's file://):
+    PosixPath('/datacube/collections/s2a_fires/2018/05/04/s2a_fires_2018-05-04_water.tif')
+
+..
+   Let's override it quietly so we don't touch real paths on their system:
+
+.. testcode::
+   :hide:
+
+   from eodatasets3 import DatasetAssembler
+
+   import tempfile
+   collection_path = Path(tempfile.mkdtemp())
+
+   names.collection_prefix = collection_path.as_uri()
+
+
+.. note ::
+
+   The collection prefix can also be a remote url: ``https://example.com/collections``, ``s3:// ...`` ... etc
+
+
+We could now start assembling some metadata if our dataset doesn't exist,
+passing it our existing fields:
+
+.. testcode::
+
+   # Our dataset doesn't exist?
+   if not names.dataset_path.exists():
+       with DatasetAssembler(names=names) as p:
+
+          # The properties are already set, thanks to our namer.
+
+          ... # Write some measurements here, etc!
+
+          p.done()
+
+   # It exists!
+   assert names.dataset_path.exists()
+
+.. note ::
+
+   ``.dataset_path`` is a convenience property to get the ``.dataset_location``
+   as a local Path, if possible.
+
+.. note ::
+
+   The assembler classes don't yet support writing to remote locations! But you can use the
+   above api to write it yourself manually (for now).
+
+Naming things yourself
+----------------------
+
+Names and paths are only auto-generated if they have not been set manually
+by the user.
+
+You can set properties yourself on the :class:`NameGenerator <eodatasets3.NameGenerator>`
+to avoid automatic generation (or to avoid their finicky metadata requirements).
+
+.. testsetup:: nametest
+
+   from eodatasets3 import DatasetPrepare
+   from pathlib import Path
+   import tempfile
+
+   collection_path = Path(tempfile.mkdtemp())
+
+.. doctest:: nametest
+
+   >>> p = DatasetPrepare(collection_path)
+   >>> p.platform = 'sentinel-2a'
+   >>> p.product_family = 'ard'
+   >>> # The namer will generate a product name:
+   >>> p.names.product_name
+   's2a_ard'
+   >>> # Let's customise the generated abbreviation:
+   >>> p.names.platform_abbreviated = "s2"
+   >>> p.names.product_name
+   's2_ard'
+
+See more examples in the assembler :attr:`.names <eodatasets3.DatasetPrepare.names>` property.
+
+
+Dataset Prepare class reference
+-------------------------------
+
+.. autoclass:: eodatasets3.DatasetPrepare
+   :members:
+   :special-members: __init__
+
+Dataset Assembler class reference
+---------------------------------
 
 .. autoclass:: eodatasets3.DatasetAssembler
    :members:
    :special-members: __init__
+
+.. _serialise_explanation:
+
+Reading/Writing YAMLs
+---------------------
+
+Methods for parsing and outputting EO3 docs as a :class:`eodatasets3.DatasetDoc`
+
+Parsing
+^^^^^^^
+
+.. autofunction:: eodatasets3.serialise.from_path
+.. autofunction:: eodatasets3.serialise.from_doc
+
+
+Writing
+^^^^^^^
+
+.. autofunction:: eodatasets3.serialise.to_path
+.. autofunction:: eodatasets3.serialise.to_stream
+.. autofunction:: eodatasets3.serialise.to_doc
+
+Name Generation API
+-------------------
+
+You may want to use the name generation alone, for instance
+to tell if a dataset has already been written before you assemble it.
+
+.. autofunction:: eodatasets3.namer
+
+.. autoclass:: eodatasets3.NameGenerator
+   :members:
+   :inherited-members:
+
+EO Metadata API
+---------------
+
+.. autoclass:: eodatasets3.properties.Eo3Interface
+   :members:
+
+
+Misc Types
+----------
+
+..
+  Catch any types we didn't add explicitly above
+
+.. automodule:: eodatasets3
+   :members:
+   :exclude-members: DatasetAssembler, DatasetPrepare, NameGenerator, namer
+
diff --git a/eodatasets3/__init__.py b/eodatasets3/__init__.py
index 0b18fb88..f9fc3c8f 100644
--- a/eodatasets3/__init__.py
+++ b/eodatasets3/__init__.py
@@ -3,11 +3,28 @@
 from __future__ import absolute_import
 
 from ._version import get_versions
-from .assemble import DatasetAssembler, IfExists
+from .assemble import DatasetAssembler, DatasetPrepare, IfExists, IncompleteDatasetError
+from .images import GridSpec
+from .model import DatasetDoc
+from .properties import Eo3Dict
+
+from .names import namer, NameGenerator
 
 REPO_URL = "https://github.com/GeoscienceAustralia/eo-datasets.git"
 
 __version__ = get_versions()["version"]
 del get_versions
 
-__all__ = (DatasetAssembler, IfExists, REPO_URL, __version__)
+__all__ = (
+    "DatasetAssembler",
+    "DatasetDoc",
+    "DatasetPrepare",
+    "Eo3Dict",
+    "GridSpec",
+    "IfExists",
+    "IncompleteDatasetError",
+    "NameGenerator",
+    "namer",
+    "REPO_URL",
+    "__version__",
+)
diff --git a/eodatasets3/assemble.py b/eodatasets3/assemble.py
index 1412c9f6..ff076a4e 100644
--- a/eodatasets3/assemble.py
+++ b/eodatasets3/assemble.py
@@ -5,42 +5,48 @@
 import tempfile
 import uuid
 import warnings
-from collections import defaultdict
 from copy import deepcopy
-from enum import Enum
-from pathlib import Path
+from enum import Enum, auto
+from pathlib import Path, PurePath, PosixPath
 from textwrap import dedent
 from typing import Dict, List, Optional, Tuple, Generator, Any, Iterable, Union
+from urllib.parse import urlsplit
 
 import numpy
 import rasterio
+import xarray
 from rasterio import DatasetReader
 from rasterio.crs import CRS
 from rasterio.enums import Resampling
-from xarray import Dataset
+from shapely.geometry.base import BaseGeometry
 
 import eodatasets3
 from eodatasets3 import serialise, validate, images, documents
 from eodatasets3.documents import find_and_read_documents
-from eodatasets3.images import FileWrite, GridSpec, MeasurementRecord
+from eodatasets3.images import FileWrite, GridSpec, MeasurementBundler
 from eodatasets3.model import (
     DatasetDoc,
     ProductDoc,
-    StacPropertyView,
-    ComplicatedNamingConventions,
-    AccessoryDoc,
     Location,
-    ComplicatedNamingConventionsDerivatives,
+    AccessoryDoc,
 )
-from eodatasets3.properties import EoFields
+from eodatasets3.names import NameGenerator, namer, resolve_location
+from eodatasets3.properties import Eo3Interface, Eo3Dict
 from eodatasets3.validate import Level, ValidationMessage
 from eodatasets3.verify import PackageChecksum
 
 
 class IfExists(Enum):
-    Skip = 0
-    Overwrite = 1
-    ThrowError = 2
+    """
+    Enum: what to do when output already exists?
+    """
+
+    #: Skip the dataset
+    Skip = auto()
+    #: Overwrite the existing dataset
+    Overwrite = auto()
+    #: Throw an error
+    ThrowError = auto()
 
 
 class AssemblyError(Exception):
@@ -48,11 +54,17 @@ class AssemblyError(Exception):
 
 
 class IncompleteDatasetError(Exception):
+    """
+    Raised when a dataset is missing essential things and so cannot be written.
+
+    (such as mandatory metadata)
+    """
+
     def __init__(self, validation: ValidationMessage) -> None:
         self.validation = validation
 
 
-class DatasetCompletenessWarning(UserWarning):
+class IncompleteDatasetWarning(UserWarning):
     """A non-critical warning for invalid or incomplete metadata"""
 
     def __init__(self, validation: ValidationMessage) -> None:
@@ -62,8 +74,36 @@ def __str__(self) -> str:
         return str(self.validation)
 
 
-class DatasetAssembler(EoFields):
-    # Properties that can be inherited from a source dataset. (when auto_inherit_properties=True)
+def _validate_property_name(name: str):
+    """
+    >>> _validate_property_name('eo:gsd')
+    >>> _validate_property_name('thumbnail:full_resolution')
+    >>> _validate_property_name('full resolution')
+    Traceback (most recent call last):
+       ...
+    ValueError: Not a valid property name 'full resolution' (must be alphanumeric with colons or underscores)
+    >>> _validate_property_name('Mr Sprinkles')
+    Traceback (most recent call last):
+      ...
+    ValueError: Not a valid property name 'Mr Sprinkles' (must be alphanumeric with colons or underscores)
+    """
+    if not name.replace(":", "").isidentifier():
+        raise ValueError(
+            f"Not a valid property name {name!r} "
+            "(must be alphanumeric with colons or underscores)"
+        )
+
+
+class DatasetPrepare(Eo3Interface):
+    """
+    Prepare dataset metadata
+    """
+
+    #: The properties that will automatically be inherited from a source dataset
+    #: when :meth:`auto_inherit_properties=True <.add_source_path>`
+    #:
+    #: These are fields that are inherent to the underlying observation, and so will
+    #: still be relevant after most 1:1 processing.
     INHERITABLE_PROPERTIES = {
         "datetime",
         "dtr:end_datetime",
@@ -123,17 +163,23 @@ class DatasetAssembler(EoFields):
 
     def __init__(
         self,
-        collection_location: Optional[Path] = None,
-        dataset_location: Optional[Location] = None,
+        collection_location: Optional[Location] = None,
+        *,
+        dataset_location: Optional[Union[str, Path]] = None,
         metadata_path: Optional[Path] = None,
         dataset_id: Optional[uuid.UUID] = None,
-        # By default, we complain if the output already exists.
-        if_exists: IfExists = IfExists.ThrowError,
         allow_absolute_paths: bool = False,
         naming_conventions: str = "default",
+        names: Optional[NameGenerator] = None,
+        dataset: Optional[DatasetDoc] = None,
     ) -> None:
         """
-        Assemble a dataset with ODC metadata, writing metadata and (optionally) its imagery as COGs.
+        Build an EO3 metadata document, with functions for reading information from imagery
+        and calculating names and paths.
+
+        In addition to the below documented methods, metadata fields can read and set using
+        :class:`Eo3Interface's <eodatasets3.properties.Eo3Interface>` fields.
+
 
         There are three optional paths that can be specified. At least one must be specified. Collection,
         dataset or metadata path.
@@ -155,16 +201,17 @@ def __init__(
             Optional base directory where the collection of datasets should live. Subfolders will be
             created accordion to the naming convention.
         :param dataset_location:
-            Optional location for this specific dataset. Otherwise it will be generated according to the collection
-            path and naming conventions.
+            Optional location for this dataset.
+
+            Otherwise it will be generated according to the collection path and naming conventions.
+
+            (this is as indexed into ODC -- ie. a file name).
         :param metadata_path:
             Optional metadata document output path. Otherwise it will be generated according to the collection path
             and naming conventions.
         :param dataset_id:
             Optional UUID for this dataset, otherwise a random one will be created. Use this if you have a stable
             way of generating your own IDs.
-        :param if_exists:
-            What to do if the output dataset already exists? By default, throw an error.
         :param allow_absolute_paths:
             Allow metadata paths to refer to files outside the dataset location. this means they will have to be
             absolute paths, and not be portable. (default: False)
@@ -173,93 +220,147 @@ def __init__(
             (try it and see -- it will tell your what's missing).
         """
 
-        # Optionally give a fixed dataset id.
-        self.dataset_id = dataset_id or uuid.uuid4()
-        self._exists_behaviour = if_exists
-
-        if not collection_location and not metadata_path:
+        if (
+            (names is None)
+            and not collection_location
+            and not dataset_location
+            and not metadata_path
+        ):
             raise ValueError(
-                "Must specify either a collection folder or a single metadata file"
+                "Must specify either a collection folder, dataset location or a single metadata file"
             )
+        if not collection_location:
+            if dataset_location and not dataset_location.is_absolute():
+                raise ValueError(
+                    "Dataset location must be absolute if no collection location specified."
+                )
 
         if collection_location and not collection_location.exists():
             raise ValueError(
                 f"Provided collection location doesn't exist: {collection_location}"
             )
 
-        self._checksum = PackageChecksum()
-        self._tmp_work_path: Optional[Path] = None
+        if not dataset:
+            dataset = DatasetDoc()
+        if not dataset.id:
+            dataset.id = dataset_id or uuid.uuid4()
 
-        self._label = None
-        self._props = StacPropertyView()
-
-        self.collection_location = collection_location
-        self._dataset_location = dataset_location
-        self._metadata_path = metadata_path
-        self._allow_absolute_paths = allow_absolute_paths
+        self._dataset = dataset
 
+        self._measurements = MeasurementBundler()
         self._accessories: Dict[str, Location] = {}
-        self._measurements = MeasurementRecord()
 
-        self._user_metadata = dict()
-        self._software_versions: List[Dict] = []
-        self._lineage: Dict[str, List[uuid.UUID]] = defaultdict(list)
-        self._inherited_geometry = None
-
-        if naming_conventions == "default":
-            self.names = ComplicatedNamingConventions(self)
-        elif naming_conventions == "dea":
-            self.names = ComplicatedNamingConventions.for_standard_dea(self)
-        elif naming_conventions == "dea_s2":
-            self.names = ComplicatedNamingConventions.for_standard_dea_s2(self)
-        elif naming_conventions == "dea_s2_derivative":
-            self.names = ComplicatedNamingConventionsDerivatives.for_s2_derivatives(
-                self
-            )
-        elif naming_conventions == "dea_c3":
-            self.names = ComplicatedNamingConventionsDerivatives.for_c3_derivatives(
-                self
-            )
-        elif naming_conventions == "deafrica":
-            self.names = (
-                ComplicatedNamingConventionsDerivatives.for_deafrica_derivatives(self)
-            )
-        else:
-            raise NotImplementedError("configurable naming conventions")
+        self._allow_absolute_paths = allow_absolute_paths
 
+        #: Valid-data polygon, in the same CRS as the measurements.
+        #:
+        #: This must cover all valid pixels to be valid in ODC
+        #: (it's allowed to be larger than the valid pixel area, but not
+        #: smaller).
+        #:
+        #: It will be computed automatically from measurements if not set
+        #: manually. You can also inherit it from source datasets in the
+        #: ``add_source_*()`` methods.
+        self.geometry: Optional[BaseGeometry] = None
+
+        # They may have given us initialised naming conventions already:
+        if names is not None:
+            #: The name generator  (an instance of :class:`NameGenerator <eodatasets3.NameGenerator>`)
+            #:
+            #: By default, all names will be generated based on metadata
+            #: fields and the chosen naming conventions.
+            #:
+            #: But you can set your own names here manually to avoid the magic.
+            #:
+            #: (for the devious among you, this can also avoid metadata field requirements
+            #: for name generation).
+            #:
+            #: Examples:
+            #:
+            #: Set a product name::
+            #:
+            #:     p.names.product_name = 'my_product_name'
+            #:
+            #: Manually set the abbreviations used in name generation
+            #:
+            #: (By default, for example, landsat-7 will be abbreviated to "ls7". But maybe
+            #: you want "ls" in all your datasets)::
+            #:
+            #:     p.names.platform_abbreviated = "ls"
+            #:     # Other abbreviations:
+            #:     p.names.instrument_abbreviated = "e"
+            #:     p.names.producer_abbreviated = "usgs"
+            #:
+            #: Set your own label
+            #: (the human identifier for the dataset, and the default prefix of filenames)::
+            #:
+            #:     p.names.dataset_label = "landsat-observations-12th-may-2021"
+            #:
+            #: Change the folder offset for each dataset. All generated files paths are relative
+            #: to this folder (and it is relative to the collection path)::
+            #:
+            #:     p.names.dataset_folder = Path("datasets/january/2021")
+            #:
+            #: Set a different pattern used for generating filenames
+            #: All filenames have a file_id (eg. "odc-metadata" or "") and a suffix (eg. "yaml")
+            #: (Can contain folder separators. It will be relative to the dataset folder)::
+            #:
+            #:     p.names.filename_pattern = "my-file.{file_id}.{suffix}"
+            #:
+            #: The path to the EO3 metadata doc (relative path to the dataset location)::
+            #:
+            #:     p.names.metadata_file = Path("my-metadata.odc-metadata.yaml")
+            #:
+            #: The URI for the product::
+            #:
+            #:     p.names.product_uri = "https://collections.earth.test.example/product/my-product"
+            #:
+            #: A full list of fields can be seen on :class:`eodatasets3.NameGenerator`
+            self.names: NameGenerator = names
+            dataset.properties = names.metadata.properties
+        else:
+            # Or create some:
+            self.names: NameGenerator = namer(
+                dataset.properties, conventions=naming_conventions
+            )
+        if collection_location:
+            self.names.collection_prefix = resolve_location(collection_location)
+        if dataset_location:
+            self.names.dataset_location = resolve_location(dataset_location)
+        if metadata_path:
+            self.names.metadata_file = metadata_path
+            if not dataset_location:
+                try:
+                    self.names.dataset_location
+                except ValueError:
+                    # They didn't set a dataset location.
+                    # Default it to the metadata path.
+                    self.names.dataset_location = resolve_location(metadata_path)
         self._is_completed = False
         self._finished_init_ = True
 
-    def _is_writing_files(self):
-        """
-        Have they written any files? Otherwise we're just writing a metadata doc
-        """
-        # A tmpdir is created on the first file written.
-        # TODO: support writing files in declared dataset_locations too.
-        return self.collection_location is not None
+    # Our with-blocks don't do anything as there's nothing to clean-up, but we want it to
+    # be a drop-in replacement for DatasetAssembler, so we let users use them.
+    # (it can also make code more readable, to have a clear block)
+    def __enter__(self) -> "DatasetPrepare":
+        return self
 
-    @property
-    def _work_path(self) -> Path:
-        """
-        The current folder path of the maybe-partially-built dataset.
-        """
-        if not self._tmp_work_path:
-            if not self.collection_location:
-                raise ValueError(
-                    "Dataset assembler was given no base path on construction: cannot write new files."
-                )
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        ...
 
-            self._tmp_work_path = Path(
-                tempfile.mkdtemp(
-                    prefix=".odcdataset-", dir=str(self.collection_location)
-                )
-            )
+    @property
+    def dataset_id(self) -> uuid.UUID:
+        return self._dataset.id
 
-        return self._tmp_work_path
+    @dataset_id.setter
+    def dataset_id(self, val: Union[uuid.UUID, str]):
+        if isinstance(val, str):
+            val = uuid.UUID(val)
+        self._dataset.id = val
 
     @property
-    def properties(self) -> StacPropertyView:
-        return self._props
+    def properties(self) -> Eo3Dict:
+        return self._dataset.properties
 
     @property
     def measurements(self) -> Dict[str, Tuple[GridSpec, Path]]:
@@ -282,14 +383,11 @@ def label(self) -> Optional[str]:
         A label will be auto-generated using the naming-conventions, but you can manually override it by
         setting this property.
         """
-        return self._label or self.names.dataset_label
+        return self._dataset.label or self.names.dataset_label
 
     @label.setter
     def label(self, val: str):
-        self._label = val
-
-    def __enter__(self) -> "DatasetAssembler":
-        return self
+        self._dataset.label = val
 
     def __setattr__(self, name: str, value: Any) -> None:
         """
@@ -306,44 +404,30 @@ def __setattr__(self, name: str, value: Any) -> None:
             )
         super().__setattr__(name, value)
 
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        # Clean up.
-        self.close()
-
-    def cancel(self):
-        """
-        Cancel the package, cleaning up temporary files.
-
-        This works like :func:`DatasetAssembler.close`, but is intentional, so no warning will
-        be raised for forgetting to complete the package first.
-        """
-        self._is_completed = True
-        self.close()
-
-    def close(self):
-        """Clean up any temporary files, even if dataset has not been written"""
-        if not self._is_completed:
-            warnings.warn(
-                "Closing assembler without finishing. "
-                "Either call `done()` or `cancel() before closing`"
-            )
-
-        if self._tmp_work_path:
-            # TODO: add implicit cleanup like tempfile.TemporaryDirectory?
-            shutil.rmtree(self._tmp_work_path, ignore_errors=True)
-
     def add_source_path(
         self,
         *paths: Path,
         classifier: str = None,
         auto_inherit_properties: bool = False,
+        inherit_geometry: bool = False,
     ):
         """
         Record a source dataset using the path to its metadata document.
 
-        :param paths:
+        :param paths: Filesystem path(s) to source metadata documents
+
+        :param classifier: How to classify the kind of source dataset. This is will automatically
+                           be filled with the family of dataset if available (eg. "level1").
+
+                           You want to set this if you have two datasets of the same type that
+                           are used for different purposes. Such as having a second level1 dataset
+                           that was used for QA (but is not this same scene).
+        :param auto_inherit_properties: Whether to copy any common properties from the dataset
+        :param inherit_geometry: Instead of re-calculating the valid bounds geometry based on the
+                            data, which can be very computationally expensive e.g. Landsat 7
+                            striped data, use the valid data geometry from this source dataset.
 
-        See other parameters in :func:`DatasetAssembler.add_source_dataset`
+        See also :meth:`.add_source_dataset`
         """
         for _, doc in find_and_read_documents(*paths):
             # Newer documents declare a schema.
@@ -352,9 +436,10 @@ def add_source_path(
                     serialise.from_doc(doc),
                     classifier=classifier,
                     auto_inherit_properties=auto_inherit_properties,
+                    inherit_geometry=inherit_geometry,
                 )
             else:
-                if auto_inherit_properties:
+                if auto_inherit_properties or inherit_geometry:
                     raise NotImplementedError(
                         "Can't (yet) inherit properties from old-style metadata"
                     )
@@ -366,7 +451,8 @@ def add_source_path(
                         "Source dataset (of old-style eo) doesn't have a 'product_type' property (eg. 'level1', 'fc'), "
                         "you must specify a classifier for the kind of source dataset."
                     )
-                self._lineage[classifier].append(doc["id"])
+                _validate_property_name(classifier)
+                self._dataset.lineage.setdefault(classifier, []).append(doc["id"])
 
     def add_source_dataset(
         self,
@@ -380,7 +466,7 @@ def add_source_dataset(
 
         It can optionally copy common properties from the source dataset (platform, instrument etc)/
 
-        (see self.INHERITABLE_PROPERTIES for the list of fields that are inheritable)
+        (see :py:obj:`.INHERITABLE_PROPERTIES` for the list of fields that are inheritable)
 
         :param dataset:
         :param auto_inherit_properties: Whether to copy any common properties from the dataset
@@ -396,10 +482,9 @@ def add_source_dataset(
                             data, which can be very computationally expensive e.g. Landsat 7
                             striped data, use the valid data geometry from this source dataset.
 
-        See :func:`add_source_path` if you have a filepath reference instead of a document.
+        See :meth:`.add_source_path` if you have a filepath reference instead of a document.
 
         """
-
         if not classifier:
             classifier = dataset.properties.get("odc:product_family")
             if not classifier:
@@ -410,11 +495,14 @@ def add_source_dataset(
                     "you must specify a classifier for the kind of source dataset."
                 )
 
-        self._lineage[classifier].append(dataset.id)
+        _validate_property_name(classifier)
+        self._dataset.lineage.setdefault(classifier, []).append(dataset.id)
         if auto_inherit_properties:
             self._inherit_properties_from(dataset)
         if inherit_geometry:
-            self._inherited_geometry = dataset.geometry
+            if self.geometry and self.geometry != dataset.geometry:
+                warnings.warn("Overriding existing geometry from source dataset")
+            self.geometry = dataset.geometry
 
     def note_source_datasets(
         self,
@@ -424,8 +512,9 @@ def note_source_datasets(
         """
         Expand the lineage with raw source dataset ids.
 
-        Note: If you have direct access to the datasets, you probably want to use :func:`add_source_path`
-        or :func:`add_source_dataset`, so that fields can be inherited from them automatically.
+        .. note::
+           If you have direct access to the datasets, you probably want to use :func:`add_source_path`
+           or :func:`add_source_dataset`, so that fields can be inherited from them automatically.
 
         :param classifier:
                 How to classify the source dataset.
@@ -451,7 +540,7 @@ def note_source_datasets(
                     raise ValueError(
                         f"Not a valid UUID for source {classifier!r} dataset: {dataset_id!r}"
                     ) from v
-            self._lineage[classifier].append(dataset_id)
+            self._dataset.lineage.setdefault(classifier, []).append(dataset_id)
 
     def _inherit_properties_from(self, source_dataset: DatasetDoc):
         for name in self.INHERITABLE_PROPERTIES:
@@ -471,430 +560,706 @@ def _inherit_properties_from(self, source_dataset: DatasetDoc):
                     f"Inheritable property {name!r} is different from current value {k.args}"
                 )
 
-    def write_measurement(
+    def note_measurement(
         self,
-        name: str,
+        name,
         path: Location,
-        overviews: Iterable[int] = images.DEFAULT_OVERVIEWS,
-        overview_resampling: Resampling = Resampling.average,
-        expand_valid_data: bool = True,
-        file_id: str = None,
-    ):
-        """
-        Write a measurement by copying it from a file path.
-
-        Assumes the file is gdal-readable.
-
-        :param name: Identifier for the measurement eg ``'blue'``.
-        :param path:
-        :param overviews: Set of overview sizes to write
-        :param overview_resampling: rasterio Resampling method to use
-        :param expand_valid_data: Include this measurement in the valid-data geometry of the metadata.
-        :param file_id: Optionally, how to identify this in the filename instead of using the name.
-                        (DEA has measurements called ``blue``, but their written filenames must be ``band04`` by
-                        convention.)
-        """
-        with rasterio.open(path) as ds:
-            self.write_measurement_rio(
-                name,
-                ds,
-                overviews=overviews,
-                expand_valid_data=expand_valid_data,
-                overview_resampling=overview_resampling,
-                file_id=file_id,
-            )
-
-    def write_measurement_rio(
-        self,
-        name: str,
-        ds: DatasetReader,
-        overviews=images.DEFAULT_OVERVIEWS,
-        overview_resampling=Resampling.average,
         expand_valid_data=True,
-        file_id=None,
+        relative_to_dataset_location=False,
+        grid: GridSpec = None,
+        pixels: numpy.ndarray = None,
+        nodata: Optional[Union[float, int]] = None,
     ):
         """
-        Write a measurement by reading it from an open rasterio dataset
-
-        :param ds: An open rasterio dataset
-
-        See :func:`write_measurement` for other parameters.
+        Reference a measurement from its existing path. It may be a Path or any URL
+        resolvable by rasterio.
+
+        The path will be opened to read geo and pixel information, unless you specify the
+        information yourself (grid, pixels, nodata). (the latter two only needed if
+        expand_valid_data==True)
+
+        :param name: measurement name
+        :param path: path to measurement
+        :param expand_valid_data: Expand the valid data bounds with this measurement's valid data.
+        :param relative_to_dataset_location: Should this be read relative to the dataset location?
+                    (requires a computed dataset location)
         """
-        if len(ds.indexes) != 1:
-            raise NotImplementedError(
-                f"TODO: Multi-band images not currently implemented (have {len(ds.indexes)})"
-            )
+        _validate_property_name(name)
+
+        # If we have a polygon already, there's no need to compute valid data.
+        if self.geometry:
+            expand_valid_data = False
+
+        # If they didn't give us grid information, read it from the input.
+        if not grid:
+            read_location = path
+            if relative_to_dataset_location:
+                read_location = self.names.resolve_file(path)
+
+            with rasterio.open(read_location) as ds:
+                ds: DatasetReader
+                grid = images.GridSpec.from_rio(ds)
+                nodata = ds.nodata
+                if expand_valid_data:
+                    if not pixels:
+                        if ds.count != 1:
+                            raise NotImplementedError(
+                                "TODO: Only single-band files currently supported"
+                            )
+                        pixels = ds.read(1)
 
-        self._write_measurement(
+        self._measurements.record_image(
             name,
-            ds.read(1),
-            images.GridSpec.from_rio(ds),
-            self.names.measurement_file_path(
-                self._work_path, name, "tif", file_id=file_id
-            ),
+            grid,
+            path,
+            pixels,
+            nodata=nodata,
             expand_valid_data=expand_valid_data,
-            nodata=ds.nodata,
-            overview_resampling=overview_resampling,
-            overviews=overviews,
         )
 
-    def write_measurement_numpy(
-        self,
-        name: str,
-        array: numpy.ndarray,
-        grid_spec: GridSpec,
-        nodata: Optional[Union[float, int]] = None,
-        overviews=images.DEFAULT_OVERVIEWS,
-        overview_resampling=Resampling.average,
-        expand_valid_data=True,
-        file_id: str = None,
-    ):
-        """
-        Write a measurement from a numpy array and grid spec.
-
-        The most common case is to copy the grid spec from your input dataset,
-        assuming you haven't reprojected.
-
-        Example::
+    def _target_metadata_path(self) -> Path:
+        return self.names.resolve_path(self.names.metadata_file)
 
-            p.write_measurement_numpy(
-                "blue",
-                new_array,
-                GridSpec.from_dataset_doc(source_dataset),
-                nodata=-999,
+    def write_eo3(
+        self,
+        path: Path = None,
+        embed_location: bool = False,
+        validate_correctness: bool = True,
+        sort_measurements: bool = True,
+    ) -> Tuple[uuid.UUID, Path]:
+        """Write the prepared metadata document to the given output path."""
+        metadata_path = path or (self._target_metadata_path())
+        dataset_location = self.names.dataset_location
+
+        # Default behaviour:
+        #   If the metadata path is not the dataset location, then record the location.
+        if embed_location is None:
+            embed_location = dataset_location != metadata_path.as_uri()
+
+        doc = serialise.to_formatted_doc(
+            self.to_dataset_doc(
+                embed_location=embed_location,
+                validate_correctness=validate_correctness,
+                sort_measurements=sort_measurements,
             )
+        )
+        # It passed validation etc. Ensure output folder exists.
+        metadata_path.parent.mkdir(parents=True, exist_ok=True)
 
-        See :func:`write_measurement` for other parameters.
+        documents.make_paths_relative(
+            doc, metadata_path.parent, allow_paths_outside_base=False
+        )
+        serialise.dump_yaml(metadata_path, doc)
+        return self._dataset.id, metadata_path
 
-        :param array:
-        :param grid_spec:
-        :param nodata:
-        """
-        self._write_measurement(
-            name,
-            array,
-            grid_spec,
-            self.names.measurement_file_path(
-                self._work_path, name, "tif", file_id=file_id
-            ),
-            expand_valid_data=expand_valid_data,
-            nodata=nodata,
-            overview_resampling=overview_resampling,
-            overviews=overviews,
+    def done(
+        self,
+        validate_correctness: bool = True,
+        sort_measurements: bool = True,
+        embed_location: Optional[bool] = False,
+    ) -> Tuple[uuid.UUID, Path]:
+        """Write the prepared metadata document to the given output path."""
+        return self.write_eo3(
+            validate_correctness=validate_correctness,
+            sort_measurements=sort_measurements,
+            embed_location=embed_location,
         )
 
-    def write_measurements_odc_xarray(
+    def to_dataset_doc(
         self,
-        dataset: Dataset,
-        nodata: Optional[Union[float, int]] = None,
-        overviews=images.DEFAULT_OVERVIEWS,
-        overview_resampling=Resampling.average,
-        expand_valid_data=True,
-        file_id=None,
-    ):
+        dataset_location: Optional[str] = None,
+        embed_location: bool = False,
+        validate_correctness: bool = True,
+        sort_measurements: bool = True,
+    ) -> DatasetDoc:
         """
-        Write measurements from an ODC xarray.Dataset
+        Create the metadata doc as an in-memory :class:`eodatasets3.DatasetDoc` instance.
 
-        The main requirement is that the Dataset contains a CRS attribute
-        and X/Y or lat/long dimensions and coordinates. These are used to
-        create an ODC GeoBox.
+        (You can manually write this out using :func:`serialise.to_path(): <eodatasets3.serialise.to_path>`
+        or :func:`serialise.to_stream() <eodatasets3.serialise.to_stream>`)
+        """
+        dataset_location = dataset_location or self.names.dataset_location
 
-        :param dataset: an xarray dataset (as returned by ``dc.load()`` and other methods)
+        def rel_path(p: Union[str, Path]) -> str:
+            if isinstance(p, PurePath):
+                if p.is_absolute():
+                    return relative_url(dataset_location, p.as_uri())
+                else:
+                    return p.as_posix()
 
-        See :func:`write_measurement` for other parameters.
-        """
-        grid_spec = images.GridSpec.from_odc_xarray(dataset)
-        for name, dataarray in dataset.data_vars.items():
-            self._write_measurement(
-                name,
-                dataarray.data,
-                grid_spec,
-                self.names.measurement_file_path(
-                    self._work_path, name, "tif", file_id=file_id
-                ),
-                expand_valid_data=expand_valid_data,
-                overview_resampling=overview_resampling,
-                overviews=overviews,
-                nodata=nodata,
-            )
+            return p
 
-    def _write_measurement(
-        self,
-        name: str,
-        data: numpy.ndarray,
-        grid: GridSpec,
-        out_path: Path,
-        expand_valid_data: bool,
-        nodata: Optional[Union[float, int]],
-        overview_resampling: Resampling,
-        overviews: Tuple[int, ...],
-    ):
-        res = FileWrite.from_existing(grid.shape).write_from_ndarray(
-            data,
-            out_path,
-            geobox=grid,
-            nodata=nodata,
-            overview_resampling=overview_resampling,
-            overviews=overviews,
-        )
+        if not dataset_location:
+            raise ValueError("No location available: cannot calculate relative paths")
 
-        # Ensure the file_format field is set to what we're writing.
-        file_format = res.file_format.name
-        if "odc:file_format" not in self.properties:
-            self.properties["odc:file_format"] = file_format
+        dataset = self._dataset
+        if not dataset.product:
+            dataset.product = ProductDoc()
 
-        if file_format != self.properties["odc:file_format"]:
-            raise RuntimeError(
-                f"Inconsistent file formats between bands. "
-                f"Was {self.properties['odc:file_format']!r}, now {file_format !r}"
-            )
+        dataset.product.name = dataset.product.name or self.names.product_name
+        dataset.product.href = dataset.product.href or self.names.product_uri
+        dataset.label = dataset.label or self.names.dataset_label
+        if embed_location:
+            dataset.locations = [dataset_location]
+        else:
+            dataset.locations = None
 
-        self._measurements.record_image(
-            name,
-            grid,
-            out_path,
-            data,
-            nodata=nodata,
-            expand_valid_data=expand_valid_data,
-        )
-        # We checksum immediately as the file has *just* been written so it may still
-        # be in os/filesystem cache.
-        self._checksum.add_file(out_path)
+        crs, grid_docs, measurement_docs = self._measurements.as_geo_docs()
 
-    def note_measurement(
+        valid_data = self.geometry or self._measurements.consume_and_get_valid_data()
+
+        # Avoid the messiness of different empty collection types.
+        # (to have a non-null geometry we'd also need non-null grids and crses)
+        if valid_data.is_empty:
+            valid_data = None
+
+        new_crs = self._crs_str(crs) if crs is not None else None
+        if dataset.crs and dataset.crs != new_crs:
+            raise AssemblyError(
+                f"New measurements have a different CRS to the underlying dataset. "
+                f"Old: {dataset.crs!r}, New: {new_crs!r}"
+            )
+        dataset.crs = dataset.crs or new_crs
+        if valid_data:
+            if dataset.geometry:
+                dataset.geometry = dataset.geometry.union(valid_data)
+            else:
+                dataset.geometry = valid_data
+
+        # TODO: this could be made smarter, as we could merge with existing grids.
+        #       for now we just throw an error if any of our generated grid names
+        #       clash with existing ones.
+        if grid_docs:
+            if dataset.grids is None:
+                dataset.grids = {}
+
+            for name, doc in grid_docs.items():
+                if name in dataset.grids:
+                    raise NotImplementedError(
+                        f"Recorded grid name already exists in the underlying dataset: {name!r},"
+                        f"and we don't yet support merging of grids."
+                    )
+                dataset.grids[name] = doc
+
+        if measurement_docs:
+            if dataset.measurements is None:
+                dataset.measurements = {}
+
+            for name, doc in measurement_docs.items():
+                if name in dataset.measurements:
+                    raise AssemblyError(
+                        f"Recorded measurement already exists in the underlying dataset: {name!r}"
+                    )
+                doc.path = rel_path(doc.path)
+                dataset.measurements[name] = doc
+        for name, path in self._accessories.items():
+            if name in dataset.accessories:
+                raise AssemblyError(
+                    f"Recorded accessory already exists in the underlying dataset: {name!r}"
+                )
+            dataset.accessories[name] = AccessoryDoc(rel_path(path), name=name)
+
+        if dataset.measurements and sort_measurements:
+            # noinspection PyTypeChecker
+            dataset.measurements = dict(sorted(dataset.measurements.items()))
+
+        if validate_correctness:
+            doc = serialise.to_doc(dataset)
+            for m in validate.validate_dataset(doc):
+                if m.level in (Level.info, Level.warning):
+                    warnings.warn(IncompleteDatasetWarning(m))
+                elif m.level == Level.error:
+                    raise IncompleteDatasetError(m)
+                else:
+                    raise RuntimeError(
+                        f"Internal error: Unhandled type of message level: {m.level}"
+                    )
+
+        return dataset
+
+    def _crs_str(self, crs: CRS) -> str:
+        # TODO: We should support more authorities here.
+        #       if rasterio>=1.1.7, can use crs.to_authority(), but almost
+        #       everyone is currently on 1.1.6
+        return f"epsg:{crs.to_epsg()}" if crs.is_epsg_code else crs.to_wkt()
+
+    def note_accessory_file(self, name: str, path: Location):
+        """
+        Record a reference to an additional file that's included in the dataset, but is
+        not a band/measurement.
+
+        Such as non-ODC metadata, thumbnails, checksums, etc. Any included file that
+        is not recorded in the measurements.
+
+        By convention, the name should have prefixes with their category, such as
+        ``metadata:`` or ``thumbnail:``.
+
+        eg. ``metadata:landsat_processor``, ``checksum:sha1``, ``thumbnail:full``.
+
+        :param name: identifying name, eg ``metadata:mtl``
+        :param path: local path to file.
+        """
+        _validate_property_name(name)
+        existing_path = self._accessories.get(name)
+        if existing_path is not None and existing_path != path:
+            raise ValueError(
+                f"Duplicate accessory name {name!r}. "
+                f"New: {path!r}, previous: {existing_path!r}"
+            )
+        self._accessories[name] = path
+
+    def note_thumbnail(self, thumb_path: Path, kind: str = None):
+        """
+        Record a reference to a thumbnail path.
+
+        Optionally specify the "kind" of thumbnail if there are multiple
+        to distinguish between. eg. 'full'
+        """
+        accessory_name = "thumbnail"
+        if kind:
+            accessory_name += f":{kind}"
+        self.note_accessory_file(accessory_name, thumb_path)
+
+    def iter_measurement_paths(
         self,
-        name,
-        path: Location,
-        expand_valid_data=True,
-        relative_to_dataset_location=False,
-    ):
+    ) -> Generator[Tuple[GridSpec, str, Path], None, None]:
+        """
+
+        .. warning::
+           *not recommended* for use - will likely change soon.
+
+        Iterate through the list of measurement names that have been written, and their current (temporary) paths.
+
+        TODO: Perhaps we want to return a real measurement structure here as it's not very extensible.
         """
-        Reference a measurement from its existing file path.
+        return self._measurements.iter_paths()
 
-        (no data is copied, but Geo information is read from it.)
+    def __str__(self):
+        status = "written" if self._is_completed else "unfinished"
+
+        try:
+            output_location = self._target_metadata_path()
+        except ValueError:
+            output_location = "(not yet computable)"
 
-        :param name:
-        :param path:
-        :param expand_valid_data:
-        :param relative_to_dataset_location:
+        measurements = list(self._measurements.iter_names())
+        properties = list(self.properties.keys())
+
+        product_name = None
+        try:
+            product_name = self.names.product_name
+        except ValueError:
+            ...
+
+        def format_list(items: List, max_len=60):
+            s = ", ".join(sorted(items))
+            if len(s) > max_len:
+                return f"{s[:max_len]}..."
+            return s
+
+        return dedent(
+            f"""
+            Assembling {product_name or ''} ({status})
+            - {len(measurements)} measurements: {format_list(measurements)}
+            - {len(properties)} properties: {format_list(properties)}
+            Writing to location: {output_location}
         """
-        read_location = path
-        if relative_to_dataset_location:
-            read_location = documents.resolve_absolute_offset(
-                self._dataset_location
-                or (self._metadata_path and self._metadata_path.parent),
-                path,
+        )
+
+    def __repr__(self):
+        return self.__str__()
+
+    def add_accessory_file(self, *args, **kwargs):
+        # This was renamed to note_accessory_file() for consistency in our method names.
+        warnings.warn(
+            "add_accessory_file() is deprecated, it has been renamed to note_accessory_file() "
+            "(it's identical: was renamed for api consistency)",
+            category=DeprecationWarning,
+        )
+        self.note_accessory_file(*args, **kwargs)
+
+
+def relative_url(base: str, offset: str, allow_absolute=False):
+    """
+    >>> relative_url('file:///tmp/dataset/odc-metadata.yaml', 'file:///tmp/dataset/my-image.tif')
+    'my-image.tif'
+    >>> relative_url('file:///tmp/dataset/odc-metadata.yaml', 'file:///tmp/dataset/images/my-image.tif')
+    'images/my-image.tif'
+    >>> relative_url(
+    ...    'https://example.test/dataset/odc-metadata.yaml',
+    ...    'https://example.test/dataset/images/my-image.tif'
+    ... )
+    'images/my-image.tif'
+    >>> # Outside the base directory
+    >>> relative_url('https://example.test/dataset/odc-metadata.yaml', 'https://example.test/my-image.tif')
+    Traceback (most recent call last):
+    ...
+    ValueError: Absolute paths are not allowed, and file 'https://example.test/my-image.tif' is outside location \
+'https://example.test/dataset/odc-metadata.yaml'
+    >>> # Matching paths, different hosts.
+    >>> relative_url('https://example.test/odc-metadata.yaml', 'https://example2.test/my-image.tif')
+    Traceback (most recent call last):
+      ...
+    ValueError: Absolute paths are not allowed, and file 'https://example2.test/my-image.tif' is outside location \
+'https://example.test/odc-metadata.yaml'
+    """
+    base_parts = urlsplit(base)
+    offset_parts = urlsplit(offset)
+    if not allow_absolute:
+        if (base_parts.hostname, base_parts.scheme) != (
+            offset_parts.hostname,
+            offset_parts.scheme,
+        ):
+            raise ValueError(
+                f"Absolute paths are not allowed, and file {offset!r} is outside location {base!r}"
             )
-        with rasterio.open(read_location) as ds:
-            ds: DatasetReader
-            if ds.count != 1:
-                raise NotImplementedError(
-                    "TODO: Only single-band files currently supported"
-                )
 
-            self._measurements.record_image(
-                name,
-                images.GridSpec.from_rio(ds),
-                path,
-                ds.read(1) if expand_valid_data else None,
-                nodata=ds.nodata,
-                expand_valid_data=expand_valid_data,
+    base_dir, _ = base_parts.path.rsplit("/", 1)
+    try:
+        return PosixPath(offset_parts.path).relative_to(base_dir).as_posix()
+    except ValueError:
+        if not allow_absolute:
+            raise ValueError(
+                f"Absolute paths are not allowed, and file {offset!r} is outside location {base!r}"
             )
+        # We can't make it relative, return the absolute.
+        return offset
 
-    def extend_user_metadata(self, section_name: str, doc: Dict[str, Any]):
+
+class DatasetAssembler(DatasetPrepare):
+    """
+    Assemble a package of a dataset, including metadata, writing COG images, thumbnails,
+    checksums etc.
+
+    You may want to use :class:`eodatasets3.DatasetPrepare` if you only need a metadata document.
+    """
+
+    def __init__(
+        self,
+        collection_location: Optional[Path] = None,
+        *,
+        dataset_location: Optional[Location] = None,
+        metadata_path: Optional[Path] = None,
+        dataset_id: Optional[uuid.UUID] = None,
+        # By default, we complain if the output already exists.
+        if_exists: IfExists = IfExists.ThrowError,
+        allow_absolute_paths: bool = False,
+        naming_conventions: str = "default",
+        names: Optional[NameGenerator] = None,
+        dataset: Optional[DatasetDoc] = None,
+    ) -> None:
         """
-        Record extra metadata from the processing of the dataset.
+        Assemble a dataset with ODC metadata, writing metadata and (optionally) its imagery as COGs.
 
-        It can be any document suitable for yaml/json serialisation, and will be written into
-        the sidecar "proc-info" metadata.
+        In addition to the below documented methods, metadata can read and set using
+        :class:`Eo3Interface's <eodatasets3.properties.Eo3Interface>` fields.
 
-        This is typically used for recording processing parameters or environment information.
+        There are three optional paths that can be specified. At least one must be specified. Collection,
+        dataset or metadata path.
 
-        :param section_name: Should be unique to your product, and identify the kind of document,
-                             eg 'brdf_ancillary'
-        :param doc: Document
+        - A *collection path* is the root folder where datasets will live (in sub-[sub]-folders).
+        - Each dataset has its own *dataset location*, as stored in an Open Data Cube index.
+          All paths inside the metadata document are relative to this location.
+        - An output *metadata document location*.
+
+        If you're writing data, you typically only need to specify the collection path, and the others
+        will be automatically generated using the naming conventions.
+
+        If you're only writing a metadata file (for existing data), you only need to specify a metadata path.
+
+        If you're storing data using an exotic URI schema, such as a 'tar://' URL path, you will need to specify
+        this as your dataset location.
+
+        :param collection_location:
+            Optional base directory where the collection of datasets should live. Subfolders will be
+            created accordion to the naming convention.
+        :param dataset_location:
+            Optional location for this specific dataset. Otherwise it will be generated according to the collection
+            path and naming conventions.
+        :param metadata_path:
+            Optional metadata document output path. Otherwise it will be generated according to the collection path
+            and naming conventions.
+        :param dataset_id:
+            Optional UUID for this dataset, otherwise a random one will be created. Use this if you have a stable
+            way of generating your own IDs.
+        :param if_exists:
+            What to do if the output dataset already exists? By default, throw an error.
+        :param allow_absolute_paths:
+            Allow metadata paths to refer to files outside the dataset location. this means they will have to be
+            absolute paths, and not be portable. (default: False)
+        :param naming_conventions:
+            Naming conventions to use. Supports `default` or `dea`. The latter has stricter metadata requirements
+            (try it and see -- it will tell your what's missing).
         """
-        if section_name in self._user_metadata:
-            raise ValueError(f"metadata section {section_name} already exists")
+        self._exists_behaviour = if_exists
+        self._checksum = PackageChecksum()
+        self._tmp_work_path: Optional[Path] = None
 
-        self._user_metadata[section_name] = deepcopy(doc)
+        self._user_metadata = {}
+        self._software_versions: List[Dict] = []
 
-    def note_software_version(self, name: str, url: str, version: str):
+        super().__init__(
+            collection_location,
+            dataset_location=dataset_location,
+            metadata_path=metadata_path,
+            dataset_id=dataset_id,
+            allow_absolute_paths=allow_absolute_paths,
+            naming_conventions=naming_conventions,
+            names=names,
+            dataset=dataset,
+        )
+
+    def _target_collection_path(self) -> Path:
+        collection = self.names.collection_path
+        if not collection:
+            raise ValueError(
+                "Dataset assembler was not given a local collection path on construction: cannot write new files."
+            )
+        return collection
+
+    @property
+    def _work_path(self) -> Path:
         """
-        Record the version of some software used to produce the dataset.
+        Require a folder for writing files into the partially-built dataset.
 
-        :param name: a short human-readable name for the software. eg "datacube-core"
-        :param url: A URL where the software is found, such as the git repository.
-        :param version: the version string, eg. "1.0.0b1"
+        The first time this is called, it becomes a packaged dataset rather than a metadata-file-
+        writer only.
         """
-        for v in self._software_versions:
-            # Uniquely identify software by the tuple (name, url)
-            if v["name"] == name and v["url"] == url:
-                existing_version = v["version"]
-                if existing_version != version:
-                    raise ValueError(
-                        f"duplicate setting of software {url!r} with different value "
-                        f"({existing_version!r} != {version!r})"
-                    )
-                return
+        if not self._tmp_work_path:
+            self._tmp_work_path = Path(
+                tempfile.mkdtemp(
+                    prefix=".odcdataset-", dir=self._target_collection_path()
+                )
+            )
 
-        self._software_versions.append(dict(name=name, url=url, version=version))
+        return self._tmp_work_path
 
-    def done(
-        self, validate_correctness: bool = True, sort_measurements: bool = True
-    ) -> Tuple[uuid.UUID, Path]:
+    def __enter__(self) -> "DatasetAssembler":
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        # Clean up.
+        self.close()
+
+    def cancel(self):
         """
-        Write the dataset and move it into place.
+        Cancel the package, cleaning up temporary files.
 
-        It will be validated, metadata will be written, and if all is correct, it will be
-        moved to the output location.
+        This works like :meth:`.close`, but is intentional, so no warning will
+        be raised for forgetting to complete the package first.
+        """
+        self._is_completed = True
+        self.close()
 
-        The final move is done atomically, so the dataset will only exist in the output
-        location if it is complete.
+    def close(self):
+        """Clean up any temporary files, even if dataset has not been written"""
+        if not self._is_completed:
+            warnings.warn(
+                "Closing assembler without finishing. "
+                "Either call `done()` or `cancel() before closing`"
+            )
 
-        :param validate_correctness: Run the eo3-validator on the resulting metadata.
-        :param sort_measurements: Order measurements alphabetically. (instead of insert-order)
-        :raises: :class:`IncompleteDatasetError` If any critical metadata is incomplete.
+        if self._tmp_work_path:
+            # TODO: add implicit cleanup like tempfile.TemporaryDirectory?
+            shutil.rmtree(self._tmp_work_path, ignore_errors=True)
 
-        :returns: The id and final path to the dataset metadata file.
+    def write_measurement(
+        self,
+        name: str,
+        input_path: Location,
+        overviews: Iterable[int] = images.DEFAULT_OVERVIEWS,
+        overview_resampling: Resampling = Resampling.average,
+        expand_valid_data: bool = True,
+        file_id: str = None,
+        path: Path = None,
+    ):
         """
-        self.note_software_version(
-            "eodatasets3",
-            "https://github.com/GeoscienceAustralia/eo-datasets",
-            eodatasets3.__version__,
+        Write a measurement by copying it from a file path.
+
+        Assumes the file is gdal-readable.
+
+        :param name: Identifier for the measurement eg ``'blue'``.
+        :param input_path: The image to read
+        :param overviews: Set of overview sizes to write
+        :param overview_resampling: rasterio Resampling method to use
+        :param expand_valid_data: Include this measurement in the valid-data geometry of the metadata.
+        :param file_id: Optionally, how to identify this in the filename instead of using the name.
+                        (DEA has measurements called ``blue``, but their written filenames must be ``band04`` by
+                        convention.)
+        :param path: Optional path to the image to write. Can be relative to the dataset.
+        """
+        with rasterio.open(input_path) as ds:
+            self.write_measurement_rio(
+                name,
+                ds,
+                overviews=overviews,
+                expand_valid_data=expand_valid_data,
+                overview_resampling=overview_resampling,
+                file_id=file_id,
+                path=path,
+            )
+
+    def write_measurement_rio(
+        self,
+        name: str,
+        ds: DatasetReader,
+        overviews=images.DEFAULT_OVERVIEWS,
+        overview_resampling=Resampling.average,
+        expand_valid_data=True,
+        file_id=None,
+        path: Path = None,
+    ):
+        """
+        Write a measurement by reading it from an open rasterio dataset
+
+        :param ds: An open rasterio dataset
+
+        See :func:`write_measurement` for other parameters.
+        """
+        if len(ds.indexes) != 1:
+            raise NotImplementedError(
+                f"TODO: Multi-band images not currently implemented (have {len(ds.indexes)})"
+            )
+
+        self._write_measurement(
+            name,
+            ds.read(1),
+            images.GridSpec.from_rio(ds),
+            self._work_path
+            / (path or self.names.measurement_filename(name, "tif", file_id=file_id)),
+            expand_valid_data=expand_valid_data,
+            nodata=ds.nodata,
+            overview_resampling=overview_resampling,
+            overviews=overviews,
         )
 
-        crs, grid_docs, measurement_docs = self._measurements.as_geo_docs()
+    def write_measurement_numpy(
+        self,
+        name: str,
+        array: numpy.ndarray,
+        grid_spec: GridSpec,
+        nodata: Optional[Union[float, int]] = None,
+        overviews=images.DEFAULT_OVERVIEWS,
+        overview_resampling=Resampling.average,
+        expand_valid_data=True,
+        file_id: str = None,
+        path: Path = None,
+    ):
+        """
+        Write a measurement from a numpy array and grid spec.
 
-        if measurement_docs and sort_measurements:
-            measurement_docs = dict(sorted(measurement_docs.items()))
+        The most common case is to copy the grid spec from your input dataset,
+        assuming you haven't reprojected.
 
-        if self._inherited_geometry:
-            valid_data = self._inherited_geometry
-        else:
-            valid_data = self._measurements.consume_and_get_valid_data()
-        # Avoid the messiness of different empty collection types.
-        # (to have a non-null geometry we'd also need non-null grids and crses)
-        if valid_data.is_empty:
-            valid_data = None
+        Example::
 
-        if self._is_writing_files():
-            # (the checksum isn't written yet -- it'll be the last file)
-            self.add_accessory_file(
-                "checksum:sha1", self.names.checksum_path(self._work_path)
+            p.write_measurement_numpy(
+                "blue",
+                new_array,
+                GridSpec.from_dataset_doc(source_dataset),
+                nodata=-999,
             )
 
-            processing_metadata = self.names.metadata_path(
-                self._work_path, suffix="proc-info.yaml"
-            )
-            self._write_yaml(
-                {**self._user_metadata, "software_versions": self._software_versions},
-                processing_metadata,
-                allow_external_paths=True,
-            )
-            self.add_accessory_file("metadata:processor", processing_metadata)
-
-        dataset = DatasetDoc(
-            id=self.dataset_id,
-            label=self.label,
-            product=ProductDoc(
-                name=self.names.product_name, href=self.names.product_uri
-            ),
-            crs=self._crs_str(crs) if crs is not None else None,
-            geometry=valid_data,
-            grids=grid_docs,
-            properties=self.properties,
-            accessories={
-                name: AccessoryDoc(path, name=name)
-                for name, path in self._accessories.items()
-            },
-            measurements=measurement_docs,
-            lineage=self._lineage,
-        )
+        See :func:`write_measurement` for other parameters.
 
-        doc = serialise.to_formatted_doc(dataset)
-        self._write_yaml(
-            doc,
-            self._metadata_path
-            or self.names.metadata_path(self._work_path, suffix="odc-metadata.yaml"),
+        :param array:
+        :param grid_spec:
+        :param nodata:
+        """
+        self._write_measurement(
+            name,
+            array,
+            grid_spec,
+            self._work_path
+            / (path or self.names.measurement_filename(name, "tif", file_id=file_id)),
+            expand_valid_data=expand_valid_data,
+            nodata=nodata,
+            overview_resampling=overview_resampling,
+            overviews=overviews,
         )
 
-        if validate_correctness:
-            for m in validate.validate_dataset(doc):
-                if m.level in (Level.info, Level.warning):
-                    warnings.warn(DatasetCompletenessWarning(m))
-                elif m.level == Level.error:
-                    raise IncompleteDatasetError(m)
-                else:
-                    raise RuntimeError(
-                        f"Internal error: Unhandled type of message level: {m.level}"
-                    )
+    def write_measurements_odc_xarray(
+        self,
+        dataset: xarray.Dataset,
+        nodata: Optional[Union[float, int]] = None,
+        overviews=images.DEFAULT_OVERVIEWS,
+        overview_resampling=Resampling.average,
+        expand_valid_data=True,
+        file_id=None,
+    ):
+        """
+        Write measurements from an ODC :class:`xarray.Dataset`
 
-        # If we're writing data, not just a metadata file, finish the package and move it into place.
-        if self._is_writing_files():
-            self._checksum.write(self._accessories["checksum:sha1"])
+        The main requirement is that the Dataset contains a CRS attribute
+        and X/Y or lat/long dimensions and coordinates. These are used to
+        create an ODC GeoBox.
 
-            # Match the lower r/w permission bits to the output folder.
-            # (Temp directories default to 700 otherwise.)
-            self._work_path.chmod(self.collection_location.stat().st_mode & 0o777)
+        :param xarray.Dataset dataset: an xarray dataset (as returned by
+                :meth:`datacube.Datacube.load` and other methods)
 
-            # GDAL writes extra metadata in aux files,
-            # but we consider it a mistake if you're using those extensions.
-            for aux_file in self._work_path.rglob("*.aux.xml"):
-                warnings.warn(
-                    f"Cleaning unexpected gdal aux file {aux_file.as_posix()!r}"
-                )
-                aux_file.unlink()
+        See :meth:`write_measurement` for other parameters.
+        """
+        grid_spec = images.GridSpec.from_odc_xarray(dataset)
+        for name, dataarray in dataset.data_vars.items():
+            name: str
+            self._write_measurement(
+                name,
+                dataarray.data,
+                grid_spec,
+                (
+                    self._work_path
+                    / self.names.measurement_filename(name, "tif", file_id=file_id)
+                ),
+                expand_valid_data=expand_valid_data,
+                overview_resampling=overview_resampling,
+                overviews=overviews,
+                nodata=nodata,
+            )
 
-            if not self._dataset_location:
-                self._dataset_location = self.names.destination_folder(
-                    self.collection_location
-                )
-            # Now atomically move to final location.
-            # Someone else may have created the output while we were working.
-            # Try, and then decide how to handle it if so.
-            try:
-                self._dataset_location.parent.mkdir(parents=True, exist_ok=True)
-                self._work_path.rename(self._dataset_location)
-            except OSError:
-                if not self._dataset_location.exists():
-                    # Some other error?
-                    raise
-
-                if self._exists_behaviour == IfExists.Skip:
-                    # Something else created it while we were busy.
-                    warnings.warn(
-                        f"Skipping -- exists: {self.names.destination_folder}"
-                    )
-                elif self._exists_behaviour == IfExists.ThrowError:
-                    raise
-                elif self._exists_behaviour == IfExists.Overwrite:
-                    raise NotImplementedError("overwriting outputs not yet implemented")
-                else:
-                    raise RuntimeError(
-                        f"Unexpected exists behaviour: {self._exists_behaviour}"
-                    )
+    def _write_measurement(
+        self,
+        name: str,
+        data: numpy.ndarray,
+        grid: GridSpec,
+        out_path: Path,
+        expand_valid_data: bool,
+        nodata: Optional[Union[float, int]],
+        overview_resampling: Resampling,
+        overviews: Tuple[int, ...],
+    ):
+        _validate_property_name(name)
 
-        target_metadata_path = self._metadata_path or self.names.metadata_path(
-            self._dataset_location, suffix="odc-metadata.yaml"
+        res = FileWrite.from_existing(grid.shape).write_from_ndarray(
+            data,
+            out_path,
+            geobox=grid,
+            nodata=nodata,
+            overview_resampling=overview_resampling,
+            overviews=overviews,
         )
-        assert target_metadata_path.exists()
-        self._is_completed = True
-        return dataset.id, target_metadata_path
 
-    def _crs_str(self, crs: CRS) -> str:
-        # TODO: We should support more authorities here.
-        #       if rasterio>=1.1.7, can use crs.to_authority(), but almost
-        #       everyone is currently on 1.1.6
-        return f"epsg:{crs.to_epsg()}" if crs.is_epsg_code else crs.to_wkt()
+        # Ensure the file_format field is set to what we're writing.
+        file_format = res.file_format.name
+        if "odc:file_format" not in self.properties:
+            self.properties["odc:file_format"] = file_format
 
-    def _document_thumbnail(self, thumb_path, kind=None):
-        self._checksum.add_file(thumb_path)
+        if file_format != self.properties["odc:file_format"]:
+            raise RuntimeError(
+                f"Inconsistent file formats between bands. "
+                f"Was {self.properties['odc:file_format']!r}, now {file_format !r}"
+            )
 
-        accessory_name = "thumbnail"
-        if kind:
-            accessory_name += f":{kind}"
-        self.add_accessory_file(accessory_name, thumb_path)
+        self._measurements.record_image(
+            name,
+            grid,
+            out_path,
+            data,
+            nodata=nodata,
+            expand_valid_data=expand_valid_data,
+        )
+        # We checksum immediately as the file has *just* been written so it may still
+        # be in os/filesystem cache.
+        self._checksum.add_file(out_path)
 
     def write_thumbnail(
         self,
@@ -906,6 +1271,7 @@ def write_thumbnail(
         percentile_stretch: Tuple[int, int] = (2, 98),
         scale_factor: int = 10,
         kind: str = None,
+        path: Optional[Path] = None,
     ):
         """
         Write a thumbnail for the dataset using the given measurements (specified by name) as r/g/b.
@@ -928,7 +1294,9 @@ def write_thumbnail(
         :param resampling: rasterio :class:`rasterio.enums.Resampling` method to use.
         :param static_stretch: Use a static upper/lower value to stretch by instead of dynamic stretch.
         """
-        thumb_path = self.names.thumbnail_name(self._work_path, kind=kind)
+        thumb_path = self._work_path / (
+            path or self.names.thumbnail_filename(kind=kind)
+        )
 
         missing_measurements = {red, green, blue} - set(self.measurements)
         if missing_measurements:
@@ -949,7 +1317,7 @@ def write_thumbnail(
         grid = unique_grids[0]
 
         FileWrite().create_thumbnail(
-            tuple(path for grid, path in rgbs),
+            (rgbs[0][1], rgbs[1][1], rgbs[2][1]),
             thumb_path,
             out_scale=scale_factor,
             resampling=resampling,
@@ -958,7 +1326,7 @@ def write_thumbnail(
             input_geobox=grid,
         )
 
-        self._document_thumbnail(thumb_path, kind)
+        self.note_thumbnail(thumb_path, kind)
 
     def write_thumbnail_singleband(
         self,
@@ -971,16 +1339,30 @@ def write_thumbnail_singleband(
         Write a singleband thumbnail out, taking in an input measurement and
         outputting a JPG with appropriate settings.
 
-        Options are to
-        EITHER
-        Use a bit (int) as the value to scale from black to white to
-        i.e., 0 will be BLACK and bit will be WHITE, with a linear scale between.
-        OR
-        Provide a lookuptable (dict) of int (key) [R, G, B] (value) fields
-        to make the image with.
+        :param measurement: Name of measurement
+        :param kind: If you have multiple thumbnails, you can specify the 'kind' name to distinguish
+                     them (it will be put in the filename).
+                     Eg. GA's ARD has two thumbnails, one of kind ``nbar`` and one of ``nbart``.
+
+        EITHER:
+
+        - Use a bit (int) as the value to scale from black to white to
+          i.e., 0 will be BLACK and bit will be WHITE, with a linear scale between,::
+
+             p.write_thumbnail_singleband("blue", bit=1)
+
+        OR:
+
+        - Provide a lookup_table (dict) of int (key) [R, G, B] (value) fields
+          to make the image with.::
+
+             p.write_thumbnail_singleband(
+                 "blue", lookup_table={1: (0, 0, 255)}
+             )
+
         """
 
-        thumb_path = self.names.thumbnail_name(self._work_path, kind=kind)
+        thumb_path = self._work_path / self.names.thumbnail_filename(kind=kind)
 
         _, image_path = self.measurements.get(measurement, (None, None))
 
@@ -1001,80 +1383,190 @@ def write_thumbnail_singleband(
             lookup_table,
         )
 
-        self._document_thumbnail(thumb_path, kind)
+        self.note_thumbnail(thumb_path, kind)
 
-    def add_accessory_file(self, name: str, path: Location):
+    def extend_user_metadata(self, section_name: str, doc: Dict[str, Any]):
         """
-        Record a reference to an additional file that's part of the dataset, but is
-        not a band/measurement.
+        Record extra metadata from the processing of the dataset.
 
-        Such as non-ODC metadata, thumbnails, checksums, etc. Any included file that
-        is not recorded in the measurements.
+        It can be any document suitable for yaml/json serialisation, and will be written into
+        the sidecar "proc-info" metadata.
 
-        By convention, the name should have prefixes with their category, such as
-        'metadata:' or 'thumbnail:'.
+        This is typically used for recording processing parameters or environment information.
+
+        :param section_name: Should be unique to your product, and identify the kind of document,
+                             eg 'brdf_ancillary'
+        :param doc: Document
+        """
+        if section_name in self._user_metadata:
+            raise ValueError(f"metadata section {section_name} already exists")
 
-        eg. 'metadata:landsat_processor', 'checksum:sha1', 'thumbnail:full'.
+        self._user_metadata[section_name] = deepcopy(doc)
 
-        :param name: identifying name, eg 'metadata:mtl'
-        :param path: local path to file.
+    def note_software_version(self, name: str, url: str, version: str):
         """
-        existing_path = self._accessories.get(name)
-        if existing_path is not None and existing_path != path:
-            raise ValueError(
-                f"Duplicate accessory name {name!r}. "
-                f"New: {path.as_posix()!r}, previous: {existing_path.as_posix()!r}"
-            )
-        self._accessories[name] = path
+        Record the version of some software used to produce the dataset.
+
+        :param name: a short human-readable name for the software. eg "datacube-core"
+        :param url: A URL where the software is found, such as the git repository.
+        :param version: the version string, eg. "1.0.0b1"
+        """
+        for v in self._software_versions:
+            # Uniquely identify software by the tuple (name, url)
+            if v["name"] == name and v["url"] == url:
+                existing_version = v["version"]
+                if existing_version != version:
+                    raise ValueError(
+                        f"duplicate setting of software {url!r} with different value "
+                        f"({existing_version!r} != {version!r})"
+                    )
+                return
+
+        self._software_versions.append(dict(name=name, url=url, version=version))
+
+    def note_accessory_file(self, name: str, path: Location):
+        # No docstring deliberately: it's part of parent class docs.
+        #
+        # We override this to include accessories in our checksums:
+        super().note_accessory_file(name, path)
+        self._checksum.add_file(path)
 
-    def _write_yaml(self, doc, path, allow_external_paths=False):
+    def _write_yaml(self, doc: dict, path: Path, allow_external_paths=False):
         documents.make_paths_relative(
             doc, path.parent, allow_paths_outside_base=allow_external_paths
         )
         serialise.dump_yaml(path, doc)
         self._checksum.add_file(path)
 
-    def iter_measurement_paths(
+    def done(
         self,
-    ) -> Generator[Tuple[GridSpec, str, Path], None, None]:
+        validate_correctness: bool = True,
+        sort_measurements: bool = True,
+        embed_location: Optional[bool] = False,
+    ) -> Tuple[uuid.UUID, Path]:
         """
+        Write the dataset and move it into place.
 
-        *not recommended* - will likely change soon.
+        It will be validated, metadata will be written, and if all is correct, it will be
+        moved to the output location.
 
-        Iterate through the list of measurement names that have been written, and their current (temporary) paths.
+        The final move is done atomically, so the dataset will only exist in the output
+        location if it is complete.
 
-        TODO: Perhaps we want to return a real measurement structure here as it's not very extensible.
+        :param validate_correctness: Run the eo3-validator on the resulting metadata.
+        :param sort_measurements: Order measurements alphabetically. (instead of insert-order)
+        :param embed_location: Include the dataset location in the metadata document?
+                 When 'None', it will automatically do it if the location is different to metadata doc.
+        :raises: :class:`IncompleteDatasetError` If any critical metadata is incomplete.
+
+        :returns: The id and final path to the dataset metadata file.
         """
-        return self._measurements.iter_paths()
+        if (
+            self._tmp_work_path is None
+            and not self._software_versions
+            and not self._checksum
+        ):
+            warnings.warn(
+                "Using DatasetAssembler for pure metadata is deprecated. "
+                "Replace DatasetAssembler() with DatasetPrepare(): "
+                "it should be a drop-in replacement. (unless you're setting fields that were previously"
+                "being ignored).",
+                category=DeprecationWarning,
+            )
+            return super(DatasetAssembler, self).done(
+                embed_location=embed_location,
+                validate_correctness=validate_correctness,
+                sort_measurements=sort_measurements,
+            )
 
-    def __str__(self):
-        status = "written" if self._is_completed else "unfinished"
-        target = (
-            self._metadata_path or self._dataset_location or self.collection_location
+        dataset_location = self.names.dataset_location
+
+        self.note_software_version(
+            "eodatasets3",
+            "https://github.com/GeoscienceAustralia/eo-datasets",
+            eodatasets3.__version__,
         )
-        measurements = list(self._measurements.iter_names())
-        properties = list(self.properties.keys())
 
-        product_name = None
-        try:
-            product_name = self.names.product_name
-        except ValueError:
-            ...
+        tmp_metadata_path = self._work_path / self.names.metadata_file
 
-        def format_list(items: List, max_len=60):
-            s = ", ".join(sorted(items))
-            if len(s) > max_len:
-                return f"{s[:max_len]}..."
-            return s
+        processing_metadata = self._work_path / self.names.metadata_filename(
+            suffix="proc-info.yaml"
+        )
+        self._write_yaml(
+            {**self._user_metadata, "software_versions": self._software_versions},
+            processing_metadata,
+            allow_external_paths=True,
+        )
+        self.note_accessory_file("metadata:processor", processing_metadata)
 
-        return dedent(
-            f"""
-            Assembling {product_name or ''} ({status})
-            - {len(measurements)} measurements: {format_list(measurements)}
-            - {len(properties)} properties: {format_list(properties)}
-            Writing to {target}
-        """
+        # (the checksum isn't written yet -- it'll be the last file)
+        # (we use the super() method as we can't add this to our checksum!)
+        super().note_accessory_file(
+            "checksum:sha1", self._work_path / self.names.checksum_file
         )
 
-    def __repr__(self):
-        return self.__str__()
+        dataset = self.to_dataset_doc(
+            dataset_location=tmp_metadata_path.as_uri(),
+            embed_location=False,
+            validate_correctness=validate_correctness,
+            sort_measurements=sort_measurements,
+        )
+
+        if embed_location is None:
+            # If the dataset location is not the metadata path, then record the location.
+            embed_location = dataset_location != self._target_metadata_path()
+
+        if embed_location:
+            dataset.locations = [dataset_location]
+        else:
+            dataset.locations = None
+
+        self._write_yaml(
+            serialise.to_formatted_doc(dataset),
+            tmp_metadata_path,
+        )
+
+        # If we're writing data, not just a metadata file, finish the package and move it into place.
+        self._checksum.write(
+            self._work_path / dataset.accessories["checksum:sha1"].path
+        )
+
+        # Match the lower r/w permission bits to the output folder.
+        # (Temp directories default to 700 otherwise.)
+        self._work_path.chmod(self._target_collection_path().stat().st_mode & 0o777)
+
+        # GDAL writes extra metadata in aux files,
+        # but we consider it a mistake if you're using those extensions.
+        for aux_file in self._work_path.rglob("*.aux.xml"):
+            warnings.warn(f"Cleaning unexpected gdal aux file {aux_file.as_posix()!r}")
+            aux_file.unlink()
+
+        dataset_folder = self.names.dataset_path.parent
+
+        # Now atomically move to final location.
+        # Someone else may have created the output while we were working.
+        # Try, and then decide how to handle it if so.
+        try:
+            dataset_folder.parent.mkdir(parents=True, exist_ok=True)
+            self._work_path.rename(dataset_folder)
+        except OSError:
+            if not dataset_folder.exists():
+                # Some other error?
+                raise
+
+            if self._exists_behaviour == IfExists.Skip:
+                # Something else created it while we were busy.
+                warnings.warn(f"Skipping -- exists: {self.names.dataset_folder}")
+            elif self._exists_behaviour == IfExists.ThrowError:
+                raise
+            elif self._exists_behaviour == IfExists.Overwrite:
+                raise NotImplementedError("overwriting outputs not yet implemented")
+            else:
+                raise RuntimeError(
+                    f"Unexpected exists behaviour: {self._exists_behaviour}"
+                )
+
+        target_metadata_path = self._target_metadata_path()
+        assert target_metadata_path.exists()
+        self._is_completed = True
+        return dataset.id, target_metadata_path
diff --git a/eodatasets3/documents.py b/eodatasets3/documents.py
index eaa83f2f..f0a26197 100644
--- a/eodatasets3/documents.py
+++ b/eodatasets3/documents.py
@@ -6,11 +6,15 @@
 
 import gzip
 import json
+import os
+import posixpath
 from copy import deepcopy
 from pathlib import Path, PurePath
-from typing import Generator, Dict, Tuple, Optional
+from typing import Generator, Dict, Tuple
+from urllib.parse import urlparse
 
 from boltons import iterutils
+
 from eodatasets3 import serialise
 
 _DOCUMENT_EXTENSIONS = (".yaml", ".yml", ".json")
@@ -229,68 +233,76 @@ def make_paths_relative(
     for doc_path, value in iterutils.research(
         doc, lambda p, k, v: isinstance(v, PurePath)
     ):
-        value: Path
-
-        if value.is_absolute():
-            if base_directory not in value.parents:
-                if not allow_paths_outside_base:
-                    raise ValueError(
-                        f"Path {value.as_posix()!r} is outside path {base_directory.as_posix()!r} "
-                        f"(allow_paths_outside_base={allow_paths_outside_base})"
-                    )
-                continue
-            value = value.relative_to(base_directory)
+        value: PurePath
+        value = relative_path(
+            value, base_directory, allow_paths_outside_base=allow_paths_outside_base
+        )
+        docpath_set(doc, doc_path, value.as_posix())
 
-        docpath_set(doc, doc_path, str(value))
 
+def relative_url(value: str, base: str, allow_paths_outside_base=False) -> str:
+    """
+    Make a single url relative to the base url if it is inside it.
+
+    By default, will throw a ValueError if not able to make it relative to the path.
 
-def resolve_absolute_offset(
-    dataset_path: Path, offset: str, target_path: Optional[Path] = None
-) -> str:
+
+    >>> relative_url('file:///g/data/v10/0/2015/blue.jpg', 'file:///g/data/v10/0/2015/odc-metadata.yaml')
+    'blue.jpg'
+    >>> relative_url('https://example.test/2015/images/blue.jpg', 'https://example.test/2015/odc-metadata.yaml')
+    'images/blue.jpg'
+    >>> relative_url('file:///g/data/v10/0/2018/blue.jpg', 'file:///g/data/v10/0/2015/odc-metadata.yaml')
+    Traceback (most recent call last):
+      ...
+    ValueError: Path 'file:///g/data/v10/0/2018/blue.jpg' is outside path 'file:///g/data/v10/0/2015/odc-metadata.yaml'\
+ (allow_paths_outside_base=False)
     """
-    Expand a filename (offset) relative to the dataset.
-
-    >>> external_metadata_loc = Path('/tmp/target-metadata.yaml')
-    >>> resolve_absolute_offset(
-    ...     Path('/tmp/great_test_dataset'),
-    ...     'band/my_great_band.jpg',
-    ...     external_metadata_loc,
-    ... )
-    '/tmp/great_test_dataset/band/my_great_band.jpg'
-    >>> resolve_absolute_offset(
-    ...     Path('/tmp/great_test_dataset.tar.gz'),
-    ...     'band/my_great_band.jpg',
-    ...     external_metadata_loc,
-    ... )
-    'tar:/tmp/great_test_dataset.tar.gz!band/my_great_band.jpg'
-    >>> resolve_absolute_offset(
-    ...     Path('/tmp/great_test_dataset.tar'),
-    ...     'band/my_great_band.jpg',
-    ... )
-    'tar:/tmp/great_test_dataset.tar!band/my_great_band.jpg'
-    >>> resolve_absolute_offset(
-    ...     Path('/tmp/great_test_dataset.zip'),
-    ...     'band/other/my_great_band.jpg',
-    ... )
-    'zip:/tmp/great_test_dataset.zip!band/other/my_great_band.jpg'
-    >>> resolve_absolute_offset(
-    ...     Path('/tmp/MY_DATASET'),
-    ...     'band/my_great_band.jpg',
-    ...     Path('/tmp/MY_DATASET/ga-metadata.yaml'),
-    ... )
-    'band/my_great_band.jpg'
+
+    if not value:
+        return value
+
+    if not value.startswith(base) and not value.startswith(os.path.dirname(base)):
+        if not allow_paths_outside_base:
+            raise ValueError(
+                f"Path {value!r} is outside path {base!r} "
+                f"(allow_paths_outside_base={allow_paths_outside_base})"
+            )
+        return value
+
+    return _make_relurl(value, base)
+
+
+def _make_relurl(target: str, base: str) -> str:
+    base = urlparse(base)
+    target = urlparse(target)
+    if base.netloc != target.netloc:
+        raise ValueError("target and base netlocs do not match")
+    base_dir = "." + posixpath.dirname(base.path)
+    target = "." + target.path
+    return posixpath.relpath(target, start=base_dir)
+
+
+def relative_path(
+    value: PurePath, base_directory: PurePath, allow_paths_outside_base=False
+) -> PurePath:
+    """
+    Make a single path relative to the base directory if it is inside it.
+
+    By default, will throw a ValueError if not able to make it relative to the path.
+
+    >>> val =  PurePath('/tmp/minimal-pkg/loch_ness_sightings_2019-07-04_blue.tif')
+    >>> base = PurePath('/tmp/minimal-pkg')
+    >>> relative_path(val, base).as_posix()
+    'loch_ness_sightings_2019-07-04_blue.tif'
     """
-    dataset_path = dataset_path.absolute()
-
-    if target_path:
-        # If metadata is stored inside the dataset, keep paths relative.
-        if str(target_path.absolute()).startswith(str(dataset_path)):
-            return offset
-    # Bands are inside a tar file
-
-    if ".tar" in dataset_path.suffixes:
-        return "tar:{}!{}".format(dataset_path, offset)
-    elif ".zip" in dataset_path.suffixes:
-        return "zip:{}!{}".format(dataset_path, offset)
-    else:
-        return str(dataset_path / offset)
+    if not value or not value.is_absolute():
+        return value
+
+    if base_directory not in value.parents:
+        if not allow_paths_outside_base:
+            raise ValueError(
+                f"Path {value.as_posix()!r} is outside path {base_directory.as_posix()!r} "
+                f"(allow_paths_outside_base={allow_paths_outside_base})"
+            )
+        return value
+    return value.relative_to(base_directory)
diff --git a/eodatasets3/images.py b/eodatasets3/images.py
index 99575046..e7619a5a 100644
--- a/eodatasets3/images.py
+++ b/eodatasets3/images.py
@@ -14,7 +14,7 @@
 import xarray
 from affine import Affine
 from collections import defaultdict
-from pathlib import Path
+from pathlib import Path, PurePath
 from rasterio import DatasetReader
 from rasterio.coords import BoundingBox
 from rasterio.crs import CRS
@@ -50,6 +50,17 @@
 @attr.s(auto_attribs=True, slots=True, hash=True, frozen=True)
 class GridSpec:
     """
+    The grid spec defines the coordinates/transform and size of pixels of a
+    measurment.
+
+    The easiest way to create one is use the ``GridSpec.from_*()`` class methods, such as
+    ``GridSpec.from_path(my_image_path)``.
+
+    To create one manually:
+
+    >>> from eodatasets3 import GridSpec
+    >>> from affine import Affine
+    >>> from rasterio.crs import CRS
     >>> g = GridSpec(shape=(7721, 7621),
     ...              transform=Affine(30.0, 0.0, 241485.0, 0.0, -30.0, -2281485.0),
     ...              crs=CRS.from_epsg(32656))
@@ -60,15 +71,22 @@ class GridSpec:
     (30.0, 30.0)
     """
 
+    #:
     shape: Tuple[int, int]
+    #:
     transform: Affine
-
+    #:
     crs: CRS = attr.ib(
         metadata=dict(doc_exclude=True), default=None, hash=False, eq=False
     )
 
     @classmethod
     def from_dataset_doc(cls, ds: DatasetDoc, grid="default") -> "GridSpec":
+        """
+        Create from an existing parsed metadata document
+
+        :param grid: Grid name to read, if not the default.
+        """
         g = ds.grids[grid]
 
         if ds.crs.startswith("epsg:"):
@@ -80,6 +98,7 @@ def from_dataset_doc(cls, ds: DatasetDoc, grid="default") -> "GridSpec":
 
     @classmethod
     def from_rio(cls, dataset: rasterio.DatasetReader) -> "GridSpec":
+        """Create from an open rasterio dataset"""
         return cls(shape=dataset.shape, transform=dataset.transform, crs=dataset.crs)
 
     @property
@@ -88,6 +107,7 @@ def resolution_yx(self):
 
     @classmethod
     def from_odc_xarray(cls, dataset: xarray.Dataset) -> "GridSpec":
+        """Create from an ODC xarray"""
         shape = set(v.shape for v in dataset.data_vars.values()).pop()
         return cls(
             shape=shape,
@@ -95,11 +115,16 @@ def from_odc_xarray(cls, dataset: xarray.Dataset) -> "GridSpec":
             crs=CRS.from_wkt(dataset.geobox.crs.crs_str),
         )
 
+    @classmethod
+    def from_path(cls, path: str) -> "GridSpec":
+        """Create from the spec of a (rio-readable) filesystem path or url"""
+        with rasterio.open(path) as rio:
+            return GridSpec.from_rio(rio)
+
     @property
     def bounds(self):
         """
         Get bounding box.
-
         """
         return BoundingBox(
             *(self.transform * (0, self.shape[0]))
@@ -230,16 +255,17 @@ def _find_a_common_name(
 
 @attr.s(auto_attribs=True, slots=True)
 class _MeasurementLocation:
-    path: Path
+    path: Union[Path, str]
     layer: str = None
 
 
 _Measurements = Dict[str, _MeasurementLocation]
 
 
-class MeasurementRecord:
+class MeasurementBundler:
     """
-    Record the information for measurements/images to later write out to metadata.
+    Incrementally record the information for a set of measurements/images to group into grids,
+    calculate geometry etc, suitable for metadata.
     """
 
     def __init__(self):
@@ -253,7 +279,7 @@ def record_image(
         self,
         name: str,
         grid: GridSpec,
-        path: Union[Path, str],
+        path: Union[PurePath, str],
         img: numpy.ndarray,
         layer: Optional[str] = None,
         nodata: Optional[Union[float, int]] = None,
@@ -815,14 +841,14 @@ def create_thumbnail_singleband(
             if bit is not None:
                 out_data = numpy.copy(data)
                 out_data[data != bit] = 0
-                stretch = [0, bit]
+                stretch = (0, bit)
             if lookup_table is not None:
                 out_data = [
                     numpy.full_like(data, 0),
                     numpy.full_like(data, 0),
                     numpy.full_like(data, 0),
                 ]
-                stretch = [0, 255]
+                stretch = (0, 255)
 
                 for value, rgb in lookup_table.items():
                     for index in range(3):
@@ -839,13 +865,13 @@ def create_thumbnail_singleband(
                 with rasterio.open(temp_file, "w", **meta) as tmpdataset:
                     tmpdataset.write(out_data)
                 self.create_thumbnail(
-                    [temp_file, temp_file, temp_file],
+                    (temp_file, temp_file, temp_file),
                     out_file,
                     static_stretch=stretch,
                 )
             else:
                 # Use three different files
-                temp_files = [Path(temp_dir) / f"temp_{i}.tif" for i in range(3)]
+                temp_files = tuple(Path(temp_dir) / f"temp_{i}.tif" for i in range(3))
 
                 for i in range(3):
                     with rasterio.open(temp_files[i], "w", **meta) as tmpdataset:
diff --git a/eodatasets3/model.py b/eodatasets3/model.py
index 6ba59b34..b5bae792 100644
--- a/eodatasets3/model.py
+++ b/eodatasets3/model.py
@@ -1,7 +1,5 @@
-import re
-from datetime import datetime
 from pathlib import Path
-from typing import Tuple, Dict, Optional, List, Sequence, Union
+from typing import Tuple, Dict, Optional, List, Union
 from uuid import UUID
 
 import affine
@@ -9,12 +7,9 @@
 from ruamel.yaml.comments import CommentedMap
 from shapely.geometry.base import BaseGeometry
 
-from eodatasets3 import utils
-from eodatasets3.properties import StacPropertyView, EoFields
+from eodatasets3.properties import Eo3Dict, Eo3Interface
 
-# TODO: these need discussion.
 DEA_URI_PREFIX = "https://collections.dea.ga.gov.au"
-DEAFRICA_URI_PREFIX = "https://digitalearthafrica.org"
 ODC_DATASET_SCHEMA_URL = "https://schemas.opendatacube.org/dataset"
 
 # Either a local filesystem path or a string URI.
@@ -22,10 +17,6 @@
 Location = Union[Path, str]
 
 
-def _dea_uri(product_name, base_uri):
-    return f"{base_uri}/product/{product_name}"
-
-
 @attr.s(auto_attribs=True, slots=True)
 class ProductDoc:
     """
@@ -39,10 +30,6 @@ class ProductDoc:
     name: str = None
     href: str = None
 
-    @classmethod
-    def dea_name(cls, name: str):
-        return ProductDoc(name=name, href=_dea_uri(name, DEA_URI_PREFIX))
-
 
 @attr.s(auto_attribs=True, slots=True, hash=True)
 class GridDoc:
@@ -81,576 +68,54 @@ class AccessoryDoc:
     name: str = attr.ib(metadata=dict(doc_exclude=True), default=None)
 
 
-class ComplicatedNamingConventions:
-    """
-    Naming conventions based on the DEA standard.
-
-    Unlike the DEA standard, almost every field is optional by default.
-    """
-
-    _ABSOLUTE_MINIMAL_PROPERTIES = {
-        "odc:product_family",
-        # Required by Stac regardless.
-        "datetime",
-    }
-
-    # Displayed to user for friendlier errors.
-    _REQUIRED_PROPERTY_HINTS = {
-        "odc:product_family": 'eg. "wofs" or "level1"',
-        "odc:processing_datetime": "Time of processing, perhaps datetime.utcnow()?",
-        "odc:producer": "Creator of data, eg 'usgs.gov' or 'ga.gov.au'",
-        "odc:dataset_version": "eg. 1.0.0",
-    }
-
-    KNOWN_PRODUCER_ABBREVIATIONS = {
-        "ga.gov.au": "ga",
-        "usgs.gov": "usgs",
-        "sinergise.com": "sinergise",
-        "digitalearthafrica.org": "deafrica",
-        "esa.int": "esa",
-        # Is there another organisation you want to use? Pull requests very welcome!
-    }
-
-    # The abbreviations mentioned in DEA naming conventions doc.
-    KNOWN_PLATFORM_ABBREVIATIONS = {
-        "landsat-5": "ls5",
-        "landsat-7": "ls7",
-        "landsat-8": "ls8",
-        "landsat-9": "ls9",
-        "sentinel-1a": "s1a",
-        "sentinel-1b": "s1b",
-        "sentinel-2a": "s2a",
-        "sentinel-2b": "s2b",
-        "aqua": "aqu",
-        "terra": "ter",
-    }
-
-    # If all platforms match a pattern, return this group name instead.
-    KNOWN_PLATFORM_GROUPINGS = {
-        "ls": re.compile(r"ls\d+"),
-        "s1": re.compile(r"s1[a-z]+"),
-        "s2": re.compile(r"s2[a-z]+"),
-    }
-
-    def __init__(
-        self,
-        dataset: EoFields,
-        base_product_uri: str = None,
-        required_fields: Sequence[str] = (),
-        dataset_separator_field: Optional[str] = None,
-        allow_unknown_abbreviations: bool = True,
-    ) -> None:
-        self.dataset = dataset
-        self.base_product_uri = base_product_uri
-        self.required_fields = self._ABSOLUTE_MINIMAL_PROPERTIES.union(required_fields)
-
-        # An extra folder to put each dataset inside, using the value of the given property name.
-        self.dataset_separator_field = dataset_separator_field
-
-        if self.dataset_separator_field is not None:
-            self.required_fields.add(dataset_separator_field)
-
-        self.allow_unknown_abbreviations = allow_unknown_abbreviations
-
-    @classmethod
-    def for_standard_dea(cls, dataset: EoFields, uri=DEA_URI_PREFIX):
-        """
-        Strict mode to follow the full DEA naming conventions.
-
-        Only use the (default) DEA URI if you're making DEA products.
-
-        Example file structure (note version number in file):
-            ga_ls8c_ones_3/090/084/2016/01/21/ga_ls8c_ones_3-0-0_090084_2016-01-21_final.odc-metadata.yaml
-        """
-        return cls(
-            dataset=dataset,
-            base_product_uri=uri,
-            # These fields are needed to fulfill official DEA naming conventions.
-            required_fields=(
-                "eo:platform",
-                "eo:instrument",
-                "odc:processing_datetime",
-                "odc:producer",
-                "odc:product_family",
-                "odc:region_code",
-                "odc:dataset_version",
-            ),
-            # DEA wants consistency via the naming-conventions doc.
-            allow_unknown_abbreviations=False,
-        )
-
-    @classmethod
-    def for_standard_dea_s2(cls, dataset: EoFields, uri=DEA_URI_PREFIX):
-        """
-        DEA naming conventions, but with an extra subfolder for each unique datatake.
-
-        It will figure out the datatake if you set a sentinel_tile_id or datastrip_id.
-        """
-        return cls(
-            dataset=dataset,
-            base_product_uri=uri,
-            # These fields are needed to fulfill official DEA naming conventions.
-            required_fields=(
-                "eo:instrument",
-                "eo:platform",
-                "odc:dataset_version",
-                "odc:processing_datetime",
-                "odc:producer",
-                "odc:product_family",
-                "odc:region_code",
-            ),
-            dataset_separator_field="sentinel:datatake_start_datetime",
-            # DEA wants consistency via the naming-conventions doc.
-            allow_unknown_abbreviations=False,
-        )
-
-    def _check_enough_properties_to_name(self):
-        """
-        Do we have enough properties to generate file or product names?
-        """
-        missing_props = []
-        for f in self.required_fields:
-            if f not in self.dataset.properties:
-                missing_props.append(f)
-        if missing_props:
-            examples = []
-            for p in sorted(missing_props):
-                hint = self._REQUIRED_PROPERTY_HINTS.get(p, "")
-                if hint:
-                    hint = f" ({hint})"
-                examples.append(f"\n- {p!r}{hint}")
-
-            raise ValueError(
-                f"Need more properties to fulfill naming conventions."
-                f"{''.join(examples)}"
-            )
-
-    @property
-    def product_name(self) -> str:
-        if self.dataset.product_name:
-            return self.dataset.product_name
-
-        self._check_enough_properties_to_name()
-
-        org_number = self._org_collection_number
-        if org_number:
-            return f"{self._product_group()}_{org_number}"
-        return self._product_group()
-
-    @property
-    def _org_collection_number(self) -> Optional[int]:
-        # An explicit collection number trumps all.
-        if self.dataset.collection_number:
-            return int(self.dataset.collection_number)
-
-        # Otherwise it's the first digit of the dataset version.
-        if not self.dataset.dataset_version:
-            return None
-        return int(self.dataset.dataset_version.split(".")[0])
-
-    def _product_group(self, subname=None) -> str:
-        parts = []
-
-        # If they've given a product name, just use it.
-        if self.dataset.product_name:
-            parts.append(self.dataset.product_name)
-            if subname:
-                parts.append(subname)
-        else:
-            self._check_enough_properties_to_name()
-
-            # They're not specifying a sub-file. Fallback to the whole product's category.
-            if not subname:
-                subname = self.dataset.product_family
-
-            if self.producer_abbreviated:
-                parts.append(self.producer_abbreviated)
-
-            platform = self.platform_abbreviated
-            inst = self.instrument_abbreviated or ""
-            if platform:
-                parts.append(f"{platform}{inst}")
-
-            if not subname:
-                raise ValueError(
-                    "Not enough metadata to create a useful filename! "
-                    'Set the `product_family` (eg. "wofs") or a subname'
-                )
-            parts.append(subname)
-
-        return "_".join(parts)
-
-    @property
-    def product_uri(self) -> Optional[str]:
-        if not self.base_product_uri:
-            return None
-
-        return _dea_uri(self.product_name, base_uri=self.base_product_uri)
-
-    @property
-    def dataset_label(self) -> str:
-        """
-        Label for a dataset
-        """
-        return self._dataset_label()
-
-    def destination_folder(self, base: Path) -> Path:
-        # DEA naming conventions folder hierarchy.
-        # Example: "ga_ls8c_ard_3/092/084/2016/06/28"
-
-        parts = [self.product_name]
-
-        # Cut the region code in subfolders
-        region_code = self.dataset.region_code
-        if region_code:
-            parts.extend(utils.subfolderise(region_code))
-
-        parts.extend(f"{self.dataset.datetime:%Y/%m/%d}".split("/"))
-
-        # If it's not a final product, append the maturity to the folder.
-        maturity: str = self.dataset.properties.get("dea:dataset_maturity")
-        if maturity and maturity != "final":
-            parts[-1] = f"{parts[-1]}_{maturity}"
-
-        if self.dataset_separator_field is not None:
-            val = self.dataset.properties[self.dataset_separator_field]
-            # TODO: choosable formatter?
-            if isinstance(val, datetime):
-                val = f"{val:%Y%m%dT%H%M%S}"
-            parts.append(val)
-        return base.joinpath(*parts)
-
-    def metadata_path(self, work_dir: Path, kind: str = "", suffix: str = "yaml"):
-        return self._file(work_dir, kind, suffix)
-
-    def checksum_path(self, work_dir: Path, suffix: str = "sha1"):
-        return self._file(work_dir, "", suffix)
-
-    def measurement_file_path(
-        self, work_dir: Path, measurement_name: str, suffix: str, file_id: str = None
-    ) -> Path:
-        if ":" in measurement_name:
-            subgroup, name = measurement_name.split(":")
-        else:
-            subgroup, name = None, measurement_name
-
-        return self._file(
-            work_dir,
-            # We use 'band01'/etc in the filename if provided, rather than 'red'
-            file_id or name,
-            suffix,
-            sub_name=subgroup,
-        )
-
-    def _dataset_label(self, sub_name: str = None):
-        p = self.dataset
-
-        version = p.dataset_version.replace(".", "-") if p.dataset_version else None
-        maturity: str = p.properties.get("dea:dataset_maturity")
-        return "_".join(
-            [
-                p
-                for p in (
-                    self._product_group(sub_name),
-                    version,
-                    self._displayable_region_code,
-                    f"{p.datetime:%Y-%m-%d}",
-                    maturity,
-                )
-                if p
-            ]
-        )
-
-    def _file(self, work_dir: Path, file_id: str, suffix: str, sub_name: str = None):
-        file_id = "_" + file_id.replace("_", "-") if file_id else ""
-
-        return work_dir / (
-            f"{self._dataset_label(sub_name=sub_name)}{file_id}.{suffix}"
-        )
-
-    @property
-    def _displayable_region_code(self):
-        return self.dataset.region_code
-
-    def thumbnail_name(self, work_dir: Path, kind: str = None, suffix: str = "jpg"):
-        self._check_enough_properties_to_name()
-        if kind:
-            name = f"{kind}:thumbnail"
-        else:
-            name = "thumbnail"
-        return self.measurement_file_path(work_dir, name, suffix)
-
-    @property
-    def platform_abbreviated(self) -> Optional[str]:
-        """Abbreviated form of a satellite, as used in dea product names. eg. 'ls7'."""
-
-        p = self.dataset.platforms
-        if not p:
-            return None
-
-        if not self.allow_unknown_abbreviations:
-            unknowns = p.difference(self.KNOWN_PLATFORM_ABBREVIATIONS)
-            if unknowns:
-                raise ValueError(
-                    f"We don't know the DEA abbreviation for platforms {unknowns!r}. "
-                    f"We'd love to add more! Raise an issue on Github: "
-                    f"https://github.com/GeoscienceAustralia/eo-datasets/issues/new' "
-                )
-
-        abbreviations = sorted(
-            self.KNOWN_PLATFORM_ABBREVIATIONS.get(s, s.replace("-", "")) for s in p
-        )
-
-        if len(abbreviations) == 1:
-            return abbreviations[0]
-
-        # If all abbreviations are in a group, name it using that group.
-        # (eg. "ls" instead of "ls5-ls7-ls8")
-        for group_name, pattern in self.KNOWN_PLATFORM_GROUPINGS.items():
-            if all(pattern.match(a) for a in abbreviations):
-                return group_name
-
-        # Otherwise, there's a mix of platforms.
-
-        # Is there a common constellation?
-        constellation = self.dataset.properties.get("constellation")
-        if constellation:
-            return constellation
-
-        # Don't bother to include platform in name for un-groupable mixes of them.
-        return None
-
-    @property
-    def instrument_abbreviated(self) -> Optional[str]:
-        """Abbreviated form of an instrument name, as used in dea product names. eg. 'c'."""
-        platforms = self.dataset.platforms
-        if not platforms or len(platforms) > 1:
-            return None
-
-        [p] = platforms
-
-        if p.startswith("sentinel-1") or p.startswith("sentinel-2"):
-            return self.dataset.instrument[0].lower()
-
-        if p.startswith("landsat"):
-            # Extract from usgs standard:
-            # landsat:landsat_product_id: LC08_L1TP_091075_20161213_20170316_01_T2
-            # landsat:landsat_scene_id: LC80910752016348LGN01
-            landsat_id = self.dataset.properties.get("landsat:landsat_product_id")
-            if landsat_id is None:
-                landsat_id = self.dataset.properties.get("landsat:landsat_scene_id")
-
-            # from USGS STAC, label is LC08_L2SP_178079_20210417_20210424_02_T1_SR and
-            # landsat:scene_id: LC81780792021107LGN00
-            if landsat_id is None:
-                landsat_id = self.dataset.properties.get("landsat:scene_id")
-
-            if not landsat_id:
-                raise NotImplementedError(
-                    "TODO: Can only currently abbreviate instruments from Landsat references."
-                )
-
-            return landsat_id[1].lower()
-
-        # Otherwise, it's unknown.
-        raise NotImplementedError(
-            f"Instrument abbreviations aren't supported for platform {p!r}. "
-            f"We'd love to add more support! Raise an issue on Github: "
-            f"https://github.com/GeoscienceAustralia/eo-datasets/issues/new' "
-        )
-
-    @property
-    def producer_abbreviated(self) -> Optional[str]:
-        """Abbreviated form of a producer, as used in dea product names. eg. 'ga', 'usgs'."""
-        if not self.dataset.producer:
-            return None
-
-        try:
-            return self.KNOWN_PRODUCER_ABBREVIATIONS[self.dataset.producer]
-        except KeyError:
-            raise NotImplementedError(
-                f"We don't know how to abbreviate organisation domain name {self.dataset.producer!r}. "
-                f"We'd love to add more orgs! Raise an issue on Github: "
-                f"https://github.com/GeoscienceAustralia/eo-datasets/issues/new' "
-            )
-
-
-class ComplicatedNamingConventionsDerivatives(ComplicatedNamingConventions):
+@attr.s(auto_attribs=True, slots=True)
+class DatasetDoc(Eo3Interface):
     """
-    Derivatives have a slightly different folder structure.
+    An EO3 dataset document
 
-    And they only show constellations (eg. "ls_" or "s2_") rather than the specific
-    satellites in their names (eg. "ls8_").
+    Includes :class:`.Eo3Interface` methods for metadata access::
 
-    They have a version-number folder instead of putting it in each filename.
+        >>> p = DatasetDoc()
+        >>> p.platform = 'LANDSAT_8'
+        >>> p.processed = '2018-04-03'
+        >>> p.properties['odc:processing_datetime']
+        datetime.datetime(2018, 4, 3, 0, 0, tzinfo=datetime.timezone.utc)
 
-    And version numbers may not match the collection number (`odc:collection_number` is
-    mandatory).
     """
 
-    @classmethod
-    def for_c3_derivatives(cls, dataset: EoFields, uri=DEA_URI_PREFIX):
-        """
-        Create naming conventions for common derived products.
-
-        Unlike plain 'DEA', they use an explicit collection number (odc:collection_number)
-        in the product name which may differ from the software's dataset version
-        (odc:dataset_version)
-
-        Example file structure (note version number in folder):
-
-            ga_ls_wo_3/1-6-0/090/081/1998/07/30/ga_ls_wo_3_090081_1998-07-30_interim.odc-metadata.yaml
-
-
-        """
-        return cls(
-            dataset=dataset,
-            base_product_uri=uri,
-            required_fields=(
-                "eo:platform",
-                "odc:dataset_version",
-                "odc:collection_number",
-                "odc:processing_datetime",
-                "odc:producer",
-                "odc:product_family",
-                "odc:region_code",
-                "odc:dataset_version",
-                "dea:dataset_maturity",
-            ),
-        )
-
-    @classmethod
-    def for_s2_derivatives(cls, dataset: EoFields, uri=DEA_URI_PREFIX):
-        """
-        DEA derivative naming conventions, but with an extra subfolder for
-        each unique datatake.
-
-        It will figure out the datatake if you set a sentinel_tile_id
-        or datastrip_id.
-        """
-        return cls(
-            dataset=dataset,
-            base_product_uri=uri,
-            # These fields are needed to fulfill official DEA naming conventions.
-            required_fields=(
-                "eo:platform",
-                "odc:dataset_version",
-                "odc:collection_number",
-                "odc:processing_datetime",
-                "odc:producer",
-                "odc:product_family",
-                "odc:region_code",
-                "dea:dataset_maturity",
-            ),
-            dataset_separator_field="sentinel:datatake_start_datetime",
-        )
-
-    @classmethod
-    def for_deafrica_derivatives(cls, dataset: EoFields, uri=DEAFRICA_URI_PREFIX):
-        """
-        DEAFRICA USGS C2 Naming Convention
-        """
-        return cls(
-            dataset=dataset,
-            base_product_uri=uri,
-            required_fields=(
-                "eo:platform",
-                "odc:producer",
-                "odc:region_code",
-                "odc:product_family",
-                "odc:dataset_version",
-            ),
-        )
-
-    def _product_group(self, subname=None) -> str:
-        # Computes product group, e.g "ga_ls_wo_3"
-        # Deliberately fail if any of these attributes not found.
-        parts = [
-            self.producer_abbreviated,
-            self.platform_abbreviated,
-            self.dataset.product_family,
-        ]
-
-        # Exceptional case for DE Africa. There must be a more elegant
-        # way to do this...
-        if self.producer_abbreviated == "deafrica":
-            parts = [self.dataset.product_family, self.platform_abbreviated]
-
-        return "_".join(parts)
-
-    def destination_folder(self, base: Path) -> Path:
-        self._check_enough_properties_to_name()
-        parts = [self.product_name, self.dataset.dataset_version.replace(".", "-")]
-        parts.extend(utils.subfolderise(self.dataset.region_code))
-        parts.extend(f"{self.dataset.datetime:%Y/%m/%d}".split("/"))
-
-        if self.dataset_separator_field is not None:
-            val = self.dataset.properties[self.dataset_separator_field]
-            # TODO: choosable formatter?
-            if isinstance(val, datetime):
-                val = f"{val:%Y%m%dT%H%M%S}"
-            parts.append(val)
-
-        return base.joinpath(*parts)
-
-    def _dataset_label(self, sub_name: str = None):
-        """
-        Responsible for producing the string of product name, regioncode, datetime and maturity
-        ex: 'ga_ls_wo_3_090081_1998-07-30_interim'
-
-        Redundant parameter sub_name is required, since the parent class and other invocations wants it so.
-        """
-        parts = [
-            self.product_name,
-            self._displayable_region_code,
-            f"{self.dataset.datetime:%Y-%m-%d}",
-            self.dataset.maturity,
-        ]
-        parts = [x for x in parts if x is not None]
-        return "_".join(parts)
-
-    @property
-    def platform_abbreviated(self) -> Optional[str]:
-        """
-        Derivatives only show group/constellation names ("ls" or "s2". Not "ls8".)
-
-        (Because individual datasets come from different platforms, a specific abbreviation
-        would be inconsistent across a product [presumably])
-        """
-        abbreviations = sorted(
-            self.KNOWN_PLATFORM_ABBREVIATIONS.get(s, s.replace("-", ""))
-            for s in self.dataset.platforms
-        )
-        # If all abbreviations are in a group, name it using that group.
-        # (eg. "ls" instead of "ls5-ls7-ls8")
-        for group_name, pattern in self.KNOWN_PLATFORM_GROUPINGS.items():
-            if all(pattern.match(a) for a in abbreviations):
-                return group_name
-
-        raise NotImplementedError(
-            f"Satellite constellation abbreviation is not known for platforms {self.dataset.platforms}. "
-            f"(for DEA derivative naming conventions.)"
-            f"    Is this a mistake? We'd love to add more! Raise an issue on Github: "
-            f"https://github.com/GeoscienceAustralia/eo-datasets/issues/new' "
-        )
-
-
-@attr.s(auto_attribs=True, slots=True)
-class DatasetDoc(EoFields):
+    #: Dataset UUID
     id: UUID = None
+    #: Human-readable identifier for the dataset
     label: str = None
+    #: The product name (local) and/or url (global)
     product: ProductDoc = None
+    #: Location(s) where this dataset is stored.
+    #:
+    #: (ODC supports multiple locations when the same dataset is stored in multiple places)
+    #:
+    #: They are fully qualified URIs (``file://...`, ``https://...``, ``s3://...``)
+    #:
+    #: All other paths in the document (measurements, accessories) are relative to the
+    #: chosen location.
     locations: List[str] = None
 
+    #: CRS string. Eg. ``epsg:3577``
     crs: str = None
+    #: Shapely geometry of the valid data coverage
+    #:
+    #: (it must contain all non-empty pixels of the image)
     geometry: BaseGeometry = None
+    #: Grid specifications for measurements
     grids: Dict[str, GridDoc] = None
-
-    properties: StacPropertyView = attr.ib(factory=StacPropertyView)
-
+    #: Raw properties
+    properties: Eo3Dict = attr.ib(factory=Eo3Dict)
+    #: Loadable measurements of the dataset
     measurements: Dict[str, MeasurementDoc] = None
-
-    # Paths to accessory files, such as thumbnails.
+    #: References to accessory files
+    #:
+    #: Such as thumbnails, checksums, other kinds of metadata files.
+    #:
+    #: (any files included in the dataset that are not measurements)
     accessories: Dict[str, AccessoryDoc] = attr.ib(factory=CommentedMap)
-
-    lineage: Dict[str, Sequence[UUID]] = attr.ib(factory=CommentedMap)
+    #: Links to source dataset uuids
+    lineage: Dict[str, List[UUID]] = attr.ib(factory=CommentedMap)
diff --git a/eodatasets3/names.py b/eodatasets3/names.py
new file mode 100644
index 00000000..e0bcb1e8
--- /dev/null
+++ b/eodatasets3/names.py
@@ -0,0 +1,961 @@
+import re
+from datetime import datetime
+from pathlib import Path
+from typing import (
+    Optional,
+    Sequence,
+    Dict,
+    Mapping,
+    Set,
+    Union,
+)
+from urllib.parse import quote, urlparse
+
+import datacube.utils.uris as dc_uris
+
+from eodatasets3 import utils
+from eodatasets3.model import DEA_URI_PREFIX
+from eodatasets3.properties import Eo3Interface, Eo3Dict
+
+# Needed when packaging zip or tar files.
+dc_uris.register_scheme("zip", "tar")
+
+
+class LazyProductName:
+    def __init__(self, include_instrument=True, include_collection=False) -> None:
+        super().__init__()
+        self.include_instrument = include_instrument
+        self.include_collection = include_collection
+
+    def __get__(self, c: "NameGenerator", owner) -> str:
+        if c.metadata.product_name:
+            return c.metadata.product_name
+
+        instrument = c.instrument_abbreviated if self.include_instrument else ""
+        return "_".join(
+            p
+            for p in (
+                c.producer_abbreviated,
+                f"{c.platform_abbreviated or ''}{instrument or ''}",
+                c.metadata.product_family,
+                (
+                    f"{c.displayed_collection_number}"
+                    if (self.include_collection and c.displayed_collection_number)
+                    else None
+                ),
+            )
+            if p
+        )
+
+
+def _strip_major_version(version: str) -> str:
+    """
+    >>> _strip_major_version('1.2.3')
+    '2.3'
+    >>> _strip_major_version('01.02.03')
+    '02.03'
+    >>> _strip_major_version('30.40')
+    '40'
+    >>> _strip_major_version('40')
+    ''
+    """
+    return ".".join(version.split(".")[1:])
+
+
+class LazyLabel:
+    def __init__(self, include_version=True, strip_major_version=False) -> None:
+        super().__init__()
+        self.strip_major_version = strip_major_version
+        self.include_version = include_version
+
+    def __get__(self, c: "NameGenerator", owner) -> str:
+        d = c.metadata
+
+        product_prefix = c.product_name
+
+        version = d.dataset_version
+        if version and self.include_version:
+            if self.strip_major_version:
+                version = _strip_major_version(version)
+            version = version.replace(".", "-")
+            product_prefix = f"{c.product_name}-{version}"
+
+        maturity: str = d.properties.get("dea:dataset_maturity")
+        return "_".join(
+            [
+                p
+                for p in (
+                    product_prefix,
+                    d.region_code,
+                    f"{d.datetime:%Y-%m-%d}",
+                    maturity,
+                )
+                if p
+            ]
+        )
+
+
+class LazyPlatformAbbreviation:
+    # The abbreviations mentioned in DEA naming conventions doc.
+    KNOWN_PLATFORM_ABBREVIATIONS = {
+        "landsat-5": "ls5",
+        "landsat-7": "ls7",
+        "landsat-8": "ls8",
+        "landsat-9": "ls9",
+        "sentinel-1a": "s1a",
+        "sentinel-1b": "s1b",
+        "sentinel-2a": "s2a",
+        "sentinel-2b": "s2b",
+        "aqua": "aqu",
+        "terra": "ter",
+    }
+
+    # If all platform (abbreviations) match a pattern, return this group name instead.
+    KNOWN_PLATFORM_GROUPINGS = {
+        "ls": re.compile(r"ls\d+"),
+        "s1": re.compile(r"s1[a-z]+"),
+        "s2": re.compile(r"s2[a-z]+"),
+    }
+
+    def __init__(
+        self,
+        *,
+        known_abbreviations: Dict = None,
+        grouped_abbreviations: Dict = None,
+        show_specific_platform=True,
+        allow_unknown_abbreviations=True,
+    ) -> None:
+        self.known_abbreviations = (
+            known_abbreviations or self.KNOWN_PLATFORM_ABBREVIATIONS
+        )
+        self.grouped_abbreviations = (
+            grouped_abbreviations or self.KNOWN_PLATFORM_GROUPINGS
+        )
+        self.show_specific_platform = show_specific_platform
+
+        self.allow_unknown_abbreviations = allow_unknown_abbreviations
+
+    def __get__(self, c: "NameGenerator", owner) -> Optional[str]:
+        """Abbreviated form of a satellite, as used in dea product names. eg. 'ls7'."""
+
+        p = c.metadata.platforms
+        if not p:
+            return None
+
+        if not self.allow_unknown_abbreviations:
+            unknowns = p.difference(self.known_abbreviations)
+            if unknowns:
+                raise ValueError(
+                    f"We don't know the DEA abbreviation for platforms {unknowns!r}. "
+                    f"We'd love to add more! Raise an issue on Github: "
+                    f"https://github.com/GeoscienceAustralia/eo-datasets/issues/new' "
+                )
+
+        abbreviations = sorted(
+            self.known_abbreviations.get(s, s.replace("-", "")) for s in p
+        )
+
+        if self.show_specific_platform and len(abbreviations) == 1:
+            return abbreviations[0]
+
+        # If all abbreviations are in a group, name it using that group.
+        # (eg. "ls" instead of "ls5-ls7-ls8")
+        for group_name, pattern in self.grouped_abbreviations.items():
+            if all(pattern.match(a) for a in abbreviations):
+                return group_name
+
+        # Otherwise, there's a mix of platforms.
+
+        # Is there a common constellation?
+        constellation = c.metadata.properties.get("constellation")
+        if constellation:
+            return constellation
+
+        # Don't bother to include platform in name for un-groupable mixes of them.
+        if not self.allow_unknown_abbreviations:
+            raise NotImplementedError(
+                f"Satellite constellation abbreviation is not known for platforms {p}. "
+                f"(for DEA derivative naming conventions.)"
+                f"    Is this a mistake? We'd love to add more! Raise an issue on Github: "
+                f"https://github.com/GeoscienceAustralia/eo-datasets/issues/new' "
+            )
+        return None
+
+
+class LazyInstrumentAbbreviation:
+    def __get__(self, c: "NameGenerator", owner) -> Optional[str]:
+        """Abbreviated form of an instrument name, as used in dea product names. eg. 'c'."""
+        if not c.metadata.instrument:
+            return None
+
+        platforms = c.metadata.platforms
+        if not platforms or len(platforms) > 1:
+            return None
+
+        [p] = platforms
+
+        if p.startswith("sentinel-1") or p.startswith("sentinel-2"):
+            return c.metadata.instrument[0].lower()
+
+        if p.startswith("landsat"):
+            # Extract from usgs standard:
+            # landsat:landsat_product_id: LC08_L1TP_091075_20161213_20170316_01_T2
+            # landsat:landsat_scene_id: LC80910752016348LGN01
+            landsat_id = c.metadata.properties.get("landsat:landsat_product_id")
+            if landsat_id is None:
+                landsat_id = c.metadata.properties.get("landsat:landsat_scene_id")
+
+            # from USGS STAC, label is LC08_L2SP_178079_20210417_20210424_02_T1_SR and
+            # landsat:scene_id: LC81780792021107LGN00
+            if landsat_id is None:
+                landsat_id = c.metadata.properties.get("landsat:scene_id")
+
+            if not landsat_id:
+                raise NotImplementedError(
+                    "No landsat scene or product id: cannot abbreviate Landsat instrument."
+                )
+
+            return landsat_id[1].lower()
+
+        # Otherwise, it's unknown.
+        raise NotImplementedError(
+            f"Instrument abbreviations aren't supported for platform {p!r}. "
+            f"We'd love to add more support! Raise an issue on Github: "
+            f"https://github.com/GeoscienceAustralia/eo-datasets/issues/new' "
+        )
+
+
+class LazyProducerAbbreviation:
+    KNOWN_PRODUCER_ABBREVIATIONS = {
+        "ga.gov.au": "ga",
+        "usgs.gov": "usgs",
+        "sinergise.com": "sinergise",
+        "digitalearthafrica.org": "deafrica",
+        "esa.int": "esa",
+        # Is there another organisation you want to use? Pull requests very welcome!
+    }
+
+    def __init__(self, *, known_abbreviations: Dict = None) -> None:
+        self.known_abbreviations = (
+            known_abbreviations or self.KNOWN_PRODUCER_ABBREVIATIONS
+        )
+
+    def __get__(self, c: "NameGenerator", owner) -> Optional[str]:
+        """Abbreviated form of a producer, as used in dea product names. eg. 'ga', 'usgs'."""
+        if not c.metadata.producer:
+            return None
+
+        try:
+            return self.known_abbreviations[c.metadata.producer]
+        except KeyError:
+            raise NotImplementedError(
+                f"We don't know how to abbreviate organisation domain name {c.metadata.producer!r}. "
+                f"We'd love to add more orgs! Raise an issue on Github: "
+                f"https://github.com/GeoscienceAustralia/eo-datasets/issues/new' "
+            )
+
+
+class LazyDestinationFolder:
+    def __init__(
+        self,
+        date_folders_format="%Y/%m/%d",
+        include_version=False,
+        include_non_final_maturity=True,
+    ) -> None:
+        super().__init__()
+        self.include_version = include_version
+        self.include_non_final_maturity = include_non_final_maturity
+        self.date_folders_format = date_folders_format
+
+    def __get__(self, c: "NameGenerator", owner) -> Path:
+        """The folder hierarchy the datasets files go into.
+
+        This is returned as a relative path.
+
+        Example: Path("ga_ls8c_ard_3/092/084/2016/06/28")
+        """
+        d = c.metadata
+        parts = [c.product_name]
+
+        if self.include_version:
+            parts.append(d.dataset_version.replace(".", "-"))
+
+        # Cut the region code in subfolders
+        region_code = d.region_code
+        if region_code:
+            parts.extend(utils.subfolderise(region_code))
+
+        parts.extend(d.datetime.strftime(self.date_folders_format).split("/"))
+
+        if self.include_non_final_maturity:
+            # If it's not a final product, append the maturity to the folder.
+            maturity: str = d.properties.get("dea:dataset_maturity")
+            if maturity and maturity != "final":
+                parts[-1] = f"{parts[-1]}_{maturity}"
+
+        if c.dataset_separator_field is not None:
+            val = d.properties[c.dataset_separator_field]
+            # TODO: choosable formatter?
+            if isinstance(val, datetime):
+                val = f"{val:%Y%m%dT%H%M%S}"
+            parts.append(val)
+        return Path(*parts)
+
+
+class LazyDatasetLocation:
+    """The location of the dataset as indexed into ODC. Defaults to the metadata path."""
+
+    def __get__(self, c: "NameGenerator", owner) -> str:
+        if not c.collection_prefix:
+            raise ValueError(
+                "collection_prefix is required if you're not setting a "
+                "dataset_location or metadata_path!"
+            )
+
+        offset = c.dataset_folder / c.metadata_file
+        if offset.is_absolute():
+            raise ValueError("Dataset offset is expected to be relative to collection")
+        return f"{c.collection_prefix}/{offset.as_posix()}"
+
+
+class MissingRequiredFields(ValueError):
+    ...
+
+
+class RequiredPropertyDict(Eo3Dict):
+    """A wrapper for Eo3 Dict that throws a loud error if a required field is accessed
+     by not yet set.
+
+    - It gives a friendly error of what fields are required, rather than the user
+    seeing obtuse "None" errors sprinkled throughout their code.
+    - It will _only_ complain about the given required fields. Other fields behave
+      like a normal dictionary.
+
+    """
+
+    # Displayed to user for friendlier errors.
+    _REQUIRED_PROPERTY_HINTS = {
+        "odc:product_family": 'eg. "wofs" or "level1"',
+        "odc:processing_datetime": "Time of processing, perhaps datetime.utcnow()?",
+        "odc:producer": "Creator of data, eg 'usgs.gov' or 'ga.gov.au'",
+        "odc:dataset_version": "eg. 1.0.0",
+    }
+
+    def __init__(
+        self,
+        required_fields: Set[str],
+        properties=None,
+    ) -> None:
+        self.required_fields = required_fields
+        super().__init__(properties)
+
+    def __getitem__(self, item):
+        try:
+            val = super().__getitem__(item)
+            if (not val) and (item in self.required_fields):
+                self._raise_all_missing_requirements()
+            return val
+        except KeyError:
+            if item in self.required_fields:
+                self._raise_all_missing_requirements()
+            raise
+
+    def _raise_all_missing_requirements(self):
+        """
+        Do we have enough properties to generate file or product names?
+        """
+        missing_props = []
+        for f in self.required_fields:
+            if f not in self._props:
+                missing_props.append(f)
+        if missing_props:
+            examples = []
+            for p in sorted(missing_props):
+                hint = self._REQUIRED_PROPERTY_HINTS.get(p, "")
+                if hint:
+                    hint = f" ({hint})"
+                examples.append(f"\n- {p!r}{hint}")
+
+            raise MissingRequiredFields(
+                f"Need more properties to fulfill naming conventions."
+                f"{''.join(examples)}"
+            )
+
+
+class EnforceRequirementProperties(Eo3Interface):
+    """
+    A wrapper for EO3 fields that throws a loud error if a field in required_properties isn't set.
+
+    Throws MissingRequiredFields error.
+    """
+
+    @property
+    def properties(self) -> Eo3Dict:
+        return self._props
+
+    def __init__(self, properties: Mapping, required_fields: Set[str]) -> None:
+        self._props = RequiredPropertyDict(required_fields, properties)
+
+
+class LazyFileName:
+    def __init__(self, file_id: str, suffix: str) -> None:
+        self.file_id = file_id
+        self.suffix = suffix
+
+    def __get__(self, c: "NameGenerator", owner) -> Path:
+        return c.filename(file_id=self.file_id, suffix=self.suffix)
+
+
+class LazyProductURI:
+    def __get__(self, n: "NameGenerator", owner) -> Optional[str]:
+        if not n.base_product_uri:
+            return None
+
+        return f"{n.base_product_uri}/product/{quote(n.product_name)}"
+
+
+def resolve_location(path: Union[str, Path]) -> str:
+    """
+    Make sure a dataset location is a URL, suitable to be
+    the dataset_location in datacube indexing.
+
+    Users may specify a pathlib.Path(), and we'll convert it as needed.
+    """
+    if isinstance(path, str):
+        if not dc_uris.is_url(path) and not dc_uris.is_vsipath(path):
+            raise ValueError(
+                "A string location is expected to be a URL or VSI path. "
+                "Perhaps you want to give it as a local pathlib.Path()?"
+            )
+        return path
+
+    path = dc_uris.normalise_path(path)
+    if ".tar" in path.suffixes:
+        return f"tar:{path}!/"
+    elif ".zip" in path.suffixes:
+        return f"zip:{path}!/"
+    else:
+        uri = path.as_uri()
+        # Base paths specified as directories must end in a slash,
+        # so they will be url joined as subfolders. (pathlib strips them)
+        if path.is_dir():
+            return f"{uri}/"
+        return uri
+
+
+def _as_path(url: str) -> Path:
+    """Try to convert the given URL to a local Path"""
+    parts = urlparse(url)
+    if not parts.scheme == "file":
+        raise ValueError(f"Expected a filesystem path, got a URL! {url!r}")
+
+    return Path(parts.path)
+
+
+class NameGenerator:
+    """
+    A generator of names for products, data labels, file paths, urls, etc.
+
+    These are generated based on a given set of naming conventions, but a user
+    can manually override any properties to avoid generation.
+
+    Create an instance by calling :meth:`eodatasets3.namer`:
+
+    .. testcode ::
+
+        from eodatasets3 import namer
+
+        properties = {
+            'eo:platform': 'sentinel-2a',
+            'eo:instrument': 'MSI',
+            'odc:product_family': 'level1',
+        }
+        n = namer(properties, conventions='default')
+        print(n.product_name)
+
+    .. testoutput ::
+
+       s2am_level1
+
+    .. note ::
+
+       You may want to use an :class:`eodatasets3.DatasetDoc` instance rather than a dict for
+       properties, to get convenience methods such as ``.platform = 'sentinel-2a'``, `.properties`, automatic
+       property normalisation etc.
+
+       See :ref:`the naming section<names_n_paths>` for an example.
+
+    Fields are lazily generated when accessed using the underlying metadata properties,
+    but you can manually set any field to avoid generation:
+
+    .. doctest ::
+
+        >>> from eodatasets3 import DatasetDoc
+        >>> p = DatasetDoc()
+        >>> p.platform = 'landsat-7'
+        >>> p.product_family = 'nbar'
+        >>>
+        >>> n = namer(conventions='default', properties=p)
+        >>> n.product_name
+        'ls7_nbar'
+        >>> # Manually override the abbreviation:
+        >>> n.platform_abbreviated = 'ls'
+        >>> n.product_name
+        'ls_nbar'
+        >>> # Or manually set the entire product name to avoid generation:
+        >>> n.product_name = 'custom_nbar_albers'
+
+    In order to calculate paths, give it a collection prefix. This can be a :class:`Path object <pathlib.Path>`
+    for local files, or a URL str for remote.
+
+    .. doctest ::
+
+       >>> p.datetime = datetime(2014, 4, 5)
+       >>> collection = "s3://dea-public-data-dev/collections"
+       >>> n = namer(conventions='default', properties=p, collection_prefix=collection)
+       >>> n.dataset_location
+       's3://dea-public-data-dev/collections/ls7_nbar/2014/04/05/ls7_nbar_2014-04-05.odc-metadata.yaml'
+
+    All fields named ``*_file`` are filenames inside (relative to) the
+    ``self.dataset_location``.
+
+        >>> n.resolve_file('thumbnail.jpg')
+        's3://dea-public-data-dev/collections/ls7_nbar/2014/04/05/thumbnail.jpg'
+
+    """
+
+    _ABSOLUTE_MINIMAL_PROPERTIES = {
+        "odc:product_family",
+        # Required by Stac regardless.
+        "datetime",
+    }
+
+    # Placed here for public usage, as people can extend the defaults.
+    KNOWN_PRODUCER_ABBREVIATIONS = LazyProducerAbbreviation.KNOWN_PRODUCER_ABBREVIATIONS
+    KNOWN_PLATFORM_ABBREVIATIONS = LazyPlatformAbbreviation.KNOWN_PLATFORM_ABBREVIATIONS
+    KNOWN_PLATFORM_GROUPINGS = LazyPlatformAbbreviation.KNOWN_PLATFORM_GROUPINGS
+
+    # These are lazily computed on read if not overridden by the user.
+    # ie. User can set th names.product_name = 'blah'
+
+    #: Product name for ODC
+    #:
+    product_name: str = LazyProductName(include_collection=True)
+    #: Identifier URL for the product
+    #: (This is seen as a global id for the product, unlike the plain product
+    #: name. It doesn't have to resolve to a real path)
+    #:
+    #: Eg. ``https://collections.dea.ga.gov.au/product/ga_ls8c_ard_3``
+    #:
+    product_uri: str = LazyProductURI()
+
+    #: Abbreviated form of the platform, used in most other
+    #: paths and names here.
+    #:
+    #: For example, ``landsat-7`` is usually abbreviated to ``ls7``
+    platform_abbreviated: str = LazyPlatformAbbreviation()
+
+    #: Abbreviated form of the instrument, used in most other
+    #: paths and names here.
+    #:
+    #: For example, ``ETM+`` is usually abbreviated to ``e``
+    instrument_abbreviated: str = LazyInstrumentAbbreviation()
+    #: Abbreviated form of the producer of the dataset
+    #: (the producing organisation)
+    #:
+    #: For example, "ga.gov.au" is abbreviated to "ga"
+    producer_abbreviated: str = LazyProducerAbbreviation()
+
+    #: The Label for the dataset. This is a human-readable alternative
+    #: to showing the UUID in most parts of ODC. It's used by default in filenames
+    #:
+    # No major version by default, as the product name contains it (the collection version).
+    dataset_label: str = LazyLabel(strip_major_version=True)
+
+    #: The pattern for generating file names.
+    #:
+    #: The pattern is in python's ``str.format()`` syntax,
+    #: with fields ``{file_id}`` and ``{suffix}``
+    #:
+    #: The namer instance is readable from ``{n}``.
+    filename_pattern: str = "{n.dataset_label}{file_id}.{suffix}"
+
+    #: The prefix where all files are stored, as a URI.
+    #:
+    #: Eg. ``file:///my/dataset/collections``
+    #:
+    #: (used if dataset_location is generated)
+    collection_prefix: Optional[str] = None
+
+    #: The folder offset from the collection_prefix.
+    #:
+    #: Example: ``Path('ga_ls8c_ones_3/090/084/2016/01/21')``
+    #:
+    #: (used if dataset_location is generated)
+    dataset_folder: Path = LazyDestinationFolder()
+
+    #: The full uri of the dataset as indexed into ODC.
+    #:
+    #: **All inner document paths are relative to this.**
+    #:
+    #: Eg. ``s3://dea-public-data/ga_ls_fc_3/2-5-0/091/086/2020/04/04/ga_ls_fc_091086_2020-04-04.odc-metadata.yaml``
+    #:
+    #: (Defaults to the metadata path inside the dataset_folder)
+    dataset_location: str = LazyDatasetLocation()
+
+    #: The path to the ODC metadata file.
+    #:
+    #: (it's relative to self.dataset_location ... but could be absolute too)
+    #:
+    #: Example: ``Path('ga_ls8c_ones_3-0-0_090084_2016-01-21_final.odc-metadata.yaml')``
+    metadata_file: Path = LazyFileName("", "odc-metadata.yaml")
+
+    #: The name of a checksum file
+    checksum_file: Path = LazyFileName("", "sha1")
+
+    def __init__(
+        self,
+        properties: Mapping,
+        base_product_uri: str = None,
+        required_fields: Sequence[str] = (),
+        dataset_separator_field: Optional[str] = None,
+        allow_unknown_abbreviations: bool = True,
+    ) -> None:
+
+        #: The default base URI used in product URI generation
+        #:
+        #: Example: ``https://collections.dea.ga.gov.au/``
+        self.base_product_uri = base_product_uri
+
+        self.required_fields = self._ABSOLUTE_MINIMAL_PROPERTIES.union(required_fields)
+
+        # An extra folder to put each dataset inside, using the value of the given property name.
+        self.dataset_separator_field = dataset_separator_field
+
+        if self.dataset_separator_field is not None:
+            self.required_fields.add(dataset_separator_field)
+
+        self.allow_unknown_abbreviations = allow_unknown_abbreviations
+        #: The underlying dataset properties used for generation.
+        self.metadata: Eo3Interface = EnforceRequirementProperties(
+            properties, self.required_fields
+        )
+
+    @property
+    def displayed_collection_number(self) -> Optional[int]:
+        # An explicit collection number trumps all.
+        if self.metadata.collection_number:
+            return int(self.metadata.collection_number)
+
+        # Otherwise it's the first digit of the dataset version.
+        if not self.metadata.dataset_version:
+            return None
+        return int(self.metadata.dataset_version.split(".")[0])
+
+    def metadata_filename(self, kind: str = "", suffix: str = "yaml") -> Path:
+        return self.filename(kind, suffix)
+
+    def measurement_filename(
+        self, measurement_name: str, suffix: str = "tif", file_id: str = None
+    ) -> Path:
+        """
+        Generate the path to a measurement for the current naming conventions.:::
+
+            >> p.names.measurement_file('blue', 'tif')
+            Path('ga_ls8c_ones_3-0-0_090084_2016-01-21_final_blue.tif')
+
+        This is the filename inside the self.dataset_folder
+        """
+        name = measurement_name.replace(":", "_")
+
+        return self.filename(
+            # We use 'band01'/etc in the filename if provided, rather than 'red'
+            file_id or name,
+            suffix,
+        )
+
+    def filename(self, file_id: str, suffix: str) -> Path:
+        """
+        Make a file name according to the current naming conventions' file pattern.
+
+        All filenames have a file_id (eg. "odc-metadata" or "") and a suffix (eg. "yaml")
+
+        Returned file paths are expected to be relative to the ``self.dataset_location``
+        """
+        file_id = "_" + file_id.replace("_", "-") if file_id else ""
+        return Path(
+            self.filename_pattern.format(file_id=file_id, suffix=suffix, n=self)
+        )
+
+    def thumbnail_filename(self, kind: str = None, suffix: str = "jpg") -> Path:
+        """
+        Get a thumbnail file path (optionally with the given kind and/or suffix.)
+        """
+        if kind:
+            name = f"{kind}_thumbnail"
+        else:
+            name = "thumbnail"
+        return self.filename(name, suffix)
+
+    def resolve_file(self, path: Union[str, Path]) -> str:
+        """
+        Convert the given file offset to a fully qualified URL within the dataset location.
+        """
+        if isinstance(path, Path):
+            if path.is_absolute():
+                return path.as_uri()
+            path = path.as_posix()
+
+        location = self.dataset_location
+        resolved = dc_uris.uri_resolve(location, path)
+        return resolved
+
+    def resolve_path(self, path: Union[str, Path]) -> Path:
+        """
+        Convert the given file offset to a local Path inside the dataset location (if possible).
+
+        :raises ValueError if the current dataset is not in a file:// location.
+        """
+        return _as_path(self.resolve_file(path))
+
+    @property
+    def dataset_path(self) -> Optional[Path]:
+        """
+        Get the dataset location as a local Path, if possible.
+        """
+        return _as_path(self.dataset_location)
+
+    @property
+    def collection_path(self) -> Optional[Path]:
+        """
+        Get the collection prefix as a local path, if possible.
+        """
+        if not self.collection_prefix:
+            return None
+        try:
+            return _as_path(self.collection_prefix)
+        except ValueError:
+            return None
+
+
+class DEANamingConventions(NameGenerator):
+    """
+    Example file structure (note version number in file):
+
+            ga_ls8c_ones_3/090/084/2016/01/21/ga_ls8c_ones_3-0-0_090084_2016-01-21_final.odc-metadata.yaml
+    """
+
+    def __init__(
+        self,
+        properties: Mapping,
+        required_fields=(
+            "eo:platform",
+            "eo:instrument",
+            "odc:processing_datetime",
+            "odc:producer",
+            "odc:product_family",
+            "odc:region_code",
+            "odc:dataset_version",
+        ),
+        dataset_separator_field: Optional[str] = None,
+    ) -> None:
+        # DEA wants consistency via the naming-conventions doc.
+        allow_unknown_abbreviations = False
+        super().__init__(
+            properties,
+            DEA_URI_PREFIX,
+            required_fields,
+            dataset_separator_field,
+            allow_unknown_abbreviations,
+        )
+
+    product_name: str = LazyProductName(include_collection=True)
+    # Stricter: only allow pre-approved abbreviations.
+    platform_abbreviated: str = LazyPlatformAbbreviation(
+        allow_unknown_abbreviations=False
+    )
+
+
+class DEAS2NamingConventions(DEANamingConventions):
+    """
+    DEA naming conventions, but with an extra subfolder for each unique datatake.
+
+    It will figure out the datatake if you set a sentinel_tile_id or datastrip_id.
+    """
+
+    def __init__(
+        self,
+        properties: Mapping,
+        required_fields=(
+            "eo:instrument",
+            "eo:platform",
+            "odc:dataset_version",
+            "odc:processing_datetime",
+            "odc:producer",
+            "odc:product_family",
+            "odc:region_code",
+        ),
+        dataset_separator_field="sentinel:datatake_start_datetime",
+    ) -> None:
+        super().__init__(
+            properties, required_fields, dataset_separator_field=dataset_separator_field
+        )
+
+
+class DEADerivativesNamingConventions(DEANamingConventions):
+    """
+    Common derived products.
+
+    Unlike plain 'DEA', they use an explicit collection number (odc:collection_number)
+    in the product name which may differ from the software's dataset version
+    (odc:dataset_version)
+
+    Example file structure (note version number in folder):
+
+        ga_ls_wo_3/1-6-0/090/081/1998/07/30/ga_ls_wo_3_090081_1998-07-30_interim.odc-metadata.yaml
+
+    Derivatives have a slightly different folder structure.
+
+    And they only show constellations (eg. "ls_" or "s2_") rather than the specific
+    satellites in their names (eg. "ls8_").
+
+    They have a version-number folder instead of putting it in each filename.
+
+    And version numbers may not match the collection number (`odc:collection_number` is
+    mandatory).
+    """
+
+    def __init__(
+        self,
+        properties: Mapping,
+        required_fields: Sequence[str] = (
+            "eo:platform",
+            "odc:dataset_version",
+            "odc:collection_number",
+            "odc:processing_datetime",
+            "odc:producer",
+            "odc:product_family",
+            "odc:region_code",
+            "dea:dataset_maturity",
+        ),
+        dataset_separator_field: Optional[str] = None,
+    ) -> None:
+        super().__init__(
+            properties,
+            required_fields=required_fields,
+            dataset_separator_field=dataset_separator_field,
+        )
+
+    # Derivates put the version in the folder instead.
+    dataset_label = LazyLabel(include_version=False)
+    product_name = LazyProductName(include_instrument=False, include_collection=True)
+
+    dataset_folder = LazyDestinationFolder(
+        include_version=True,
+        include_non_final_maturity=False,
+    )
+
+    platform_abbreviated = LazyPlatformAbbreviation(
+        show_specific_platform=False,
+        allow_unknown_abbreviations=False,
+    )
+
+
+class DEAS2DerivativesNamingConventions(DEADerivativesNamingConventions):
+    """
+    Sentinel-2-based DEA derivative naming conventions. Unlike regular
+    derivatives, there's an extra subfolder for each unique datatake.
+
+    It will figure out the datatake if you set a sentinel_tile_id
+    or datastrip_id.
+    """
+
+    def __init__(self, properties: Mapping) -> None:
+        super().__init__(
+            properties,
+            dataset_separator_field="sentinel:datatake_start_datetime",
+        )
+
+
+class AfricaProductName:
+    def __get__(self, c: "NameGenerator", owner) -> str:
+        if c.metadata.product_name:
+            return c.metadata.product_name
+        return f"{c.metadata.product_family}_{c.platform_abbreviated}"
+
+
+class DEAfricaNamingConventions(NameGenerator):
+    """
+    DEAfrica avoids org names and uses simpler "{family}_{platform}" product names.
+
+    Eg. "wo_ls" (water observations of landsat)
+
+    """
+
+    product_name = AfricaProductName()
+    dataset_label = LazyLabel(include_version=False)
+    dataset_folder = LazyDestinationFolder(
+        include_version=True,
+        include_non_final_maturity=False,
+    )
+    platform_abbreviated = LazyPlatformAbbreviation(
+        show_specific_platform=False,
+        allow_unknown_abbreviations=False,
+    )
+
+    def __init__(
+        self,
+        properties: Mapping,
+    ) -> None:
+        super().__init__(
+            properties,
+            base_product_uri="https://digitalearthafrica.org",
+            required_fields=(
+                "eo:platform",
+                "odc:producer",
+                "odc:region_code",
+                "odc:product_family",
+                "odc:dataset_version",
+            ),
+        )
+
+
+KNOWN_CONVENTIONS = dict(
+    default=NameGenerator,
+    dea=DEANamingConventions,
+    dea_s2=DEAS2NamingConventions,
+    dea_s2_derivative=DEAS2DerivativesNamingConventions,
+    dea_c3=DEADerivativesNamingConventions,
+    deafrica=DEAfricaNamingConventions,
+)
+
+
+def namer(
+    properties: Union[Eo3Dict, Eo3Interface, dict] = None,
+    *,
+    collection_prefix: Union[str, Path] = None,
+    conventions: str = "default",
+) -> "NameGenerator":
+    """
+    Create a naming instance of the given conventions.
+
+    Conventions: 'default', 'dea', 'deafrica', ...
+
+    You usually give it existing properties, but you can use the return value's
+    :attr:`.metadata <eodatasets3.NameGenerator.metadata>` field to set properties afterwards.
+
+    """
+    if conventions not in KNOWN_CONVENTIONS:
+        available = ", ".join(KNOWN_CONVENTIONS.keys())
+        raise ValueError(
+            f"Unknown naming conventions: {conventions}. Possibilities: {available}"
+        )
+    if isinstance(properties, Eo3Interface):
+        properties = properties.properties
+    if properties is None:
+        properties = Eo3Dict()
+    conventions = KNOWN_CONVENTIONS[conventions](properties)
+    if collection_prefix:
+        conventions.collection_prefix = resolve_location(collection_prefix)
+
+    return conventions
diff --git a/eodatasets3/prepare/landsat_l1_prepare.py b/eodatasets3/prepare/landsat_l1_prepare.py
index 18db3468..c1b9cc9e 100644
--- a/eodatasets3/prepare/landsat_l1_prepare.py
+++ b/eodatasets3/prepare/landsat_l1_prepare.py
@@ -16,7 +16,8 @@
 
 import click
 import rasterio
-from eodatasets3 import serialise, utils, DatasetAssembler, IfExists
+
+from eodatasets3 import serialise, utils, DatasetPrepare
 from eodatasets3.properties import FileFormat
 from eodatasets3.ui import PathPath
 
@@ -383,13 +384,12 @@ def prepare_and_write(
         ]  # for C1 only
         leveln_landsat_data_type = mtl_doc[coll_map["product_contents_of"]]["data_type"]
 
-    with DatasetAssembler(
+    with DatasetPrepare(
         metadata_path=output_yaml_path,
         dataset_location=ds_path,
         # Detministic ID based on USGS's product id (which changes when the scene is reprocessed by them)
         dataset_id=uuid.uuid5(USGS_UUID_NAMESPACE, leveln_product_id),
         naming_conventions="dea",
-        if_exists=IfExists.Overwrite,
     ) as p:
         if source_telemetry:
             # Only GA's data has source telemetry...
@@ -490,8 +490,7 @@ def prepare_and_write(
                     relative_to_dataset_location=True,
                     expand_valid_data=False,
                 )
-        p.add_accessory_file("metadata:landsat_mtl", Path(mtl_filename))
-
+        p.note_accessory_file("metadata:landsat_mtl", Path(mtl_filename))
         return p.done()
 
 
@@ -545,7 +544,7 @@ def main(
     with rasterio.Env():
         for ds in datasets:
             if output_base:
-                output = output_base.joinpath(
+                output = output_base.absolute().joinpath(
                     *utils.subfolderise(_dataset_region_code(ds))
                 )
                 output.mkdir(parents=True, exist_ok=True)
diff --git a/eodatasets3/prepare/sentinel_l1c_prepare.py b/eodatasets3/prepare/sentinel_l1c_prepare.py
index 0032e62d..7c826ffa 100644
--- a/eodatasets3/prepare/sentinel_l1c_prepare.py
+++ b/eodatasets3/prepare/sentinel_l1c_prepare.py
@@ -15,7 +15,8 @@
 import click
 from defusedxml import minidom
 
-from eodatasets3 import DatasetAssembler
+from eodatasets3 import DatasetPrepare
+from eodatasets3.properties import Eo3Interface
 from eodatasets3.ui import PathPath
 
 # Static namespace to generate uuids for datacube indexing
@@ -159,7 +160,7 @@ def process_user_product_metadata(contents: str) -> Dict:
     }
 
 
-def _get_stable_id(p: DatasetAssembler) -> uuid.UUID:
+def _get_stable_id(p: Eo3Interface) -> uuid.UUID:
     # These should have been extracted from our metadata files!
     producer = p.producer
     product_name = p.properties["sentinel:product_name"]
@@ -171,20 +172,20 @@ def _get_stable_id(p: DatasetAssembler) -> uuid.UUID:
 
 
 def prepare_and_write(
-    ds_path: Path,
+    dataset_location: Path,
     output_yaml: Path,
     producer: str,
 ) -> Tuple[uuid.UUID, Path]:
-    with DatasetAssembler(
+    with DatasetPrepare(
         metadata_path=output_yaml,
-        dataset_location=ds_path,
+        dataset_location=dataset_location,
     ) as p:
         p.properties["odc:producer"] = producer
 
         if producer == "esa.int":
-            jp2_offsets = _extract_esa_fields(ds_path, p)
+            jp2_offsets = _extract_esa_fields(dataset_location, p)
         elif producer == "sinergise.com":
-            jp2_offsets = _extract_sinergise_fields(ds_path, p)
+            jp2_offsets = _extract_sinergise_fields(dataset_location.parent, p)
         else:
             raise NotImplementedError(
                 f"Unknown s2 producer {producer}. Expected 'sinergise.com' or 'esa.int'"
@@ -235,7 +236,7 @@ def _get_platform_name(p: Mapping) -> str:
     return platform_name
 
 
-def _extract_sinergise_fields(path: Path, p: DatasetAssembler) -> Iterable[Path]:
+def _extract_sinergise_fields(path: Path, p: DatasetPrepare) -> Iterable[Path]:
     """Extract Sinergise metadata and return list of image offsets"""
     product_info_path = path / "productInfo.json"
     metadata_xml_path = path / "metadata.xml"
@@ -247,14 +248,14 @@ def _extract_sinergise_fields(path: Path, p: DatasetAssembler) -> Iterable[Path]
         )
 
     p.properties.update(process_sinergise_product_info(product_info_path))
-    p.add_accessory_file("metadata:sinergise_product_info", product_info_path)
+    p.note_accessory_file("metadata:sinergise_product_info", product_info_path)
 
     p.properties.update(process_tile_metadata(metadata_xml_path.read_text()))
-    p.add_accessory_file("metadata:s2_tile", metadata_xml_path)
+    p.note_accessory_file("metadata:s2_tile", metadata_xml_path)
 
     # TODO: sinergise folders could `process_datastrip_metadata()` in an outer directory?
 
-    return path.glob("*.jp2")
+    return list(path.glob("*.jp2"))
 
 
 def _extract_esa_fields(dataset, p) -> Iterable[Path]:
@@ -274,17 +275,17 @@ def one(suffix: str) -> str:
         p.properties.update(
             process_datastrip_metadata(z.read(datastrip_md).decode("utf-8"))
         )
-        p.add_accessory_file("metadata:s2_datastrip", datastrip_md)
+        p.note_accessory_file("metadata:s2_datastrip", datastrip_md)
 
         tile_md = one("MTD_TL.xml")
         p.properties.update(process_tile_metadata(z.read(tile_md).decode("utf-8")))
-        p.add_accessory_file("metadata:s2_tile", tile_md)
+        p.note_accessory_file("metadata:s2_tile", tile_md)
 
         user_product_md = one("MTD_MSIL1C.xml")
         p.properties.update(
             process_user_product_metadata(z.read(user_product_md).decode("utf-8"))
         )
-        p.add_accessory_file("metadata:s2_user_product", user_product_md)
+        p.note_accessory_file("metadata:s2_user_product", user_product_md)
 
         return [Path(p) for p in z.namelist() if "IMG_DATA" in p and p.endswith(".jp2")]
 
@@ -340,8 +341,8 @@ def main(
             *(
                 (
                     "sinergise.com",
-                    # Input is the folder
-                    p.parent,
+                    # Input folder: our indexed dataset location is the metadata file.
+                    (p.parent / f"{p.parent.stem}.odc-metadata.yaml"),
                     # Output is an inner metadata file, with the same name as the folder (usually S2A....).
                     (p.parent / f"{p.parent.stem}.odc-metadata.yaml"),
                 )
@@ -350,7 +351,7 @@ def main(
             *(
                 (
                     "esa.int",
-                    # Input is zip file
+                    # Input is zip file, so the zip should be our location.
                     p,
                     # Output is a sibling file with metadata suffix.
                     p.with_suffix(".odc-metadata.yaml"),
@@ -366,7 +367,7 @@ def main(
 
         for producer, ds_path, output_yaml in in_out_paths:
             if output_base:
-                output_yaml = output_base / output_yaml.name
+                output_yaml = output_base.absolute() / output_yaml.name
 
             if output_yaml.exists():
                 if not overwrite_existing:
@@ -375,8 +376,8 @@ def main(
 
                 logging.info("Output exists: overwriting %s", output_yaml)
 
-            uuid, path = prepare_and_write(ds_path, output_yaml, producer)
-            logging.info("Wrote dataset %s to %s", uuid, path)
+            uuid_, path = prepare_and_write(ds_path, output_yaml, producer)
+            logging.info("Wrote dataset %s to %s", uuid_, path)
 
         sys.exit(0)
 
diff --git a/eodatasets3/properties.py b/eodatasets3/properties.py
index 8ab021eb..d3982d20 100644
--- a/eodatasets3/properties.py
+++ b/eodatasets3/properties.py
@@ -261,11 +261,19 @@ def parsed_sentinel_datastrip_id(tile_id) -> Tuple[str, Dict]:
 }
 
 
-class StacPropertyView(collections.abc.MutableMapping):
+class Eo3Dict(collections.abc.MutableMapping):
+    """
+    This acts like a dictionary, but will normalise known properties (consistent
+    case, types etc) and warn about common mistakes.
+
+    It wraps an inner dictionary. By default it will normalise the fields in
+    the input dictionary on creation, but you can disable this with `normalise_input=False`.
+    """
+
     # Every property we've seen or dealt with so far. Feel free to expand with abandon...
     # This is to minimise minor typos, case differences, etc, which plagued previous systems.
     # Keep sorted.
-    KNOWN_STAC_PROPERTIES: Mapping[str, Optional[NormaliseValueFn]] = {
+    KNOWN_PROPERTIES: Mapping[str, Optional[NormaliseValueFn]] = {
         "datetime": datetime_type,
         "dea:dataset_maturity": of_enum_type(("final", "interim", "nrt"), lower=True),
         "dtr:end_datetime": datetime_type,
@@ -303,9 +311,17 @@ class StacPropertyView(collections.abc.MutableMapping):
         **_SENTINEL_EXTENDED_PROPS,
     }
 
-    def __init__(self, properties=None) -> None:
-        self._props = properties or {}
-
+    # For backwards compatibility, in case users are extending at runtime.
+    KNOWN_STAC_PROPERTIES = KNOWN_PROPERTIES
+
+    def __init__(self, properties: Mapping = None, normalise_input=True) -> None:
+        if properties is None:
+            properties = {}
+        self._props = properties
+        # We normalise the properties they gave us.
+        if normalise_input:
+            for key in self._props:
+                self.normalise_and_set(key, self._props[key], expect_override=True)
         self._finished_init_ = True
 
     def __setattr__(self, name: str, value: Any) -> None:
@@ -331,6 +347,9 @@ def __len__(self):
     def __delitem__(self, name: str) -> None:
         del self._props[name]
 
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({self._props!r})"
+
     def __setitem__(self, key, value):
         self.normalise_and_set(
             key,
@@ -339,12 +358,20 @@ def __setitem__(self, key, value):
             allow_override=True,
         )
 
-    def normalise_and_set(self, key, value, allow_override=True):
+    def normalise_and_set(self, key, value, allow_override=True, expect_override=False):
         """
-        Normalise the given value if it's a known key (eg. dates should be dates),
-        and set it on the given dictionary.
+        Set a property with the usual normalisation.
+
+        This has some options that are not available on normal dictionary item
+        setting (``self[key] = val``)
+
+        The default behaviour of this class is very conservative in order to catch common errors
+        of users. You can loosen the settings here.
+
+        :argument allow_override: Is it okay to overwrite an existing value? (if not, error will be thrown)
+        :argument expect_override: We expect to overwrite a property, so don't produce a warning or error.
         """
-        if key not in self.KNOWN_STAC_PROPERTIES:
+        if key not in self.KNOWN_PROPERTIES:
             warnings.warn(
                 f"Unknown Stac property {key!r}. "
                 f"If this is valid property, please tell us on Github here so we can add it: "
@@ -352,7 +379,7 @@ def normalise_and_set(self, key, value, allow_override=True):
             )
 
         if value is not None:
-            normalise = self.KNOWN_STAC_PROPERTIES.get(key)
+            normalise = self.KNOWN_PROPERTIES.get(key)
             if normalise:
                 value = normalise(value)
                 # If the normaliser has extracted extra properties, we'll get two return values.
@@ -365,7 +392,7 @@ def normalise_and_set(self, key, value, allow_override=True):
                             )
                         self.normalise_and_set(k, v, allow_override=allow_override)
 
-        if key in self._props and value != self[key]:
+        if key in self._props and value != self[key] and (not expect_override):
             message = (
                 f"Overriding property {key!r} " f"(from {self[key]!r} to {value!r})"
             )
@@ -380,20 +407,47 @@ def nested(self):
         return nest_properties(self._props)
 
 
+class StacPropertyView(Eo3Dict):
+    """
+    Backwards compatibility class name. Deprecated.
+
+    Use the identical 'Eo3Dict' instead.
+
+    These were called  "StacProperties" in Stac 0.6, but many of them have
+    changed in newer versions and we're sticking to the old names for consistency
+    and backwards-compatibility. So they're now EO3 Properties.
+
+    (The eo3-to-stac tool to convert EO3 properties to real Stac properties.)
+    """
+
+    def __init__(self, properties=None) -> None:
+        super().__init__(properties)
+        warnings.warn(
+            "The class name 'StacPropertyView' is deprecated as it's misleading. "
+            "Please change your import to the (identical) 'Eo3Dict'.",
+            category=DeprecationWarning,
+        )
+
+
 class PropertyOverrideWarning(UserWarning):
     """A warning that a property was set twice with different values."""
 
     ...
 
 
-class EoFields:
+class Eo3Interface:
     """
-    Convenient access fields for the most common/essential properties in datasets
+    These are convenience properties for common metadata fields. They are available
+    on DatasetAssemblers and within other naming APIs.
+
+    (This is abstract. If you want one of these of your own, you probably want to create
+    an :class:`eodatasets3.DatasetDoc`)
+
     """
 
     @property
     @abstractmethod
-    def properties(self) -> StacPropertyView:
+    def properties(self) -> Eo3Dict:
         raise NotImplementedError
 
     @property
@@ -401,12 +455,12 @@ def platform(self) -> Optional[str]:
         """
         Unique name of the specific platform the instrument is attached to.
 
-        For satellites this would be the name of the satellite (e.g., landsat-8, sentinel-2A),
+        For satellites this would be the name of the satellite (e.g., ``landsat-8``, ``sentinel-2a``),
         whereas for drones this would be a unique name for the drone.
 
-        In derivative products, multiple platforms can be specified with a comma "landsat5,landsat7".
+        In derivative products, multiple platforms can be specified with a comma: ``landsat-5,landsat-7``.
 
-        Shorthand for 'eo:platform' property
+        Shorthand for ``eo:platform`` property
         """
         return self.properties.get("eo:platform")
 
@@ -435,7 +489,7 @@ def instrument(self) -> str:
         """
         Name of instrument or sensor used (e.g., MODIS, ASTER, OLI, Canon F-1).
 
-        Shorthand for 'eo:instrument' property
+        Shorthand for ``eo:instrument`` property
         """
         return self.properties.get("eo:instrument")
 
@@ -446,7 +500,7 @@ def instrument(self, value: str):
     @property
     def constellation(self) -> str:
         """
-        Constellation. Eg 'sentinel-2".
+        Constellation. Eg ``sentinel-2``.
         """
         return self.properties.get("eo:constellation")
 
@@ -457,7 +511,7 @@ def constellation(self, value: str):
     @property
     def product_name(self) -> Optional[str]:
         """
-        The ODC product
+        The ODC product name
         """
         return self.properties.get("odc:product")
 
@@ -470,9 +524,9 @@ def producer(self) -> str:
         """
         Organisation that produced the data.
 
-        eg. usgs.gov or ga.gov.au
+        eg. ``usgs.gov`` or ``ga.gov.au``
 
-        Shorthand for 'odc:producer' property
+        Shorthand for ``odc:producer`` property
         """
         return self.properties.get("odc:producer")
 
@@ -485,10 +539,10 @@ def datetime_range(self) -> Tuple[datetime, datetime]:
         """
         An optional date range for the dataset.
 
-        The `datetime` is still mandatory when this is set.
+        The ``datetime`` is still mandatory when this is set.
 
         This field is a shorthand for reading/setting the datetime-range
-        stac extension properties: 'dtr:start_datetime' and 'dtr:end_datetime'
+        stac 0.6 extension properties: ``dtr:start_datetime`` and ``dtr:end_datetime``
         """
         return (
             self.properties.get("dtr:start_datetime"),
@@ -502,28 +556,16 @@ def datetime_range(self, val: Tuple[datetime, datetime]):
         self.properties["dtr:start_datetime"] = start
         self.properties["dtr:end_datetime"] = end
 
-    @property
-    def datetime(self) -> datetime:
-        """
-        The searchable date and time of the assets. (Default to UTC if not specified)
-        """
-        return self.properties.get("datetime")
-
-    @datetime.setter
-    def datetime(self, val: datetime) -> datetime:
-        self.properties["datetime"] = val
-
     @property
     def processed(self) -> datetime:
-        """
-        When the dataset was processed (Default to UTC if not specified)
+        """When the dataset was created (Defaults to UTC if not specified)
 
-        Shorthand for the 'odc:processing_datetime' field
+        Shorthand for the ``odc:processing_datetime`` field
         """
         return self.properties.get("odc:processing_datetime")
 
     @processed.setter
-    def processed(self, value):
+    def processed(self, value: Union[str, datetime]):
         self.properties["odc:processing_datetime"] = value
 
     def processed_now(self):
@@ -537,7 +579,7 @@ def dataset_version(self) -> str:
         """
         The version of the dataset.
 
-        Typically digits separated by a dot. Eg. '1.0.0'
+        Typically digits separated by a dot. Eg. `1.0.0`
 
         The first digit is usually the collection number for
         this 'producer' organisation, such as USGS Collection 1 or
@@ -549,11 +591,14 @@ def dataset_version(self) -> str:
     def collection_number(self) -> int:
         """
         The version of the collection.
-        Eg:
-          metadata:
-            product_family: wofs
-            dataset_version: 1.6.0
-            collection_number: 3
+
+        Eg.::
+
+           metadata:
+             product_family: wofs
+             dataset_version: 1.6.0
+             collection_number: 3
+
         """
         return self.properties.get("odc:collection_number")
 
@@ -576,17 +621,17 @@ def naming_conventions(self, value):
     @property
     def product_family(self) -> str:
         """
-        The identifier for this "family" of products, such as 'ard', 'level1` or 'fc'.
+        The identifier for this "family" of products, such as ``ard``, ``level1`` or ``fc``.
         It's used for grouping similar products together.
 
         They products in a family are usually produced the same way but have small variations:
         they come from different sensors, or are written in different projections, etc.
 
-        'ard' family of products: 'ls7_ard', 'ls5_ard' ....
+        ``ard`` family of products: ``ls7_ard``, ``ls5_ard`` ....
 
-        On older versions of opendatacube this was called "product_type".
+        On older versions of Open Data Cube this was called ``product_type``.
 
-        Shorthand for 'odc:product_family' property.
+        Shorthand for ``odc:product_family`` property.
         """
         return self.properties.get("odc:product_family")
 
@@ -594,6 +639,10 @@ def product_family(self) -> str:
     def product_family(self, value):
         self.properties["odc:product_family"] = value
 
+    @product_family.deleter
+    def product_family(self):
+        del self.properties["odc:product_family"]
+
     @property
     def region_code(self) -> Optional[str]:
         """
@@ -603,11 +652,11 @@ def region_code(self) -> Optional[str]:
 
         It's generally treated as an opaque string to group datasets and process as stacks.
 
-        For Landsat products it's the concatenated '{path}{row}' (both numbers formatted to three digits).
+        For Landsat products it's the concatenated ``{path}{row}`` (both numbers formatted to three digits).
 
         For Sentinel 2, it's the MGRS grid (TODO presumably?).
 
-        Shorthand for 'odc:region_code' property.
+        Shorthand for ``odc:region_code`` property.
         """
         return self.properties.get("odc:region_code")
 
@@ -621,7 +670,7 @@ def maturity(self) -> str:
         The dataset maturity. The same data may be processed multiple times -- becoming more
         mature -- as new ancillary data becomes available.
 
-        Typical values (from least to most mature): "nrt", "interim", "final"
+        Typical values (from least to most mature): ``nrt`` (near real time), ``interim``, ``final``
         """
         return self.properties.get("dea:dataset_maturity")
 
@@ -629,6 +678,22 @@ def maturity(self) -> str:
     def maturity(self, value):
         self.properties["dea:dataset_maturity"] = value
 
+    # Note that giving a method the name 'datetime' will override the 'datetime' type
+    # for class-level declarations (ie, for any types on functions!)
+    # So we make an alias:
+    from datetime import datetime as datetime_
+
+    @property
+    def datetime(self) -> datetime_:
+        """
+        The searchable date and time of the assets. (Default to UTC if not specified)
+        """
+        return self.properties.get("datetime")
+
+    @datetime.setter
+    def datetime(self, val: datetime_):
+        self.properties["datetime"] = val
+
 
 def _github_suggest_new_property_url(key: str, value: object) -> str:
     """Get a URL to create a Github issue suggesting new properties to be added."""
diff --git a/eodatasets3/serialise.py b/eodatasets3/serialise.py
index 5f41b3af..615fe8f2 100644
--- a/eodatasets3/serialise.py
+++ b/eodatasets3/serialise.py
@@ -24,7 +24,7 @@
 from eodatasets3.model import (
     DatasetDoc,
     ODC_DATASET_SCHEMA_URL,
-    StacPropertyView,
+    Eo3Dict,
 )
 from eodatasets3.properties import FileFormat
 
@@ -42,7 +42,7 @@ def _uuid_representer(dumper, data):
     return dumper.represent_scalar("tag:yaml.org,2002:str", f"{data}")
 
 
-def represent_datetime(self, data: datetime):
+def _represent_datetime(self, data: datetime):
     """
     The default Ruamel representer strips 'Z' suffixes for UTC.
 
@@ -57,11 +57,11 @@ def represent_datetime(self, data: datetime):
     return self.represent_scalar("tag:yaml.org,2002:timestamp", value)
 
 
-def represent_numpy_datetime(self, data: numpy.datetime64):
-    return represent_datetime(self, data.astype("M8[ms]").tolist())
+def _represent_numpy_datetime(self, data: numpy.datetime64):
+    return _represent_datetime(self, data.astype("M8[ms]").tolist())
 
 
-def represent_paths(self, data: PurePath):
+def _represent_paths(self, data: PurePath):
     return Representer.represent_str(self, data.as_posix())
 
 
@@ -70,8 +70,8 @@ def _init_yaml() -> YAML:
 
     yaml.representer.add_representer(FileFormat, _format_representer)
     yaml.representer.add_multi_representer(UUID, _uuid_representer)
-    yaml.representer.add_representer(datetime, represent_datetime)
-    yaml.representer.add_multi_representer(PurePath, represent_paths)
+    yaml.representer.add_representer(datetime, _represent_datetime)
+    yaml.representer.add_multi_representer(PurePath, _represent_paths)
 
     # WAGL spits out many numpy primitives in docs.
     yaml.representer.add_representer(numpy.int8, Representer.represent_int)
@@ -85,7 +85,7 @@ def _init_yaml() -> YAML:
     yaml.representer.add_representer(numpy.float32, Representer.represent_float)
     yaml.representer.add_representer(numpy.float64, Representer.represent_float)
     yaml.representer.add_representer(numpy.ndarray, Representer.represent_list)
-    yaml.representer.add_representer(numpy.datetime64, represent_numpy_datetime)
+    yaml.representer.add_representer(numpy.datetime64, _represent_numpy_datetime)
 
     # Match yamllint default expectations. (Explicit start/end are recommended to tell if a file is cut off)
     yaml.width = 80
@@ -126,6 +126,13 @@ def loads_yaml(stream: Union[Text, IO]) -> Iterable[Dict]:
 
 
 def from_path(path: Path, skip_validation=False) -> DatasetDoc:
+    """
+    Parse an EO3 document from a filesystem path
+
+    :param path: Filesystem path
+    :param skip_validation: Optionally disable validation (it's faster, but I hope your
+            doc is structured correctly)
+    """
     if path.suffix.lower() not in (".yaml", ".yml"):
         raise ValueError(f"Unexpected file type {path.suffix}. Expected yaml")
 
@@ -186,10 +193,14 @@ def doc_reference(path):
 
 def from_doc(doc: Dict, skip_validation=False) -> DatasetDoc:
     """
-    Convert a document to a dataset.
+    Parse a dictionary into an EO3 dataset.
 
     By default it will validate it against the schema, which will result in far more
     useful error messages if fields are missing.
+
+    :param doc: A dictionary, such as is returned from yaml.load or json.load
+    :param skip_validation: Optionally disable validation (it's faster, but I hope your
+            doc is structured correctly)
     """
 
     if not skip_validation:
@@ -202,11 +213,11 @@ def from_doc(doc: Dict, skip_validation=False) -> DatasetDoc:
     c = cattr.Converter()
     c.register_structure_hook(uuid.UUID, _structure_as_uuid)
     c.register_structure_hook(BaseGeometry, _structure_as_shape)
-    c.register_structure_hook(StacPropertyView, _structure_as_stac_props)
+    c.register_structure_hook(Eo3Dict, _structure_as_stac_props)
 
     c.register_structure_hook(Affine, _structure_as_affine)
 
-    c.register_unstructure_hook(StacPropertyView, _unstructure_as_stac_props)
+    c.register_unstructure_hook(Eo3Dict, _unstructure_as_stac_props)
     return c.structure(doc, DatasetDoc)
 
 
@@ -215,7 +226,8 @@ def _structure_as_uuid(d, t):
 
 
 def _structure_as_stac_props(d, t):
-    return StacPropertyView(d)
+    # We don't normalise properties as we want it to reflect the original file.
+    return Eo3Dict(d, normalise_input=False)
 
 
 def _structure_as_affine(d: Tuple, t):
@@ -230,7 +242,7 @@ def _structure_as_affine(d: Tuple, t):
     return Affine(*d[:-3])
 
 
-def _unstructure_as_stac_props(v: StacPropertyView):
+def _unstructure_as_stac_props(v: Eo3Dict):
     return v._props
 
 
@@ -243,7 +255,7 @@ def to_doc(d: DatasetDoc) -> Dict:
     Serialise a DatasetDoc to a dict
 
     If you plan to write this out as a yaml file on disk, you're
-    better off with `to_formatted_doc()`.
+    better off with one of our formatted writers: :func:`.to_stream`, :func:`.to_path`.
     """
     doc = attr.asdict(
         d,
@@ -276,6 +288,24 @@ def to_formatted_doc(d: DatasetDoc) -> CommentedMap:
     return doc
 
 
+def to_path(path: Path, *ds: DatasetDoc):
+    """
+    Output dataset(s) as a formatted YAML to a local path
+
+    (multiple datasets will result in a multi-document yaml file)
+    """
+    dump_yaml(path, *(to_formatted_doc(d) for d in ds))
+
+
+def to_stream(stream, *ds: DatasetDoc):
+    """
+    Output dataset(s) as a formatted YAML to an output stream
+
+    (multiple datasets will result in a multi-document yaml file)
+    """
+    dumps_yaml(stream, *(to_formatted_doc(d) for d in ds))
+
+
 def _stac_key_order(key: str):
     """All keys in alphabetical order, but unprefixed keys first."""
     if ":" in key:
diff --git a/eodatasets3/validate.py b/eodatasets3/validate.py
index aed22265..5963db10 100644
--- a/eodatasets3/validate.py
+++ b/eodatasets3/validate.py
@@ -633,11 +633,11 @@ def validate_eo3_doc(
 
 def _validate_stac_properties(dataset: DatasetDoc):
     for name, value in dataset.properties.items():
-        if name not in dataset.properties.KNOWN_STAC_PROPERTIES:
+        if name not in dataset.properties.KNOWN_PROPERTIES:
             yield _warning("unknown_property", f"Unknown stac property {name!r}")
 
         else:
-            normaliser = dataset.properties.KNOWN_STAC_PROPERTIES.get(name)
+            normaliser = dataset.properties.KNOWN_PROPERTIES.get(name)
             if normaliser and value is not None:
                 try:
                     normalised_value = normaliser(value)
diff --git a/eodatasets3/verify.py b/eodatasets3/verify.py
index 87329137..296cc3e1 100644
--- a/eodatasets3/verify.py
+++ b/eodatasets3/verify.py
@@ -173,6 +173,9 @@ def iteratively_verify(self):
             calculated_hash = self._checksum(path)
             yield path, calculated_hash == hash_
 
+    def __bool__(self):
+        return bool(self._file_hashes)
+
     def __eq__(self, other):
         if isinstance(other, self.__class__):
             # pylint 1.6.4 isn't smart enough to know that this is protected access of the same class
diff --git a/eodatasets3/wagl.py b/eodatasets3/wagl.py
index 9b7a5894..fd2160a1 100644
--- a/eodatasets3/wagl.py
+++ b/eodatasets3/wagl.py
@@ -27,6 +27,7 @@
 from eodatasets3 import serialise, utils, images, DatasetAssembler
 from eodatasets3.images import GridSpec
 from eodatasets3.model import DatasetDoc
+from eodatasets3.properties import Eo3Interface
 from eodatasets3.serialise import loads_yaml
 from eodatasets3.ui import bool_style
 from eodatasets3.utils import default_utc
@@ -83,10 +84,11 @@ def _unpack_products(
     img_paths = _find_h5_paths(h5group, "IMAGE")
 
     for product in product_list:
-        with do(f"Starting {product}", heading=True):
+        with sub_product(product, p):
             for pathname in [
                 p for p in img_paths if "/{}/".format(product.upper()) in p
             ]:
+
                 with do(f"Path {pathname!r}"):
                     dataset = h5group[pathname]
                     band_name = utils.normalise_band_name(dataset.attrs["alias"])
@@ -102,13 +104,22 @@ def _unpack_products(
                 red, green, blue = _THUMBNAILS[product]
                 with do(f"Thumbnailing {product}"):
                     p.write_thumbnail(
-                        red, green, blue, kind=product, static_stretch=(1, 3000)
+                        red,
+                        green,
+                        blue,
+                        static_stretch=(1, 3000),
+                        # Because of our strange sub-products and filename standards, we want the
+                        # 'kind' to be included in the recorded thumbnail accessory metadata,
+                        # but not in the filename.
+                        # So we manually calculate a filename without the 'kind' field included.
+                        kind=product,
+                        path=p.names.thumbnail_filename(),
                     )
 
 
 def write_measurement_h5(
     p: DatasetAssembler,
-    name: str,
+    full_name: str,
     g: h5py.Dataset,
     overviews=images.DEFAULT_OVERVIEWS,
     overview_resampling=Resampling.nearest,
@@ -123,8 +134,8 @@ def write_measurement_h5(
     else:
         data = g
 
+    product_name, band_name = full_name.split(":")
     p.write_measurement_numpy(
-        name=name,
         array=data,
         grid_spec=images.GridSpec(
             shape=g.shape,
@@ -136,6 +147,12 @@ def write_measurement_h5(
         overview_resampling=overview_resampling,
         expand_valid_data=expand_valid_data,
         file_id=file_id,
+        # Because of our strange sub-products and filename standards, we want the
+        # product_name to be included in the recorded band metadata,
+        # but not in its filename.
+        # So we manually calculate a filename without the extra product name prefix.
+        name=full_name,
+        path=p.names.measurement_filename(band_name, "tif", file_id=file_id),
     )
 
 
@@ -301,6 +318,11 @@ def _create_contiguity(
             nodata=255,
             overviews=None,
             expand_valid_data=False,
+            # Because of our strange sub-products and filename standards, we want the
+            # 'oa_' prefix to be included in the recorded band metadata,
+            # but not in its filename.
+            # So we manually calculate a filename without the extra prefix.
+            path=p.names.measurement_filename(f"{product.lower()}_contiguity"),
         )
 
         # masking the timedelta_data with contiguity mask to get max and min timedelta within the NBAR product
@@ -346,6 +368,24 @@ def val(v: Any):
         secho("(done)")
 
 
+@contextlib.contextmanager
+def sub_product(name: str, p: Eo3Interface):
+    """
+    Set the product family temporarily within a block of code.
+
+    This is done for sub-products that WAGL contains, which have
+    a different 'family' in their filenames.
+    """
+    with do(f"Product {name}", heading=True):
+        original_family = p.product_family
+        # We delete first to show that we're deliberately changing the value (no 'overridding property" warning)
+        del p.product_family
+        p.product_family = name
+        yield
+        del p.product_family
+        p.product_family = original_family
+
+
 def _extract_reference_code(p: DatasetAssembler, granule: str) -> Optional[str]:
     matches = None
     if p.platform.startswith("landsat"):
@@ -617,22 +657,28 @@ def package(
             _unpack_products(p, included_products, granule_group)
 
             if include_oa:
-                with do("Starting OA", heading=True):
-                    _unpack_observation_attributes(
-                        p,
-                        included_products,
-                        granule_group,
-                        infer_datetime_range=p.platform.startswith("landsat"),
-                        oa_resolution=oa_resolution,
-                    )
-                if granule.fmask_image:
-                    with do(f"Writing fmask from {granule.fmask_image} "):
-                        p.write_measurement(
-                            "oa:fmask",
-                            granule.fmask_image,
-                            expand_valid_data=False,
-                            overview_resampling=Resampling.mode,
+                with sub_product("oa", p):
+                    with do("Starting OA", heading=True):
+                        _unpack_observation_attributes(
+                            p,
+                            included_products,
+                            granule_group,
+                            infer_datetime_range=p.platform.startswith("landsat"),
+                            oa_resolution=oa_resolution,
                         )
+                    if granule.fmask_image:
+                        with do(f"Writing fmask from {granule.fmask_image} "):
+                            p.write_measurement(
+                                "oa:fmask",
+                                granule.fmask_image,
+                                expand_valid_data=False,
+                                overview_resampling=Resampling.mode,
+                                # Because of our strange sub-products and filename standards, we want the
+                                # 'oa_' prefix to be included in the recorded band metadata,
+                                # but not in its filename.
+                                # So we manually calculate a filename without the extra prefix.
+                                path=p.names.measurement_filename("fmask"),
+                            )
 
             with do("Finishing package"):
                 return p.done()
diff --git a/setup.cfg b/setup.cfg
index 0fee15e6..33ad424f 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,8 +1,7 @@
 [tool:pytest]
 testpaths = eodatasets3 tests
 addopts = --doctest-modules
-# addopts = --doctest-modules --isort
-# timeout = 1000
+timeout = 1000
 filterwarnings = ignore::FutureWarning
 norecursedirs = .* build dist .git tmp*
 
diff --git a/setup.py b/setup.py
index 6c672ff5..0ed94622 100755
--- a/setup.py
+++ b/setup.py
@@ -55,6 +55,7 @@
     ],
     url="https://github.com/GeoscienceAustralia/eo-datasets",
     install_requires=[
+        "affine",
         "attrs>=18.1",  # 18.1 adds 'factory' syntactic sugar
         "boltons",
         "cattrs",


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index aee34126..139c38ec 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -25,7 +25,7 @@ jobs:
 
       - name: Run tests
         run: |
-          make test
+          make test-root
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v1
diff --git a/run-tests.sh b/run-tests.sh
new file mode 100755
index 00000000..4a5a423d
--- /dev/null
+++ b/run-tests.sh
@@ -0,0 +1,20 @@
+#!/usr/bin/env bash
+# Convenience script for running just the tests
+
+set -eu
+set -x
+
+script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )"
+
+# Run tests, taking coverage.
+# Users can specify extra folders as arguments.
+pytest --cov eodatasets3 --durations=5 "${script_dir}" $@
+
+# Run sphinx inline tests
+#
+# Unfortunately the pytest sphinx plugin doesn't work well
+# with multiple test blocks, so we run the genuine sphinx
+# tester separately instead. It will not count in coverage.
+pushd "${script_dir}/docs"
+    make doctest
+popd
diff --git a/tests/common.py b/tests/common.py
index d3ade49b..1c8efebe 100644
--- a/tests/common.py
+++ b/tests/common.py
@@ -24,10 +24,10 @@ def check_prepare_outputs(
     __tracebackhide__ = operator.methodcaller("errisinstance", AssertionError)
     run_prepare_cli(invoke_script, *run_args)
 
-    assert_expected_eo3_doc(expected_doc, expected_metadata_path, ignore_fields)
+    assert_expected_eo3_path(expected_doc, expected_metadata_path, ignore_fields)
 
 
-def assert_expected_eo3_doc(
+def assert_expected_eo3_path(
     expected_doc: Dict,
     expected_path: Path,
     ignore_fields=(),
@@ -50,9 +50,12 @@ def assert_expected_eo3_doc(
     # Compare geometry after parsing, rather than comparing the raw dict values.
     produced_dataset = serialise.from_path(expected_path)
     expected_dataset = serialise.from_doc(expected_doc, skip_validation=True)
-    assert_shapes_mostly_equal(
-        produced_dataset.geometry, expected_dataset.geometry, 0.00000001
-    )
+    if expected_dataset.geometry is None:
+        assert produced_dataset.geometry is None
+    else:
+        assert_shapes_mostly_equal(
+            produced_dataset.geometry, expected_dataset.geometry, 0.00000001
+        )
 
 
 def assert_shapes_mostly_equal(
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index cf3c0ca9..55341603 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -60,7 +60,7 @@ def l1_c2_ls8_folder(tmp_path: Path) -> Path:
 @pytest.fixture
 def l1_ls8_metadata_path(l1_ls8_folder: Path, l1_ls8_dataset: DatasetDoc) -> Path:
     path = l1_ls8_folder / f"{l1_ls8_dataset.label}.odc-metadata.yaml"
-    serialise.dump_yaml(path, serialise.to_formatted_doc(l1_ls8_dataset))
+    serialise.to_path(path, l1_ls8_dataset)
     return path
 
 
diff --git a/tests/integration/prepare/test_prepare_landsat_l1.py b/tests/integration/prepare/test_prepare_landsat_l1.py
index f6f408ce..843a5370 100644
--- a/tests/integration/prepare/test_prepare_landsat_l1.py
+++ b/tests/integration/prepare/test_prepare_landsat_l1.py
@@ -287,7 +287,7 @@ def test_skips_old_datasets(l1_ls7_tarball):
 
 
 def expected_lc08_l2_c2_post_20210507_folder(
-    l2_c2_ls8_folder=lc08_l2_c2_post_20210507_folder,
+    l2_c2_ls8_folder: Path = None,
     offset: Callable[[Path, str], str] = relative_offset,
     organisation="usgs.gov",
     collection="2",
@@ -454,7 +454,7 @@ def expected_lc08_l2_c2_post_20210507_folder(
 
 
 def expected_lt05_l2_c2_folder(
-    l2_c2_ls8_folder=lt05_l2_c2_folder,
+    l2_c2_ls8_folder: Path = None,
     offset: Callable[[Path, str], str] = relative_offset,
     organisation="usgs.gov",
     collection="2",
@@ -617,7 +617,7 @@ def expected_lt05_l2_c2_folder(
 
 
 def expected_le07_l2_c2_folder(
-    l2_c2_ls8_folder=le07_l2_c2_folder,
+    l2_c2_ls8_folder: Path = None,
     offset: Callable[[Path, str], str] = relative_offset,
     organisation="usgs.gov",
     collection="2",
diff --git a/tests/integration/prepare/test_prepare_sinergise_sentinel_l1.py b/tests/integration/prepare/test_prepare_sinergise_sentinel_l1.py
index 7d6327d5..b3e9b905 100644
--- a/tests/integration/prepare/test_prepare_sinergise_sentinel_l1.py
+++ b/tests/integration/prepare/test_prepare_sinergise_sentinel_l1.py
@@ -18,6 +18,9 @@ def expected_dataset_document():
         "$schema": "https://schemas.opendatacube.org/dataset",
         "id": "f3e0eee1-573c-5035-870e-8d8392df8e33",
         "crs": "epsg:32755",
+        "product": {
+            "name": "sinergise_s2bm_level1_1",
+        },
         "geometry": {
             "coordinates": [
                 [
@@ -102,7 +105,6 @@ def expected_dataset_document():
                 "path": "B09.jp2",
             },
         },
-        "product": {"name": "sinergise_s2bm_level1_1"},
         "properties": {
             "datetime": datetime.datetime(2020, 10, 11, 0, 6, 49, 882566),
             "eo:cloud_cover": 24.9912,
diff --git a/tests/integration/test_assemble.py b/tests/integration/test_assemble.py
index 3ef7c4f3..d13075d6 100644
--- a/tests/integration/test_assemble.py
+++ b/tests/integration/test_assemble.py
@@ -1,6 +1,12 @@
-import uuid
+"""
+Basic tests of DatasetAssembler.
+
+Some features are testsed in other sibling test files, such as alternative
+naming conventions.
+"""
 from datetime import datetime, timezone
 from pathlib import Path
+from pprint import pprint
 from textwrap import dedent
 from uuid import UUID
 
@@ -8,11 +14,14 @@
 import pytest
 from ruamel import yaml
 
-from eodatasets3 import DatasetAssembler
+from eodatasets3 import DatasetAssembler, namer, DatasetPrepare, serialise
 from eodatasets3.images import GridSpec
 from eodatasets3.model import DatasetDoc
 from tests import assert_file_structure
-from tests.common import assert_same_as_file, assert_expected_eo3_doc
+from tests.common import (
+    assert_expected_eo3_path,
+    assert_same,
+)
 
 
 def test_dea_style_package(
@@ -84,14 +93,14 @@ def test_dea_style_package(
             "ga_ls8c_ones_3-0-0_090084_2016-01-21_final_thumbnail.jpg": "",
             "ga_ls8c_ones_3-0-0_090084_2016-01-21_final.proc-info.yaml": "",
             "ga_ls8c_ones_3-0-0_090084_2016-01-21_final.sha1": "",
-            "ga_ls8c_singleband_3-0-0_090084_2016-01-21_final_thumbnail.jpg": "",
-            "ga_ls8c_singleband_lut_3-0-0_090084_2016-01-21_final_thumbnail.jpg": "",
+            "ga_ls8c_ones_3-0-0_090084_2016-01-21_final_singleband-thumbnail.jpg": "",
+            "ga_ls8c_ones_3-0-0_090084_2016-01-21_final_singleband-lut-thumbnail.jpg": "",
         },
     )
 
     # TODO: check sha1 checksum list.
 
-    assert_expected_eo3_doc(
+    assert_expected_eo3_path(
         {
             "$schema": "https://schemas.opendatacube.org/dataset",
             "id": dataset_id,
@@ -175,10 +184,10 @@ def test_dea_style_package(
                     "path": "ga_ls8c_ones_3-0-0_090084_2016-01-21_final_thumbnail.jpg"
                 },
                 "thumbnail:singleband": {
-                    "path": "ga_ls8c_singleband_3-0-0_090084_2016-01-21_final_thumbnail.jpg"
+                    "path": "ga_ls8c_ones_3-0-0_090084_2016-01-21_final_singleband-thumbnail.jpg"
                 },
                 "thumbnail:singleband_lut": {
-                    "path": "ga_ls8c_singleband_lut_3-0-0_090084_2016-01-21_final_thumbnail.jpg"
+                    "path": "ga_ls8c_ones_3-0-0_090084_2016-01-21_final_singleband-lut-thumbnail.jpg"
                 },
             },
             "lineage": {"level1": ["a780754e-a884-58a7-9ac0-df518a67f59d"]},
@@ -226,6 +235,88 @@ def test_minimal_package_with_product_name(tmp_path: Path, l1_ls8_folder: Path):
     )
 
 
+def test_in_memory_dataset(tmp_path: Path, l1_ls8_folder: Path):
+    """
+    You can create metadata fully in-memory, without touching paths.
+    """
+
+    out = tmp_path / "out"
+    out.mkdir()
+
+    [blue_geotiff_path] = l1_ls8_folder.rglob("L*_B2.TIF")
+
+    p = DatasetPrepare(dataset_location=out / "dataset.txt")
+    p.datetime = datetime(2019, 7, 4, 13, 7, 5)
+    p.product_name = "loch_ness_sightings"
+    p.processed = datetime(2019, 7, 4, 13, 8, 7)
+
+    pretend_path = out / "our_image_dont_read_it.tif"
+    p.note_measurement(
+        "blue",
+        pretend_path,
+        # We give it grid information, so it doesn't have to read it itself.
+        # (reading will fail if it tries, because the path is fake!)
+        grid=GridSpec.from_path(blue_geotiff_path),
+        pixels=numpy.ones((60, 60), numpy.int16),
+        nodata=-1,
+    )
+
+    dataset: DatasetDoc = p.to_dataset_doc()
+    doc: dict = serialise.to_doc(dataset)
+
+    # We're testing geometry calc in other tests.
+    assert doc["geometry"] is not None, "Expected geometry"
+    del doc["geometry"]
+    assert doc["id"] is not None, "Expected an id"
+    del doc["id"]
+
+    # Users can ask the generator for file names:
+    assert p.names.measurement_filename("red") == Path(
+        "loch_ness_sightings_2019-07-04_red.tif"
+    )
+
+    assert p.names.dataset_folder / p.names.measurement_filename("red") == Path(
+        "loch_ness_sightings/2019/07/04/loch_ness_sightings_2019-07-04_red.tif"
+    )
+
+    pprint(doc)
+    assert_same(
+        {
+            "$schema": "https://schemas.opendatacube.org/dataset",
+            "label": "loch_ness_sightings_2019-07-04",
+            "crs": "epsg:32655",
+            "measurements": {"blue": {"path": "our_image_dont_read_it.tif"}},
+            "product": {"name": "loch_ness_sightings"},
+            "properties": {
+                "datetime": datetime(2019, 7, 4, 13, 7, 5, tzinfo=timezone.utc),
+                "odc:processing_datetime": datetime(
+                    2019, 7, 4, 13, 8, 7, tzinfo=timezone.utc
+                ),
+                "odc:product": "loch_ness_sightings",
+            },
+            "grids": {
+                "default": {
+                    "shape": [60, 60],
+                    "transform": [
+                        3955.5,
+                        0.0,
+                        641985.0,
+                        0.0,
+                        -3975.5000000000005,
+                        -3714585.0,
+                        0.0,
+                        0.0,
+                        1.0,
+                    ],
+                }
+            },
+            "accessories": {},
+            "lineage": {},
+        },
+        doc,
+    )
+
+
 def test_minimal_generated_naming_package(tmp_path: Path, l1_ls8_folder: Path):
     """
     What's the minimum number of fields we can set and still generate file/product
@@ -250,7 +341,7 @@ def test_minimal_generated_naming_package(tmp_path: Path, l1_ls8_folder: Path):
             Assembling quaternarius (unfinished)
             - 1 measurements: blue
             - 4 properties: datetime, odc:file_format, odc:processing_datetime, odc:prod...
-            Writing to {out}
+            Writing to location: {out}/quaternarius/2019/07/04/quaternarius_2019-07-04.odc-metadata.yaml
         """
         )
 
@@ -295,285 +386,24 @@ def test_dataset_no_measurements(tmp_path: Path):
     assert doc["label"] == "chipmonk_sightings_2019", "Couldn't override label field"
 
 
-def test_minimal_s1_dataset(tmp_path: Path):
-    """A minimal dataset with sentinel-1a/b platform/instrument"""
-    with DatasetAssembler(tmp_path) as p:
-        # A custom label too.
-        p.platform = "sentinel-1a"
-        p.instrument = "c-sar"
-        p.datetime = datetime(2018, 11, 4)
-        p.product_family = "bck"
-        p.processed = "2018-11-05T12:23:23"
-
-        dataset_id, metadata_path = p.done()
-
-    with metadata_path.open("r") as f:
-        doc = yaml.YAML(typ="safe").load(f)
-
-    assert doc["label"] == "s1ac_bck_2018-11-04", "Unexpected dataset label"
-
-
-def test_minimal_s2_dataset_normal(tmp_path: Path):
-    """A minimal dataset with sentinel platform/instrument"""
-    with DatasetAssembler(tmp_path) as p:
-        # A custom label too.
-        p.platform = "sentinel-2a"
-        p.instrument = "msi"
-        p.datetime = datetime(2018, 11, 4)
-        p.product_family = "blueberries"
-        p.processed = "2018-11-05T12:23:23"
-        p.properties[
-            "sentinel:sentinel_tile_id"
-        ] = "S2A_OPER_MSI_L1C_TL_SGS__20170822T015626_A011310_T54KYU_N02.05"
-
-        dataset_id, metadata_path = p.done()
-
-    with metadata_path.open("r") as f:
-        doc = yaml.YAML(typ="safe").load(f)
-
-    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
-    assert metadata_path_offset == (
-        "s2am_blueberries/2018/11/04/s2am_blueberries_2018-11-04.odc-metadata.yaml"
-    )
-
-    assert doc["label"] == "s2am_blueberries_2018-11-04", "Unexpected dataset label"
-
-
-def test_s2_naming_conventions(tmp_path: Path):
-    """A minimal dataset with sentinel platform/instrument"""
-    p = DatasetAssembler(tmp_path, naming_conventions="dea_s2")
-    p.platform = "sentinel-2a"
-    p.instrument = "msi"
-    p.datetime = datetime(2018, 11, 4)
-    p.product_family = "blueberries"
-    p.processed = "2018-11-05T12:23:23"
-    p.producer = "ga.gov.au"
-    p.dataset_version = "1.0.0"
-    p.region_code = "Oz"
-    p.properties["odc:file_format"] = "GeoTIFF"
-    p.properties[
-        "sentinel:sentinel_tile_id"
-    ] = "S2A_OPER_MSI_L1C_TL_SGS__20170822T015626_A011310_T54KYU_N02.05"
-
-    p.note_source_datasets(
-        "telemetry",
-        # Accepts multiple, and they can be strings or UUIDs:
-        "ca705033-0fc4-4f38-a47e-f425dfb4d0c7",
-        uuid.UUID("3781e90f-b677-40af-9439-b40f6e4dfadd"),
-    )
-
-    # The property normaliser should have extracted inner fields
-    assert p.properties["sentinel:datatake_start_datetime"] == datetime(
-        2017, 8, 22, 1, 56, 26, tzinfo=timezone.utc
-    )
-
-    dataset_id, metadata_path = p.done()
-
-    # The s2 naming conventions have an extra subfolder of the datatake start time.
-    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
-
-    assert metadata_path_offset == (
-        "ga_s2am_blueberries_1/Oz/2018/11/04/20170822T015626/"
-        "ga_s2am_blueberries_1-0-0_Oz_2018-11-04.odc-metadata.yaml"
-    )
-
-    assert_same_as_file(
-        {
-            "$schema": "https://schemas.opendatacube.org/dataset",
-            "accessories": {
-                "checksum:sha1": {
-                    "path": "ga_s2am_blueberries_1-0-0_Oz_2018-11-04.sha1"
-                },
-                "metadata:processor": {
-                    "path": "ga_s2am_blueberries_1-0-0_Oz_2018-11-04.proc-info.yaml"
-                },
-            },
-            "id": dataset_id,
-            "label": "ga_s2am_blueberries_1-0-0_Oz_2018-11-04",
-            "product": {
-                "href": "https://collections.dea.ga.gov.au/product/ga_s2am_blueberries_1",
-                "name": "ga_s2am_blueberries_1",
-            },
-            "properties": {
-                "datetime": datetime(2018, 11, 4, 0, 0),
-                "eo:instrument": "msi",
-                "eo:platform": "sentinel-2a",
-                "odc:dataset_version": "1.0.0",
-                "odc:file_format": "GeoTIFF",
-                "odc:processing_datetime": datetime(2018, 11, 5, 12, 23, 23),
-                "odc:producer": "ga.gov.au",
-                "odc:product_family": "blueberries",
-                "odc:region_code": "Oz",
-                "sentinel:datatake_start_datetime": datetime(2017, 8, 22, 1, 56, 26),
-                "sentinel:sentinel_tile_id": "S2A_OPER_MSI_L1C_TL_SGS__20170822T015626_A011310_T54KYU_N02.05",
-            },
-            "lineage": {
-                "telemetry": [
-                    "ca705033-0fc4-4f38-a47e-f425dfb4d0c7",
-                    "3781e90f-b677-40af-9439-b40f6e4dfadd",
-                ]
-            },
-        },
-        generated_file=metadata_path,
-    )
-
-
-def test_complain_about_missing_fields(tmp_path: Path, l1_ls8_folder: Path):
-    """
-    It should complain immediately if I add a file without enough metadata to write the filename.
-
-    (and with a friendly error message)
-    """
-
-    out = tmp_path / "out"
-    out.mkdir()
-
-    [blue_geotiff_path] = l1_ls8_folder.rglob("L*_B2.TIF")
-
-    # Default simple naming conventions need at least a date and family...
-    with pytest.raises(
-        ValueError, match="Need more properties to fulfill naming conventions."
-    ):
-        with DatasetAssembler(out) as p:
-            p.write_measurement("blue", blue_geotiff_path)
-
-    # It should mention the field that's missing (we added a date, so product_family is needed)
-    with DatasetAssembler(out) as p:
-        with pytest.raises(ValueError, match="odc:product_family"):
-            p.datetime = datetime(2019, 7, 4, 13, 7, 5)
-            p.write_measurement("blue", blue_geotiff_path)
-
-    # DEA naming conventions should have stricter standards, and will tell your which fields you need to add.
-    with DatasetAssembler(out, naming_conventions="dea") as p:
-        # We set all the fields that work in default naming conventions.
-        p.datetime = datetime(2019, 7, 4, 13, 7, 5)
-        p.product_family = "quaternarius"
-        p.processed_now()
-
-        # These fields are mandatory for DEA, and so should be complained about.
-        expected_extra_fields_needed = (
-            "eo:platform",
-            "eo:instrument",
-            "odc:dataset_version",
-            "odc:producer",
-            "odc:region_code",
-        )
-        with pytest.raises(ValueError) as got_error:
-            p.write_measurement("blue", blue_geotiff_path)
-
-        # All needed fields should have been in the error message.
-        for needed_field_name in expected_extra_fields_needed:
-            assert needed_field_name in got_error.value.args[0], (
-                f"Expected field {needed_field_name} to "
-                f"be listed as mandatory in the error message"
-            )
-
-
-def test_dea_interim_folder_calculation(tmp_path: Path):
-    """
-    DEA Naming conventions should include maturity in the folder name
-    when it's not a 'final' dataset.
-    """
-    with DatasetAssembler(tmp_path, naming_conventions="dea") as p:
-        p.platform = "landsat-7"
-        p.instrument = "ETM+"
-        p.datetime = datetime(1998, 7, 30)
-        p.product_family = "frogs"
-        p.processed = "1999-11-20 00:00:53.152462Z"
-        p.maturity = "interim"
-        p.producer = "ga.gov.au"
-        p.properties["landsat:landsat_scene_id"] = "LE70930821999324EDC00"
-        p.dataset_version = "1.2.3"
-        p.region_code = "093082"
-
-        p.done()
-
-    [metadata_path] = tmp_path.rglob("*.odc-metadata.yaml")
-    calculated_path: Path = metadata_path.relative_to(tmp_path)
-    assert calculated_path == Path(
-        #                                  ⇩⇩⇩⇩⇩⇩⇩⇩ Adds interim flag
-        "ga_ls7e_frogs_1/093/082/1998/07/30_interim/ga_ls7e_frogs_1-2-3_093082_1998-07-30_interim.odc-metadata.yaml"
-    )
-
-
-def test_dea_c3_naming_conventions(tmp_path: Path):
-    """
-    A sample scene for Alchemist C3 processing that tests the naming conventions.
-    """
-    p = DatasetAssembler(tmp_path, naming_conventions="dea_c3")
-    p.platform = "landsat-7"
-    p.datetime = datetime(1998, 7, 30)
-    p.product_family = "wo"
-    p.processed = "1998-07-30T12:23:23"
-    p.maturity = "interim"
-    p.producer = "ga.gov.au"
-    p.region_code = "090081"
-
-    # Try missing few fields and expect ValueError
-    with pytest.raises(
-        ValueError, match="Need more properties to fulfill naming conventions."
-    ):
-        p.done()
-
-    # Put back the missed ones
-    p.dataset_version = "1.6.0"
-    p.collection_number = "3"
-
-    # Collection number returned as integer via the getter.
-    assert p.collection_number == 3
-
-    # Success case
-    dataset_id, metadata_path = p.done()
-    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
-    assert (
-        metadata_path_offset
-        == "ga_ls_wo_3/1-6-0/090/081/1998/07/30/ga_ls_wo_3_090081_1998-07-30_interim.odc-metadata.yaml"
-    )
-
+def test_dataset_given_properties(tmp_path: Path):
+    """Can we give existing properties to the assembler?"""
 
-def test_dataset_multi_platform(tmp_path: Path):
-    """Can we make a dataset derived from multiple platforms?"""
-
-    # No platform is included in names when there's a mix.
-    with DatasetAssembler(tmp_path) as p:
-        p.platforms = ["Sentinel_2a", "landsat_7"]
-        assert p.platform == "landsat-7,sentinel-2a"
-
-        p.datetime = datetime(2019, 1, 1)
-        p.product_family = "peanuts"
-        p.processed_now()
+    properties = {
+        "datetime": datetime(2019, 1, 1),
+        "odc:product_family": "chipmonk_sightings",
+        "odc:processing_datetime": "2021-06-15T01:33:43.378850",
+    }
+    names = namer(properties=properties)
+    with DatasetAssembler(tmp_path, names=names) as p:
+        # It should have normalised properties!
+        assert p.processed == datetime(2021, 6, 15, 1, 33, 43, 378850, timezone.utc)
 
         dataset_id, metadata_path = p.done()
 
-    with metadata_path.open("r") as f:
-        doc = yaml.YAML(typ="safe").load(f)
-
-    assert doc["label"] == "peanuts_2019-01-01"
-    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
-    assert (
-        metadata_path_offset
-        == "peanuts/2019/01/01/peanuts_2019-01-01.odc-metadata.yaml"
-    )
-
-    # ... but show the platform abbreviation when there's a known group.
-    with DatasetAssembler(tmp_path) as p:
-        p.platforms = ["Sentinel_2a", "sentinel_2b"]
-        assert p.platform == "sentinel-2a,sentinel-2b"
-
-        p.datetime = datetime(2019, 1, 1)
-        p.product_family = "peanuts"
-        p.processed_now()
-
-        dataset_id, metadata_path = p.done()
-
-    with metadata_path.open("r") as f:
-        doc = yaml.YAML(typ="safe").load(f)
-
-    assert doc["label"] == "s2_peanuts_2019-01-01"
-    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
-    assert (
-        metadata_path_offset
-        == "s2_peanuts/2019/01/01/s2_peanuts_2019-01-01.odc-metadata.yaml"
+    relative_path = metadata_path.relative_to(tmp_path)
+    assert relative_path == Path(
+        "chipmonk_sightings/2019/01/01/chipmonk_sightings_2019-01-01.odc-metadata.yaml"
     )
 
 
@@ -592,6 +422,10 @@ def test_add_source_dataset(tmp_path: Path, inherit_geom):
     p.add_source_dataset(
         source_dataset, auto_inherit_properties=True, inherit_geometry=inherit_geom
     )
+    if inherit_geom:
+        assert p.geometry == source_dataset.geometry
+    else:
+        assert p.geometry is None
 
     p.maturity = "interim"
     p.collection_number = "3"
@@ -614,47 +448,5 @@ def test_add_source_dataset(tmp_path: Path, inherit_geom):
     else:
         # POLYGON((684285 - 3439275, 684285 - 3444495, 689925 - 3444495, 689925 - 3439275, 684285 - 3439275))
         # Geometry is not set from the source dataset, but instead from the added wofs measurement
+        assert output.geometry is not None
         assert output.geometry != source_dataset.geometry
-
-
-def test_africa_naming_conventions(tmp_path: Path):
-    """
-    Minimal fields needed for DEAfrica naming conventions
-    """
-    with DatasetAssembler(tmp_path, naming_conventions="deafrica") as p:
-
-        # Just the fields listed in required_fields.
-        p.producer = "digitalearthafrica.org"
-        p.datetime = datetime(1998, 7, 30)
-        p.region_code = "090081"
-        p.product_family = "wofs"
-        p.platform = "LANDSAT_8"
-        p.processed_now()
-        p.dataset_version = "0.1.2"
-
-        dataset_id, metadata_path = p.done()
-
-    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
-    assert (
-        metadata_path_offset
-        == "wofs_ls/0-1-2/090/081/1998/07/30/wofs_ls_090081_1998-07-30.odc-metadata.yaml"
-    )
-
-    with DatasetAssembler(tmp_path, naming_conventions="deafrica") as p:
-
-        # Just the fields listed in required_fields.
-        p.producer = "digitalearthafrica.org"
-        p.datetime = datetime(1998, 7, 30)
-        p.region_code = "090081"
-        p.product_family = "fc"
-        p.platform = "LANDSAT_8"
-        p.processed_now()
-        p.dataset_version = "0.1.2"
-
-        dataset_id, metadata_path = p.done()
-
-    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
-    assert (
-        metadata_path_offset
-        == "fc_ls/0-1-2/090/081/1998/07/30/fc_ls_090081_1998-07-30.odc-metadata.yaml"
-    )
diff --git a/tests/integration/test_naming_conventions.py b/tests/integration/test_naming_conventions.py
new file mode 100644
index 00000000..189b6679
--- /dev/null
+++ b/tests/integration/test_naming_conventions.py
@@ -0,0 +1,461 @@
+import operator
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Mapping
+
+import pytest
+from ruamel import yaml
+
+from eodatasets3 import DatasetAssembler, DatasetDoc, namer
+from tests.common import assert_expected_eo3_path
+
+
+def assert_names_match(
+    tmp_path: Path,
+    # Given:
+    conventions,
+    properties: Mapping,
+    # Then expect:
+    expect_metadata_path: str = None,
+    expect_label: str = None,
+):
+    __tracebackhide__ = operator.methodcaller("errisinstance", AssertionError)
+    """
+    Easily test a set of naming conventions: Do certain properties lead to expected file names?
+    """
+
+    with DatasetAssembler(tmp_path, naming_conventions=conventions) as p:
+        p.properties.update(properties)
+
+        dataset_id, metadata_path = p.done()
+
+    if expect_metadata_path:
+        metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+        assert metadata_path_offset == expect_metadata_path
+
+    with metadata_path.open("r") as f:
+        doc = yaml.YAML(typ="safe").load(f)
+
+    if expect_label:
+        assert doc["label"] == expect_label, "Unexpected dataset label"
+
+
+def test_minimal_s1_dataset(tmp_path: Path):
+    assert_names_match(
+        tmp_path,
+        conventions="default",
+        properties={
+            "eo:platform": "sentinel-1a",
+            "eo:instrument": "c-sar",
+            "datetime": datetime(2018, 11, 4),
+            "odc:product_family": "bck",
+            "odc:processing_datetime": "2018-11-05T12:23:23",
+        },
+        expect_label="s1ac_bck_2018-11-04",
+        expect_metadata_path="s1ac_bck/2018/11/04/s1ac_bck_2018-11-04.odc-metadata.yaml",
+    )
+
+
+def test_dea_s2_derivate_names(tmp_path: Path):
+    assert_names_match(
+        tmp_path,
+        conventions="dea_s2_derivative",
+        properties={
+            "eo:platform": "sentinel-2a",
+            "datetime": datetime(2018, 11, 4, 5, 23, 3),
+            "odc:product_family": "eucalyptus",
+            "odc:processing_datetime": "2018-11-05T12:23:23",
+            "odc:collection_number": 3,
+            "dea:dataset_maturity": "final",
+            "odc:dataset_version": "1.2.3",
+            "odc:producer": "esa.int",
+            "odc:region_code": "55HFA",
+            "sentinel:sentinel_tile_id": "S2B_OPER_MSI_L1C_TL_EPAE_20201011T011446_A018789_T55HFA_N02.09",
+        },
+        expect_label="esa_s2_eucalyptus_3_55HFA_2018-11-04_final",
+        expect_metadata_path="esa_s2_eucalyptus_3/1-2-3/55/HFA/2018/11/04/20201011T011446/"
+        "esa_s2_eucalyptus_3_55HFA_2018-11-04_final.odc-metadata.yaml",
+    )
+
+
+def test_minimal_s2_dataset_normal(tmp_path: Path):
+    """A minimal dataset with sentinel platform/instrument"""
+    with DatasetAssembler(tmp_path) as p:
+        p.platform = "sentinel-2a"
+        p.instrument = "msi"
+        p.datetime = datetime(2018, 11, 4)
+        p.product_family = "blueberries"
+        p.processed = "2018-11-05T12:23:23"
+        p.properties[
+            "sentinel:sentinel_tile_id"
+        ] = "S2A_OPER_MSI_L1C_TL_SGS__20170822T015626_A011310_T54KYU_N02.05"
+
+        dataset_id, metadata_path = p.done()
+
+    with metadata_path.open("r") as f:
+        doc = yaml.YAML(typ="safe").load(f)
+
+    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+    assert metadata_path_offset == (
+        "s2am_blueberries/2018/11/04/s2am_blueberries_2018-11-04.odc-metadata.yaml"
+    )
+
+    assert doc["label"] == "s2am_blueberries_2018-11-04", "Unexpected dataset label"
+
+
+def test_s2_naming_conventions(tmp_path: Path):
+    """A minimal dataset with sentinel platform/instrument"""
+    p = DatasetAssembler(tmp_path, naming_conventions="dea_s2")
+    p.platform = "sentinel-2a"
+    p.instrument = "msi"
+    p.datetime = datetime(2018, 11, 4)
+    p.product_family = "blueberries"
+    p.processed = "2018-11-05T12:23:23"
+    p.producer = "ga.gov.au"
+    p.dataset_version = "1.0.0"
+    p.region_code = "Oz"
+    p.properties["odc:file_format"] = "GeoTIFF"
+    p.properties[
+        "sentinel:sentinel_tile_id"
+    ] = "S2A_OPER_MSI_L1C_TL_SGS__20170822T015626_A011310_T54KYU_N02.05"
+
+    p.note_source_datasets(
+        "telemetry",
+        # Accepts multiple, and they can be strings or UUIDs:
+        "ca705033-0fc4-4f38-a47e-f425dfb4d0c7",
+        uuid.UUID("3781e90f-b677-40af-9439-b40f6e4dfadd"),
+    )
+
+    # The property normaliser should have extracted inner fields
+    assert p.properties["sentinel:datatake_start_datetime"] == datetime(
+        2017, 8, 22, 1, 56, 26, tzinfo=timezone.utc
+    )
+
+    dataset_id, metadata_path = p.done()
+
+    # The s2 naming conventions have an extra subfolder of the datatake start time.
+    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+
+    assert metadata_path_offset == (
+        "ga_s2am_blueberries_1/Oz/2018/11/04/20170822T015626/"
+        "ga_s2am_blueberries_1-0-0_Oz_2018-11-04.odc-metadata.yaml"
+    )
+
+    assert_expected_eo3_path(
+        {
+            "$schema": "https://schemas.opendatacube.org/dataset",
+            "accessories": {},
+            "id": dataset_id,
+            "label": "ga_s2am_blueberries_1-0-0_Oz_2018-11-04",
+            "product": {
+                "href": "https://collections.dea.ga.gov.au/product/ga_s2am_blueberries_1",
+                "name": "ga_s2am_blueberries_1",
+            },
+            "properties": {
+                "datetime": datetime(2018, 11, 4, 0, 0),
+                "eo:instrument": "msi",
+                "eo:platform": "sentinel-2a",
+                "odc:dataset_version": "1.0.0",
+                "odc:file_format": "GeoTIFF",
+                "odc:processing_datetime": datetime(2018, 11, 5, 12, 23, 23),
+                "odc:producer": "ga.gov.au",
+                "odc:product_family": "blueberries",
+                "odc:region_code": "Oz",
+                "sentinel:datatake_start_datetime": datetime(2017, 8, 22, 1, 56, 26),
+                "sentinel:sentinel_tile_id": "S2A_OPER_MSI_L1C_TL_SGS__20170822T015626_A011310_T54KYU_N02.05",
+            },
+            "lineage": {
+                "telemetry": [
+                    "ca705033-0fc4-4f38-a47e-f425dfb4d0c7",
+                    "3781e90f-b677-40af-9439-b40f6e4dfadd",
+                ]
+            },
+        },
+        expected_path=metadata_path,
+    )
+
+
+def test_complain_about_missing_fields(tmp_path: Path, l1_ls8_folder: Path):
+    """
+    It should complain immediately if I add a file without enough metadata to write the filename.
+
+    (and with a friendly error message)
+    """
+
+    out = tmp_path / "out"
+    out.mkdir()
+
+    [blue_geotiff_path] = l1_ls8_folder.rglob("L*_B2.TIF")
+
+    # Default simple naming conventions need at least a date and family...
+    with pytest.raises(
+        ValueError, match="Need more properties to fulfill naming conventions."
+    ):
+        with DatasetAssembler(out) as p:
+            p.write_measurement("blue", blue_geotiff_path)
+
+    # It should mention the field that's missing (we added a date, so product_family is needed)
+    with DatasetAssembler(out) as p:
+        with pytest.raises(ValueError, match="odc:product_family"):
+            p.datetime = datetime(2019, 7, 4, 13, 7, 5)
+            p.write_measurement("blue", blue_geotiff_path)
+
+    # DEA naming conventions should have stricter standards, and will tell your which fields you need to add.
+    with DatasetAssembler(out, naming_conventions="dea") as p:
+        # We set all the fields that work in default naming conventions.
+        p.datetime = datetime(2019, 7, 4, 13, 7, 5)
+        p.product_family = "quaternarius"
+        p.processed_now()
+
+        # These fields are mandatory for DEA, and so should be complained about.
+        expected_extra_fields_needed = (
+            "eo:platform",
+            "eo:instrument",
+            "odc:dataset_version",
+            "odc:producer",
+            "odc:region_code",
+        )
+        with pytest.raises(ValueError) as got_error:
+            p.write_measurement("blue", blue_geotiff_path)
+
+        # All needed fields should have been in the error message.
+        for needed_field_name in expected_extra_fields_needed:
+            assert needed_field_name in got_error.value.args[0], (
+                f"Expected field {needed_field_name} to "
+                f"be listed as mandatory in the error message"
+            )
+
+
+def test_dea_interim_folder_calculation(tmp_path: Path):
+    """
+    DEA Naming conventions should include maturity in the folder name
+    when it's not a 'final' dataset.
+    """
+    with DatasetAssembler(tmp_path, naming_conventions="dea") as p:
+        p.platform = "landsat-7"
+        p.instrument = "ETM+"
+        p.datetime = datetime(1998, 7, 30)
+        p.product_family = "frogs"
+        p.processed = "1999-11-20 00:00:53.152462Z"
+        p.maturity = "interim"
+        p.producer = "ga.gov.au"
+        p.properties["landsat:landsat_scene_id"] = "LE70930821999324EDC00"
+        p.dataset_version = "1.2.3"
+        p.region_code = "093082"
+
+        p.done()
+
+    [metadata_path] = tmp_path.rglob("*.odc-metadata.yaml")
+    calculated_path: Path = metadata_path.relative_to(tmp_path)
+    assert calculated_path == Path(
+        #                                  ⇩⇩⇩⇩⇩⇩⇩⇩ Adds interim flag
+        "ga_ls7e_frogs_1/093/082/1998/07/30_interim/ga_ls7e_frogs_1-2-3_093082_1998-07-30_interim.odc-metadata.yaml"
+    )
+
+
+def test_dea_c3_naming_conventions(tmp_path: Path):
+    """
+    A sample scene for Alchemist C3 processing that tests the naming conventions.
+    """
+    p = DatasetAssembler(tmp_path, naming_conventions="dea_c3")
+    p.platform = "landsat-7"
+    p.datetime = datetime(1998, 7, 30)
+    p.product_family = "wo"
+    p.processed = "1998-07-30T12:23:23"
+    p.maturity = "interim"
+    p.producer = "ga.gov.au"
+    p.region_code = "090081"
+
+    # Try missing few fields and expect ValueError
+    with pytest.raises(
+        ValueError, match="Need more properties to fulfill naming conventions."
+    ):
+        p.done()
+
+    # Put back the missed ones
+    p.dataset_version = "1.6.0"
+    p.collection_number = "3"
+
+    # Collection number returned as integer via the getter.
+    assert p.collection_number == 3
+
+    # Success case
+    dataset_id, metadata_path = p.done()
+    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+    assert (
+        metadata_path_offset
+        == "ga_ls_wo_3/1-6-0/090/081/1998/07/30/ga_ls_wo_3_090081_1998-07-30_interim.odc-metadata.yaml"
+    )
+
+
+def test_dataset_multi_platform(tmp_path: Path):
+    """Can we make a dataset derived from multiple platforms?"""
+
+    # No platform is included in names when there's a mix.
+    with DatasetAssembler(tmp_path) as p:
+        p.platforms = ["Sentinel_2a", "landsat_7"]
+        assert p.platform == "landsat-7,sentinel-2a"
+
+        p.datetime = datetime(2019, 1, 1)
+        p.product_family = "peanuts"
+        p.processed_now()
+
+        dataset_id, metadata_path = p.done()
+
+    with metadata_path.open("r") as f:
+        doc = yaml.YAML(typ="safe").load(f)
+
+    assert doc["label"] == "peanuts_2019-01-01"
+    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+    assert (
+        metadata_path_offset
+        == "peanuts/2019/01/01/peanuts_2019-01-01.odc-metadata.yaml"
+    )
+
+    # ... but show the platform abbreviation when there's a known group.
+    with DatasetAssembler(tmp_path) as p:
+        p.platforms = ["Sentinel_2a", "sentinel_2b"]
+        assert p.platform == "sentinel-2a,sentinel-2b"
+
+        p.datetime = datetime(2019, 1, 1)
+        p.product_family = "peanuts"
+        p.processed_now()
+
+        dataset_id, metadata_path = p.done()
+
+    with metadata_path.open("r") as f:
+        doc = yaml.YAML(typ="safe").load(f)
+
+    assert doc["label"] == "s2_peanuts_2019-01-01"
+    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+    assert (
+        metadata_path_offset
+        == "s2_peanuts/2019/01/01/s2_peanuts_2019-01-01.odc-metadata.yaml"
+    )
+
+
+def test_africa_naming_conventions(tmp_path: Path):
+    """
+    Minimal fields needed for DEAfrica naming conventions
+    """
+    with DatasetAssembler(tmp_path, naming_conventions="deafrica") as p:
+
+        # Just the fields listed in required_fields.
+        p.producer = "digitalearthafrica.org"
+        p.datetime = datetime(1998, 7, 30)
+        p.region_code = "090081"
+        p.product_family = "wofs"
+        p.platform = "LANDSAT_8"
+        p.processed_now()
+        p.dataset_version = "0.1.2"
+
+        dataset_id, metadata_path = p.done()
+
+    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+    assert (
+        metadata_path_offset
+        == "wofs_ls/0-1-2/090/081/1998/07/30/wofs_ls_090081_1998-07-30.odc-metadata.yaml"
+    )
+
+    with DatasetAssembler(tmp_path, naming_conventions="deafrica") as p:
+
+        # Just the fields listed in required_fields.
+        p.producer = "digitalearthafrica.org"
+        p.datetime = datetime(1998, 7, 30)
+        p.region_code = "090081"
+        p.product_family = "fc"
+        p.platform = "LANDSAT_8"
+        p.processed_now()
+        p.dataset_version = "0.1.2"
+
+        dataset_id, metadata_path = p.done()
+
+    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+    assert (
+        metadata_path_offset
+        == "fc_ls/0-1-2/090/081/1998/07/30/fc_ls_090081_1998-07-30.odc-metadata.yaml"
+    )
+
+
+def test_names_alone(tmp_path: Path):
+    p = _basic_properties_set()
+    convention = namer(p, conventions="dea", collection_prefix="s3://test-bucket")
+
+    assert convention.product_name == "ga_s2am_tester_1"
+    assert convention.dataset_folder == Path("ga_s2am_tester_1/023/543/2013/02/03")
+    assert convention.metadata_filename(kind="sidecar") == Path(
+        "ga_s2am_tester_1-2-3_023543_2013-02-03_sidecar.yaml"
+    )
+    assert convention.dataset_location == (
+        "s3://test-bucket/"
+        "ga_s2am_tester_1/023/543/2013/02/03/"
+        "ga_s2am_tester_1-2-3_023543_2013-02-03.odc-metadata.yaml"
+    )
+
+    # Can we override generated names?
+
+    convention.dataset_folder = Path("custom/dataset/offset/")
+    # Now the generated metadata path will be inside it:
+    assert convention.dataset_location == (
+        "s3://test-bucket/"
+        "custom/dataset/offset/"
+        "ga_s2am_tester_1-2-3_023543_2013-02-03.odc-metadata.yaml"
+    )
+
+    # Custom product name?
+    convention.product_name = "my_custom_product"
+    assert convention.dataset_location == (
+        "s3://test-bucket/"
+        "custom/dataset/offset/"
+        "my_custom_product-2-3_023543_2013-02-03.odc-metadata.yaml"
+    )
+
+
+def test_local_path_naming(tmp_path: Path):
+    p = _basic_properties_set()
+    # The collection prefix can be given as a local path:
+
+    convention = namer(p, conventions="dea", collection_prefix=Path("/my/collections"))
+    assert convention.dataset_location == (
+        "file:///my/collections/"
+        "ga_s2am_tester_1/023/543/2013/02/03/"
+        "ga_s2am_tester_1-2-3_023543_2013-02-03.odc-metadata.yaml"
+    )
+    # We can get it as a pathlib object
+    assert convention.dataset_path == Path(
+        "/my/collections/"
+        "ga_s2am_tester_1/023/543/2013/02/03/"
+        "ga_s2am_tester_1-2-3_023543_2013-02-03.odc-metadata.yaml"
+    )
+
+
+def _basic_properties_set() -> DatasetDoc:
+    p = DatasetDoc()
+    p.platform = "sentinel-2a"
+    p.instrument = "MSI"
+    p.datetime = datetime(2013, 2, 3, 6, 5, 2)
+    p.region_code = "023543"
+    p.processed_now()
+    p.producer = "ga.gov.au"
+    p.dataset_version = "1.2.3"
+    p.product_family = "tester"
+    return p
+
+
+def test_custom_naming(tmp_path: Path):
+    """
+    We can create naming conventions separately, and later give it to assembler.
+    """
+    p = _basic_properties_set()
+    convention = namer(properties=p)
+    convention.dataset_folder = Path("my/custom/folder/")
+
+    with DatasetAssembler(tmp_path, names=convention) as a:
+        dataset_id, metadata_path = a.done()
+
+    metadata_path_offset = metadata_path.relative_to(tmp_path).as_posix()
+    assert (
+        metadata_path_offset
+        == "my/custom/folder/ga_s2am_tester_1-2-3_023543_2013-02-03.odc-metadata.yaml"
+    )
diff --git a/tests/integration/test_packagewagl.py b/tests/integration/test_packagewagl.py
index df4a0b39..07cd054f 100644
--- a/tests/integration/test_packagewagl.py
+++ b/tests/integration/test_packagewagl.py
@@ -3,10 +3,10 @@
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
-from osgeo import gdal
 import pytest
 import rasterio
 from click.testing import CliRunner
+from osgeo import gdal
 from rasterio import DatasetReader
 from rasterio.enums import Compression
 from rio_cogeo import cogeo
@@ -14,7 +14,7 @@
 import eodatasets3
 from eodatasets3.model import DatasetDoc
 from tests import assert_file_structure
-from tests.common import assert_same_as_file, assert_expected_eo3_doc
+from tests.common import assert_same_as_file, assert_expected_eo3_path
 from . import assert_image
 
 h5py = pytest.importorskip(
@@ -147,7 +147,7 @@ def test_whole_landsat_wagl_package(
     [image] = expected_folder.rglob("*_oa_*nbart-contiguity.tif")
     assert_image(image, nodata=255, unique_pixel_counts={0: 1979, 1: 4183})
 
-    assert_expected_eo3_doc(
+    assert_expected_eo3_path(
         {
             "$schema": "https://schemas.opendatacube.org/dataset",
             # A stable ID is taken from the WAGL doc.
@@ -712,7 +712,7 @@ def test_esa_sentinel_wagl_package(tmp_path: Path):
     [image] = expected_folder.rglob("*_oa_*nbart-contiguity.tif")
     assert_image(image, nodata=255, unique_pixel_counts={0: 5367, 1: 6733})
 
-    assert_expected_eo3_doc(
+    assert_expected_eo3_path(
         {
             "$schema": "https://schemas.opendatacube.org/dataset",
             "id": "14cfa990-7e2f-4f0c-bd5e-b4cb28c27e8d",
@@ -1125,7 +1125,7 @@ def test_sinergise_sentinel_wagl_package(tmp_path: Path):
     )
     [output_metadata] = expected_folder.rglob("*.odc-metadata.yaml")
 
-    assert_expected_eo3_doc(
+    assert_expected_eo3_path(
         {
             "$schema": "https://schemas.opendatacube.org/dataset",
             "id": "28f110b8-ee82-4fdd-87e1-3d126c32c5e4",
diff --git a/tests/test_properties.py b/tests/test_properties.py
index 3d2d248e..3e2c9fce 100644
--- a/tests/test_properties.py
+++ b/tests/test_properties.py
@@ -3,8 +3,12 @@
 
 import pytest
 
-from eodatasets3.model import DatasetDoc, ComplicatedNamingConventions
-from eodatasets3.properties import PropertyOverrideWarning
+from eodatasets3 import namer
+from eodatasets3.model import DatasetDoc
+from eodatasets3.names import NameGenerator
+from eodatasets3.properties import (
+    PropertyOverrideWarning,
+)
 
 
 def test_multi_platform_fields():
@@ -41,7 +45,7 @@ def ignore_property_overrides():
 
 def test_naming_abbreviations():
     d = DatasetDoc()
-    names = ComplicatedNamingConventions(d)
+    names = NameGenerator(d.properties)
 
     with ignore_property_overrides():
         assert names.platform_abbreviated is None
@@ -63,7 +67,7 @@ def test_naming_abbreviations():
 
 def test_unknown_abbreviations():
     d = DatasetDoc()
-    names = ComplicatedNamingConventions(d)
+    names = NameGenerator(d.properties)
 
     with ignore_property_overrides():
         # Unknown platforms are abbreviated by just removing dashes.
@@ -77,7 +81,7 @@ def test_unknown_abbreviations():
 
         # Unless unknown platforms aren't allowed
         # (DEA wants to be stricter and add real abbreviations for everything.)
-        names = ComplicatedNamingConventions.for_standard_dea(d)
+        names = namer(d.properties, conventions="dea")
         with pytest.raises(
             ValueError, match="don't know the DEA abbreviation for platform"
         ):