# Path: moonlight/staves/staffline_extractor.py
# DEFAULT_TARGET_HEIGHT = 18
# DEFAULT_NUM_SECTIONS = 19
# DEFAULT_STAFFLINE_DISTANCE_MULTIPLE = 3
#   STAFF = 0
#   POSITION = 1
#   Y = 2
#   X = 3
# class Axes(enum.IntEnum):
# class StafflineExtractor(object):
# class StafflinePatchExtractor(object):
# def get_staffline(y_position, extracted_staff_arr):
# def y_position_to_index(y_position, num_stafflines):
#   def __init__(self,
#                image,
#                staves,
#                target_height=DEFAULT_TARGET_HEIGHT,
#                num_sections=DEFAULT_NUM_SECTIONS,
#                staffline_distance_multiple=DEFAULT_STAFFLINE_DISTANCE_MULTIPLE):
#   def extract_staves(self):
#     def do_extract_staves():
#       def extract_staff(i):
#         def extract_staffline_by_index(j):
#   def _extract_staffline(self, staff_y, staffline_distance, staffline_num):
#   def _get_resized_width(self, staffline_distance):
#   def _get_staffline_window_size(self, staffline_distance):
#   def __init__(self,
#                num_sections=DEFAULT_NUM_SECTIONS,
#                patch_height=15,
#                patch_width=12,
#                run_options=None):
#   def extract_staff_strip(self, filename, staff_index, y_position):
#   def extract_staff_patch(self, filename, staff_index, y_position, image_x):
#   def page_patch_iterator(self, filename):
#     def generator():
#   def make_patch_id(filename, staff_index, y_position, image_x):
# 
# Path: moonlight/training/clustering/staffline_patches_dofn.py
# def _filter_patch(patch, min_num_dark_pixels=10):
#   def __init__(self, patch_height, patch_width, num_stafflines, timeout_ms,
#                max_patches_per_page):
#   def start_bundle(self):
#   def process(self, png_path):
#   def finish_bundle(self):
# class StafflinePatchesDoFn(beam.DoFn):


import os.path
import tempfile
import apache_beam as beam
import tensorflow as tf
from absl.testing import absltest
from tensorflow.python.lib.io import tf_record
from moonlight.staves import staffline_extractor
from moonlight.training.clustering import staffline_patches_dofn




PATCH_HEIGHT = 9
PATCH_WIDTH = 7
NUM_STAFFLINES = 9
TIMEOUT_MS = 60000
MAX_PATCHES_PER_PAGE = 10


class StafflinePatchesDoFnTest(absltest.TestCase):

  def testPipeline_corpusImage(self):
    filename = os.path.join(tf.resource_loader.get_data_files_path(),
                            '../../testdata/IMSLP00747-000.png')
    with tempfile.NamedTemporaryFile() as output_examples:
      # Run the pipeline to get the staffline patches.
      with beam.Pipeline() as pipeline:
        dofn = staffline_patches_dofn.StafflinePatchesDoFn(
            PATCH_HEIGHT, PATCH_WIDTH, NUM_STAFFLINES, TIMEOUT_MS,
            MAX_PATCHES_PER_PAGE)
        # pylint: disable=expression-not-assigned
        (pipeline | beam.transforms.Create([filename])
         | beam.transforms.ParDo(dofn) | beam.io.WriteToTFRecord(
             output_examples.name,
             beam.coders.ProtoCoder(tf.train.Example),
             shard_name_template=''))
      # Get the staffline images from a local TensorFlow session.
      extractor = staffline_extractor.StafflinePatchExtractor(

# Path: moonlight/staves/staffline_extractor.py
# DEFAULT_TARGET_HEIGHT = 18
# DEFAULT_NUM_SECTIONS = 19
# DEFAULT_STAFFLINE_DISTANCE_MULTIPLE = 3
#   STAFF = 0
#   POSITION = 1
#   Y = 2
#   X = 3
# class Axes(enum.IntEnum):
# class StafflineExtractor(object):
# class StafflinePatchExtractor(object):
# def get_staffline(y_position, extracted_staff_arr):
# def y_position_to_index(y_position, num_stafflines):
#   def __init__(self,
#                image,
#                staves,
#                target_height=DEFAULT_TARGET_HEIGHT,
#                num_sections=DEFAULT_NUM_SECTIONS,
#                staffline_distance_multiple=DEFAULT_STAFFLINE_DISTANCE_MULTIPLE):
#   def extract_staves(self):
#     def do_extract_staves():
#       def extract_staff(i):
#         def extract_staffline_by_index(j):
#   def _extract_staffline(self, staff_y, staffline_distance, staffline_num):
#   def _get_resized_width(self, staffline_distance):
#   def _get_staffline_window_size(self, staffline_distance):
#   def __init__(self,
#                num_sections=DEFAULT_NUM_SECTIONS,
#                patch_height=15,
#                patch_width=12,
#                run_options=None):
#   def extract_staff_strip(self, filename, staff_index, y_position):
#   def extract_staff_patch(self, filename, staff_index, y_position, image_x):
#   def page_patch_iterator(self, filename):
#     def generator():
#   def make_patch_id(filename, staff_index, y_position, image_x):
# 
# Path: moonlight/training/clustering/staffline_patches_dofn.py
# def _filter_patch(patch, min_num_dark_pixels=10):
#   def __init__(self, patch_height, patch_width, num_stafflines, timeout_ms,
#                max_patches_per_page):
#   def start_bundle(self):
#   def process(self, png_path):
#   def finish_bundle(self):
# class StafflinePatchesDoFn(beam.DoFn):


# Path: moonlight/training/clustering/staffline_patches_dofn_test.py
import os.path
import tempfile
import apache_beam as beam
import tensorflow as tf
from absl.testing import absltest
from tensorflow.python.lib.io import tf_record
from moonlight.staves import staffline_extractor
from moonlight.training.clustering import staffline_patches_dofn





PATCH_HEIGHT = 9
PATCH_WIDTH = 7
NUM_STAFFLINES = 9
TIMEOUT_MS = 60000
MAX_PATCHES_PER_PAGE = 10


class StafflinePatchesDoFnTest(absltest.TestCase):

  def testPipeline_corpusImage(self):
    filename = os.path.join(tf.resource_loader.get_data_files_path(),
                            '../../testdata/IMSLP00747-000.png')
    with tempfile.NamedTemporaryFile() as output_examples:
      # Run the pipeline to get the staffline patches.
      with beam.Pipeline() as pipeline:
        dofn = staffline_patches_dofn.StafflinePatchesDoFn(
            PATCH_HEIGHT, PATCH_WIDTH, NUM_STAFFLINES, TIMEOUT_MS,
            MAX_PATCHES_PER_PAGE)
        # pylint: disable=expression-not-assigned
        (pipeline | beam.transforms.Create([filename])
         | beam.transforms.ParDo(dofn) | beam.io.WriteToTFRecord(
             output_examples.name,
             beam.coders.ProtoCoder(tf.train.Example),
             shard_name_template=''))
      # Get the staffline images from a local TensorFlow session.
      extractor = staffline_extractor.StafflinePatchExtractor(

          staffline_extractor.DEFAULT_NUM_SECTIONS, PATCH_HEIGHT, PATCH_WIDTH)