def create(dataset, annotations=None, feature=None, model='darknet-yolo',
           classes=None, batch_size=0, max_iterations=0, verbose=True,
           **kwargs):
    """
    Create a :class:`ObjectDetector` model.

    Parameters
    ----------
    dataset : SFrame
        Input data. The columns named by the ``feature`` and ``annotations``
        parameters will be extracted for training the detector.

    annotations : string
        Name of the column containing the object detection annotations.  This
        column should be a list of dictionaries (or a single dictionary), with
        each dictionary representing a bounding box of an object instance. Here
        is an example of the annotations for a single image with two object
        instances::

            [{'label': 'dog',
              'type': 'rectangle',
              'coordinates': {'x': 223, 'y': 198,
                              'width': 130, 'height': 230}},
             {'label': 'cat',
              'type': 'rectangle',
              'coordinates': {'x': 40, 'y': 73,
                              'width': 80, 'height': 123}}]

        The value for `x` is the horizontal center of the box paired with
        `width` and `y` is the vertical center of the box paired with `height`.
        'None' (the default) indicates the only list column in `dataset` should
        be used for the annotations.

    feature : string
        Name of the column containing the input images. 'None' (the default)
        indicates the only image column in `dataset` should be used as the
        feature.

    model : string optional
        Object detection model to use:

           - "darknet-yolo" : Fast and medium-sized model

    classes : list optional
        List of strings containing the names of the classes of objects.
        Inferred from the data if not provided.

    batch_size: int
        The number of images per training iteration. If 0, then it will be
        automatically determined based on resource availability.

    max_iterations : int
        The number of training iterations. If 0, then it will be automatically
        be determined based on the amount of data you provide.

    verbose : bool, optional
        If True, print progress updates and model details.

    Returns
    -------
    out : ObjectDetector
        A trained :class:`ObjectDetector` model.

    See Also
    --------
    ObjectDetector

    Examples
    --------
    .. sourcecode:: python

        # Train an object detector model
        >>> model = turicreate.object_detector.create(data)

        # Make predictions on the training set and as column to the SFrame
        >>> data['predictions'] = model.predict(data)

        # Visualize predictions by generating a new column of marked up images
        >>> data['image_pred'] = turicreate.object_detector.util.draw_bounding_boxes(data['image'], data['predictions'])
    """
    _raise_error_if_not_sframe(dataset, "dataset")
    from ._mx_detector import YOLOLoss as _YOLOLoss
    from ._model import tiny_darknet as _tiny_darknet
    from ._sframe_loader import SFrameDetectionIter as _SFrameDetectionIter
    from ._manual_scheduler import ManualScheduler as _ManualScheduler
    import mxnet as _mx
    from .._mxnet import _mxnet_utils

    if len(dataset) == 0:
        raise _ToolkitError('Unable to train on empty dataset')

    _numeric_param_check_range('max_iterations', max_iterations, 0, _six.MAXSIZE)
    start_time = _time.time()

    supported_detectors = ['darknet-yolo']

    if feature is None:
        feature = _tkutl._find_only_image_column(dataset)
        if verbose:
            print("Using '%s' as feature column" % feature)
    if annotations is None:
        annotations = _tkutl._find_only_column_of_type(dataset,
                                                       target_type=[list, dict],
                                                       type_name='list',
                                                       col_name='annotations')
        if verbose:
            print("Using '%s' as annotations column" % annotations)

    _raise_error_if_not_detection_sframe(dataset, feature, annotations,
                                         require_annotations=True)
    is_annotations_list = dataset[annotations].dtype == list

    _tkutl._check_categorical_option_type('model', model,
            supported_detectors)

    base_model = model.split('-', 1)[0]
    ref_model = _pre_trained_models.OBJECT_DETECTION_BASE_MODELS[base_model]()

    params = {
        'anchors': [
            (1.0, 2.0), (1.0, 1.0), (2.0, 1.0),
            (2.0, 4.0), (2.0, 2.0), (4.0, 2.0),
            (4.0, 8.0), (4.0, 4.0), (8.0, 4.0),
            (8.0, 16.0), (8.0, 8.0), (16.0, 8.0),
            (16.0, 32.0), (16.0, 16.0), (32.0, 16.0),
        ],
        'grid_shape': [13, 13],
        'aug_resize': 0,
        'aug_rand_crop': 0.9,
        'aug_rand_pad': 0.9,
        'aug_rand_gray': 0.0,
        'aug_aspect_ratio': 1.25,
        'aug_hue': 0.05,
        'aug_brightness': 0.05,
        'aug_saturation': 0.05,
        'aug_contrast': 0.05,
        'aug_horizontal_flip': True,
        'aug_min_object_covered': 0,
        'aug_min_eject_coverage': 0.5,
        'aug_area_range': (.15, 2),
        'aug_pca_noise': 0.0,
        'aug_max_attempts': 20,
        'aug_inter_method': 2,
        'lmb_coord_xy': 10.0,
        'lmb_coord_wh': 10.0,
        'lmb_obj': 100.0,
        'lmb_noobj': 5.0,
        'lmb_class': 2.0,
        'non_maximum_suppression_threshold': 0.45,
        'rescore': True,
        'clip_gradients': 0.025,
        'weight_decay': 0.0005,
        'sgd_momentum': 0.9,
        'learning_rate': 1.0e-3,
        'shuffle': True,
        'mps_loss_mult': 8,
        # This large buffer size (8 batches) is an attempt to mitigate against
        # the SFrame shuffle operation that can occur after each epoch.
        'io_thread_buffer_size': 8,
    }

    if '_advanced_parameters' in kwargs:
        # Make sure no additional parameters are provided
        new_keys = set(kwargs['_advanced_parameters'].keys())
        set_keys = set(params.keys()) 
        unsupported = new_keys - set_keys
        if unsupported:
            raise _ToolkitError('Unknown advanced parameters: {}'.format(unsupported))

        params.update(kwargs['_advanced_parameters'])

    anchors = params['anchors']
    num_anchors = len(anchors)

    if batch_size < 1:
        batch_size = 32  # Default if not user-specified
    cuda_gpus = _mxnet_utils.get_gpus_in_use(max_devices=batch_size)
    num_mxnet_gpus = len(cuda_gpus)
    use_mps = _use_mps() and num_mxnet_gpus == 0
    batch_size_each = batch_size // max(num_mxnet_gpus, 1)
    if use_mps and _mps_device_memory_limit() < 4 * 1024 * 1024 * 1024:
        # Reduce batch size for GPUs with less than 4GB RAM
        batch_size_each = 16
    # Note, this may slightly alter the batch size to fit evenly on the GPUs
    batch_size = max(num_mxnet_gpus, 1) * batch_size_each
    if verbose:
        print("Setting 'batch_size' to {}".format(batch_size))


    # The IO thread also handles MXNet-powered data augmentation. This seems
    # to be problematic to run independently of a MXNet-powered neural network
    # in a separate thread. For this reason, we restrict IO threads to when
    # the neural network backend is MPS.
    io_thread_buffer_size = params['io_thread_buffer_size'] if use_mps else 0

    if verbose:
        # Estimate memory usage (based on experiments)
        cuda_mem_req = 550 + batch_size_each * 85

        _tkutl._print_neural_compute_device(cuda_gpus=cuda_gpus, use_mps=use_mps,
                                            cuda_mem_req=cuda_mem_req)

    grid_shape = params['grid_shape']
    input_image_shape = (3,
                         grid_shape[0] * ref_model.spatial_reduction,
                         grid_shape[1] * ref_model.spatial_reduction)

    try:
        if is_annotations_list:
            instances = (dataset.stack(annotations, new_column_name='_bbox', drop_na=True)
                            .unpack('_bbox', limit=['label']))
        else:
            instances = dataset.rename({annotations: '_bbox'}).dropna('_bbox')
            instances = instances.unpack('_bbox', limit=['label'])

    except (TypeError, RuntimeError):
        # If this fails, the annotation format isinvalid at the coarsest level
        raise _ToolkitError("Annotations format is invalid. Must be a list of "
           "dictionaries or single dictionary containing 'label' and 'coordinates'.")

    num_images = len(dataset)
    num_instances = len(instances)
    if classes is None:
        classes = instances['_bbox.label'].unique()
    classes = sorted(classes)

    # Make a class-to-index look-up table
    class_to_index = {name: index for index, name in enumerate(classes)}
    num_classes = len(classes)

    if max_iterations == 0:
        # Set number of iterations through a heuristic
        num_iterations_raw = 5000 * _np.sqrt(num_instances) / batch_size
        num_iterations = 1000 * max(1, int(round(num_iterations_raw / 1000)))
        if verbose:
            print("Setting 'max_iterations' to {}".format(num_iterations))
    else:
        num_iterations = max_iterations

    # Create data loader
    loader = _SFrameDetectionIter(dataset,
                                  batch_size=batch_size,
                                  input_shape=input_image_shape[1:],
                                  output_shape=grid_shape,
                                  anchors=anchors,
                                  class_to_index=class_to_index,
                                  aug_params=params,
                                  shuffle=params['shuffle'],
                                  loader_type='augmented',
                                  feature_column=feature,
                                  annotations_column=annotations,
                                  io_thread_buffer_size=io_thread_buffer_size,
                                  iterations=num_iterations)

    # Predictions per anchor box: x/y + w/h + object confidence + class probs
    preds_per_box = 5 + num_classes
    output_size = preds_per_box * num_anchors
    ymap_shape = (batch_size_each,) + tuple(grid_shape) + (num_anchors, preds_per_box)

    net = _tiny_darknet(output_size=output_size)

    loss = _YOLOLoss(input_shape=input_image_shape[1:],
                     output_shape=grid_shape,
                     batch_size=batch_size_each,
                     num_classes=num_classes,
                     anchors=anchors,
                     parameters=params)

    base_lr = params['learning_rate']
    steps = [num_iterations // 2, 3 * num_iterations // 4, num_iterations]
    steps_and_factors = [(step, 10**(-i)) for i, step in enumerate(steps)]

    steps, factors = zip(*steps_and_factors)
    lr_scheduler = _ManualScheduler(step=steps, factor=factors)

    ctx = _mxnet_utils.get_mxnet_context(max_devices=batch_size)

    net_params = net.collect_params()
    net_params.initialize(_mx.init.Xavier(), ctx=ctx)
    net_params['conv7_weight'].initialize(_mx.init.Xavier(factor_type='avg'), ctx=ctx, force_reinit=True)
    net_params['conv8_weight'].initialize(_mx.init.Uniform(0.00005), ctx=ctx, force_reinit=True)
    # Initialize object confidence low, preventing an unnecessary adjustment
    # period toward conservative estimates
    bias = _np.zeros(output_size, dtype=_np.float32)
    bias[4::preds_per_box] -= 6
    from ._mx_detector import ConstantArray
    net_params['conv8_bias'].initialize(ConstantArray(bias), ctx, force_reinit=True)

    # Take a subset and then load the rest of the parameters. It is possible to
    # do allow_missing=True directly on net_params. However, this will more
    # easily hide bugs caused by names getting out of sync.
    ref_model.available_parameters_subset(net_params).load(ref_model.model_path, ctx)

    column_names = ['Iteration', 'Loss', 'Elapsed Time']
    num_columns = len(column_names)
    column_width = max(map(lambda x: len(x), column_names)) + 2
    hr = '+' + '+'.join(['-' * column_width] * num_columns) + '+'

    progress = {'smoothed_loss': None, 'last_time': 0}
    iteration = 0

    def update_progress(cur_loss, iteration):
        iteration_base1 = iteration + 1
        if progress['smoothed_loss'] is None:
            progress['smoothed_loss'] = cur_loss
        else:
            progress['smoothed_loss'] = 0.9 * progress['smoothed_loss'] + 0.1 * cur_loss
        cur_time = _time.time()

        # Printing of table header is deferred, so that start-of-training
        # warnings appear above the table
        if verbose and iteration == 0:
            # Print progress table header
            print(hr)
            print(('| {:<{width}}' * num_columns + '|').format(*column_names, width=column_width-1))
            print(hr)

        if verbose and (cur_time > progress['last_time'] + 10 or
                        iteration_base1 == max_iterations):
            # Print progress table row
            elapsed_time = cur_time - start_time
            print("| {cur_iter:<{width}}| {loss:<{width}.3f}| {time:<{width}.1f}|".format(
                cur_iter=iteration_base1, loss=progress['smoothed_loss'],
                time=elapsed_time , width=column_width-1))
            progress['last_time'] = cur_time

    if use_mps:
        # Force initialization of net_params
        # TODO: Do not rely on MXNet to initialize MPS-based network
        net.forward(_mx.nd.uniform(0, 1, (batch_size_each,) + input_image_shape))
        mps_net_params = {}
        keys = list(net_params)
        for k in keys:
            mps_net_params[k] = net_params[k].data().asnumpy()

        # Multiplies the loss to move the fp16 gradients away from subnormals
        # and gradual underflow. The learning rate is correspondingly divided
        # by the same multiple to make training mathematically equivalent. The
        # update is done in fp32, which is why this trick works. Does not
        # affect how loss is presented to the user.
        mps_loss_mult = params['mps_loss_mult']

        mps_config = {
            'mode': _MpsGraphMode.Train,
            'use_sgd': True,
            'learning_rate': base_lr / params['mps_loss_mult'],
            'gradient_clipping': params.get('clip_gradients', 0.0) * mps_loss_mult,
            'weight_decay': params['weight_decay'],
            'od_include_network': True,
            'od_include_loss': True,
            'od_scale_xy': params['lmb_coord_xy'] * mps_loss_mult,
            'od_scale_wh': params['lmb_coord_wh'] * mps_loss_mult,
            'od_scale_no_object': params['lmb_noobj'] * mps_loss_mult,
            'od_scale_object': params['lmb_obj'] * mps_loss_mult,
            'od_scale_class': params['lmb_class'] * mps_loss_mult,
            'od_max_iou_for_no_object': 0.3,
            'od_min_iou_for_object': 0.7,
            'od_rescore': params['rescore'],
        }

        mps_net = _get_mps_od_net(input_image_shape=input_image_shape,
                                  batch_size=batch_size,
                                  output_size=output_size,
                                  anchors=anchors,
                                  config=mps_config,
                                  weights=mps_net_params)

        # Use worker threads to isolate different points of synchronization
        # and/or waiting for non-Python tasks to finish. The
        # sframe_worker_thread will spend most of its time waiting for SFrame
        # operations, largely image I/O and decoding, along with scheduling
        # MXNet data augmentation. The numpy_worker_thread will spend most of
        # its time waiting for MXNet data augmentation to complete, along with
        # copying the results into NumPy arrays. Finally, the main thread will
        # spend most of its time copying NumPy data into MPS and waiting for the
        # results. Note that using three threads here only makes sense because
        # each thread spends time waiting for non-Python code to finish (so that
        # no thread hogs the global interpreter lock).
        mxnet_batch_queue = _Queue(1)
        numpy_batch_queue = _Queue(1)
        def sframe_worker():
            # Once a batch is loaded into NumPy, pass it immediately to the
            # numpy_worker so that we can start I/O and decoding for the next
            # batch.
            for batch in loader:
                mxnet_batch_queue.put(batch)
            mxnet_batch_queue.put(None)
        def numpy_worker():
            while True:
                batch = mxnet_batch_queue.get()
                if batch is None:
                    break

                for x, y in zip(batch.data, batch.label):
                    # Convert to NumPy arrays with required shapes. Note that
                    # asnumpy waits for any pending MXNet operations to finish.
                    input_data = _mxnet_to_mps(x.asnumpy())
                    label_data = y.asnumpy().reshape(y.shape[:-2] + (-1,))

                    # Convert to packed 32-bit arrays.
                    input_data = input_data.astype(_np.float32)
                    if not input_data.flags.c_contiguous:
                        input_data = input_data.copy()
                    label_data = label_data.astype(_np.float32)
                    if not label_data.flags.c_contiguous:
                        label_data = label_data.copy()

                    # Push this batch to the main thread.
                    numpy_batch_queue.put({'input'     : input_data,
                                           'label'     : label_data,
                                           'iteration' : batch.iteration})
            # Tell the main thread there's no more data.
            numpy_batch_queue.put(None)
        sframe_worker_thread = _Thread(target=sframe_worker)
        sframe_worker_thread.start()
        numpy_worker_thread = _Thread(target=numpy_worker)
        numpy_worker_thread.start()

        batch_queue = []
        def wait_for_batch():
            pending_loss = batch_queue.pop(0)
            batch_loss = pending_loss.asnumpy()  # Waits for the batch to finish
            return batch_loss.sum() / mps_loss_mult

        while True:
            batch = numpy_batch_queue.get()
            if batch is None:
                break

            # Adjust learning rate according to our schedule.
            if batch['iteration'] in steps:
                ii = steps.index(batch['iteration']) + 1
                new_lr = factors[ii] * base_lr
                mps_net.set_learning_rate(new_lr / mps_loss_mult)

            # Submit this match to MPS.
            batch_queue.append(mps_net.train(batch['input'], batch['label']))

            # If we have two batches in flight, wait for the first one.
            if len(batch_queue) > 1:
                cur_loss = wait_for_batch()

                # If we just submitted the first batch of an iteration, update
                # progress for the iteration completed by the last batch we just
                # waited for.
                if batch['iteration'] > iteration:
                    update_progress(cur_loss, iteration)
            iteration = batch['iteration']

        # Wait for any pending batches and finalize our progress updates.
        while len(batch_queue) > 0:
            cur_loss = wait_for_batch()
        update_progress(cur_loss, iteration)

        sframe_worker_thread.join()
        numpy_worker_thread.join()

        # Load back into mxnet
        mps_net_params = mps_net.export()
        keys = mps_net_params.keys()
        for k in keys:
            if k in net_params:
                net_params[k].set_data(mps_net_params[k])

    else:  # Use MxNet
        net.hybridize()

        options = {'learning_rate': base_lr, 'lr_scheduler': lr_scheduler,
                   'momentum': params['sgd_momentum'], 'wd': params['weight_decay'], 'rescale_grad': 1.0}
        clip_grad = params.get('clip_gradients')
        if clip_grad:
            options['clip_gradient'] = clip_grad
        trainer = _mx.gluon.Trainer(net.collect_params(), 'sgd', options)

        for batch in loader:
            data = _mx.gluon.utils.split_and_load(batch.data[0], ctx_list=ctx, batch_axis=0)
            label = _mx.gluon.utils.split_and_load(batch.label[0], ctx_list=ctx, batch_axis=0)

            Ls = []
            Zs = []

            with _mx.autograd.record():
                for x, y in zip(data, label):
                    z = net(x)
                    z0 = _mx.nd.transpose(z, [0, 2, 3, 1]).reshape(ymap_shape)
                    L = loss(z0, y)
                    Ls.append(L)
                for L in Ls:
                    L.backward()

            trainer.step(1)
            cur_loss = _np.mean([L.asnumpy()[0] for L in Ls])

            update_progress(cur_loss, batch.iteration)
            iteration = batch.iteration

    training_time = _time.time() - start_time
    if verbose:
        print(hr)   # progress table footer

    # Save the model
    training_iterations = iteration + 1
    state = {
        '_model': net,
        '_class_to_index': class_to_index,
        '_training_time_as_string': _seconds_as_string(training_time),
        '_grid_shape': grid_shape,
        'anchors': anchors,
        'model': model,
        'classes': classes,
        'batch_size': batch_size,
        'input_image_shape': input_image_shape,
        'feature': feature,
        'non_maximum_suppression_threshold': params['non_maximum_suppression_threshold'],
        'annotations': annotations,
        'num_classes': num_classes,
        'num_examples': num_images,
        'num_bounding_boxes': num_instances,
        'training_time': training_time,
        'training_epochs': training_iterations * batch_size // num_images,
        'training_iterations': training_iterations,
        'max_iterations': max_iterations,
        'training_loss': progress['smoothed_loss'],
    }
    return ObjectDetector(state)