<commit_before># -*- encoding: utf-8 -*-
from __future__ import print_function
import itertools
from abjad.tools import abctools
from abjad.tools import datastructuretools
from abjad.tools import durationtools
from abjad.tools import metertools
from abjad.tools import scoretools
from abjad.tools import timespantools
from abjad.tools.topleveltools import inspect_
from abjad.tools.topleveltools import mutate
from abjad.tools.topleveltools import new
from supriya.tools import timetools


class TimeManager(abctools.AbjadValueObject):

    ### PUBLIC METHODS ###

    @staticmethod
    def consolidate_demultiplexed_timespans(demultiplexed_timespans):
        for voice_name in demultiplexed_timespans:
            timespans = demultiplexed_timespans[voice_name]
            consolidated_timespans = TimeManager.consolidate_timespans(
                timespans)
            demultiplexed_timespans[voice_name] = consolidated_timespans

    @staticmethod
    def consolidate_timespans(timespans):
        consolidated_timespans = timespantools.TimespanInventory()
        for music_specifier, grouped_timespans in \
            TimeManager.group_timespans(timespans):
            if music_specifier is None:
                continue
            divisions = tuple(_.duration for _ in grouped_timespans)
            first_timespan = grouped_timespans[0]
            last_timespan = grouped_timespans[-1]
            consolidated_timespan = new(
                first_timespan,
                divisions=divisions,
                stop_offset=last_timespan.stop_offset,
                original_stop_offset=last_timespan.original_stop_offset,
                )
            consolidated_timespans.append(consolidated_timespan)
        consolidated_timespans.sort()
        return consolidated_timespans

    @staticmethod
    def demultiplex_timespans(multiplexed_timespans):
        demultiplexed_timespans = {}
        for timespan in multiplexed_timespans:
            voice_name, layer = timespan.voice_name, timespan.layer
            if voice_name not in demultiplexed_timespans:
                demultiplexed_timespans[voice_name] = {}
            if layer not in demultiplexed_timespans[voice_name]:
                demultiplexed_timespans[voice_name][layer] = \
                    timespantools.TimespanInventory()
            demultiplexed_timespans[voice_name][layer].append(
                timespan)
            demultiplexed_timespans[voice_name][layer]
        for voice_name in demultiplexed_timespans:
            timespan_inventories = demultiplexed_timespans[voice_name]
            timespan_inventory = \
                TimeManager.resolve_timespan_inventories(
                    timespan_inventories)
            demultiplexed_timespans[voice_name] = timespan_inventory
        return demultiplexed_timespans

    @staticmethod
    def execute(
        discard_final_silence=None,
        permitted_time_signatures=None,
        segment_session=None,
        target_duration=None,
        score_template=None,
        settings=None,
        ):

        score = score_template()
        multiplexed_timespans = timespantools.TimespanInventory()

        # populate independent timespans
        TimeManager._populate_multiplexed_timespans(
            dependent=False,
            score=score,
            score_template=score_template,
            settings=settings,
            target_duration=target_duration,
            timespan_inventory=multiplexed_timespans,
            )

        # find meters
        segment_session.meters = TimeManager.find_meters(
            discard_final_silence=discard_final_silence,
            permitted_time_signatures=permitted_time_signatures,
            target_duration=target_duration,
            timespan_inventory=multiplexed_timespans,
            )

        # demultiplex
        demultiplexed_timespans = TimeManager.demultiplex_timespans(
            multiplexed_timespans)

        # split performed timespans by meter offsets
        TimeManager.split_demultiplexed_timespans(
            segment_session.measure_offsets,
            demultiplexed_timespans,
            )

        # consolidate performed timespans by music specifier
        TimeManager.consolidate_demultiplexed_timespans(
            segment_session.measure_offsets,
            demultiplexed_timespans,
            )

        # inscribe grouped performed timespans with rhythms

            # rebuild performed timespans without silence divisions
            # (the inscription process should do the rebuilding)

        # make dependent timespans

        # make (pre-split) silent timespans

        # rewrite meters? (magic)

        # populate score

        # perform other rhythmic processing

        # collect attack points

    @staticmethod
    def find_meters(
        discard_final_silence=None,
        permitted_time_signatures=None,
        target_duration=None,
        timespan_inventory=None,
        ):
        offset_counter = datastructuretools.TypedCounter(
            item_class=durationtools.Offset,
            )
        for timespan in timespan_inventory:
            offset_counter[timespan.start_offset] += 2
            offset_counter[timespan.stop_offset] += 1
        offset_counter[target_duration] += 100
        if discard_final_silence is None:
            discard_final_silence = False
        else:
            discard_final_silence = bool(discard_final_silence)
        meters = metertools.Meter.fit_meters_to_expr(
            offset_counter,
            permitted_time_signatures,
            discard_final_orphan_downbeat=discard_final_silence,
            maximum_repetitions=2,
            )
        return tuple(meters)

    @staticmethod
    def group_timespans(timespans):
        def grouper(timespan):
            music_specifier = None
            if isinstance(timespan, consort.PerformedTimespan):
                music_specifier = timespan.music_specifier
                if music_specifier is None:
                    music_specifier = consort.MusicSpecifier()
            return music_specifier
        import consort
        for music_specifier, grouped_timespans in itertools.groupby(
            timespans, grouper):
            grouped_timespans = timespantools.TimespanInventory(
                grouped_timespans)
            yield music_specifier, grouped_timespans

    @staticmethod
    def make_simple_music(rhythm_maker, durations, seed):
        music = rhythm_maker(durations, seeds=seed)
        for i, x in enumerate(music):
            if len(x) == 1 and isinstance(x[0], scoretools.Tuplet):
                music[i] = x[0]
            else:
                music[i] = scoretools.Container(x)
        music = scoretools.Container(music)
        for x in music[:]:
            if isinstance(x, scoretools.Tuplet) and x.multiplier == 1:
                mutate(x).swap(scoretools.Container())
        return music

    @staticmethod
    def populate_timespan(timespan, seed=None):
        rhythm_maker = timespan.music_specifier.rhythm_maker
        durations = timespan.divisions[:]
        music = TimeManager.make_simple_music(
            rhythm_maker,
            durations,
            seed,
            )
        assert inspect_(music).get_duration() == timespan.duration

    @staticmethod
    def populate_multiplexed_timespans(
        dependent=False,
        score=None,
        score_template=None,
        settings=None,
        target_duration=None,
        timespan_inventory=None,
        ):
        target_timespan = timespantools.Timespan(0, target_duration)
        if timespan_inventory is None:
            timespan_inventory = timespantools.TimespanInventory()
        independent_settings = [setting for setting in settings
            if not setting.timespan_maker.is_dependent
            ]
        dependent_settings = [setting for setting in settings
            if setting.timespan_maker.is_dependent
            ]
        if dependent:
            settings = dependent_settings
            start_index = len(independent_settings)
        else:
            settings = independent_settings
            start_index = 0
        for layer, music_setting in enumerate(settings, start_index):
            music_setting(
                layer=layer,
                score=score,<commit_msg>Fleshed out TimeManager.<commit_after># -*- encoding: utf-8 -*-
from __future__ import print_function
import collections
import itertools
from abjad.tools import abctools
from abjad.tools import datastructuretools
from abjad.tools import durationtools
from abjad.tools import metertools
from abjad.tools import rhythmmakertools
from abjad.tools import scoretools
from abjad.tools import spannertools
from abjad.tools import timespantools
from abjad.tools.topleveltools import attach
from abjad.tools.topleveltools import inspect_
from abjad.tools.topleveltools import mutate
from abjad.tools.topleveltools import new
from supriya.tools import timetools


class TimeManager(abctools.AbjadValueObject):

    ### PUBLIC METHODS ###

    @staticmethod
    def consolidate_demultiplexed_timespans(demultiplexed_timespans):
        for voice_name in demultiplexed_timespans:
            timespans = demultiplexed_timespans[voice_name]
            consolidated_timespans = TimeManager.consolidate_timespans(
                timespans)
            demultiplexed_timespans[voice_name] = consolidated_timespans

    @staticmethod
    def consolidate_timespans(timespans):
        consolidated_timespans = timespantools.TimespanInventory()
        for music_specifier, grouped_timespans in \
            TimeManager.group_timespans(timespans):
            if music_specifier is None:
                continue
            divisions = tuple(_.duration for _ in grouped_timespans)
            first_timespan = grouped_timespans[0]
            last_timespan = grouped_timespans[-1]
            consolidated_timespan = new(
                first_timespan,
                divisions=divisions,
                stop_offset=last_timespan.stop_offset,
                original_stop_offset=last_timespan.original_stop_offset,
                )
            consolidated_timespans.append(consolidated_timespan)
        consolidated_timespans.sort()
        return consolidated_timespans

    @staticmethod
    def demultiplex_timespans(multiplexed_timespans):
        demultiplexed_timespans = {}
        for timespan in multiplexed_timespans:
            voice_name, layer = timespan.voice_name, timespan.layer
            if voice_name not in demultiplexed_timespans:
                demultiplexed_timespans[voice_name] = {}
            if layer not in demultiplexed_timespans[voice_name]:
                demultiplexed_timespans[voice_name][layer] = \
                    timespantools.TimespanInventory()
            demultiplexed_timespans[voice_name][layer].append(
                timespan)
            demultiplexed_timespans[voice_name][layer]
        for voice_name in demultiplexed_timespans:
            timespan_inventories = demultiplexed_timespans[voice_name]
            timespan_inventory = \
                TimeManager.resolve_timespan_inventories(
                    timespan_inventories)
            demultiplexed_timespans[voice_name] = timespan_inventory
        return demultiplexed_timespans

    @staticmethod
    def execute(
        discard_final_silence=None,
        permitted_time_signatures=None,
        segment_session=None,
        target_duration=None,
        score_template=None,
        settings=None,
        ):

        score = score_template()
        multiplexed_timespans = timespantools.TimespanInventory()

        # populate independent timespans
        TimeManager._populate_multiplexed_timespans(
            dependent=False,
            score=score,
            score_template=score_template,
            settings=settings,
            target_duration=target_duration,
            timespan_inventory=multiplexed_timespans,
            )

        # find meters
        segment_session.meters = TimeManager.find_meters(
            discard_final_silence=discard_final_silence,
            permitted_time_signatures=permitted_time_signatures,
            target_duration=target_duration,
            timespan_inventory=multiplexed_timespans,
            )

        # demultiplex
        demultiplexed_timespans = TimeManager.demultiplex_timespans(
            multiplexed_timespans)

        # split performed timespans by meter offsets
        TimeManager.split_demultiplexed_timespans(
            segment_session.measure_offsets,
            demultiplexed_timespans,
            )

        # consolidate performed timespans by music specifier
        TimeManager.consolidate_demultiplexed_timespans(
            segment_session.measure_offsets,
            demultiplexed_timespans,
            )

        # inscribe grouped performed timespans with rhythms

            # rebuild performed timespans without silence divisions
            # (the inscription process should do the rebuilding)

        # make dependent timespans

        # make (pre-split) silent timespans

        # rewrite meters? (magic)

        # populate score

        # perform other rhythmic processing

        # collect attack points

    @staticmethod
    def find_meters(
        discard_final_silence=None,
        permitted_time_signatures=None,
        target_duration=None,
        timespan_inventory=None,
        ):
        offset_counter = datastructuretools.TypedCounter(
            item_class=durationtools.Offset,
            )
        for timespan in timespan_inventory:
            offset_counter[timespan.start_offset] += 2
            offset_counter[timespan.stop_offset] += 1
        offset_counter[target_duration] += 100
        if discard_final_silence is None:
            discard_final_silence = False
        else:
            discard_final_silence = bool(discard_final_silence)
        meters = metertools.Meter.fit_meters_to_expr(
            offset_counter,
            permitted_time_signatures,
            discard_final_orphan_downbeat=discard_final_silence,
            maximum_repetitions=2,
            )
        return tuple(meters)

    @staticmethod
    def get_rhythm_maker(music_specifier):
        if music_specifier is None:
            rhythm_maker = rhythmmakertools.NoteRhythmMaker(
                output_masks=[
                    rhythmmakertools.BooleanPattern(
                        indices=[0],
                        period=1,
                        )
                    ],
                )
        elif music_specifier.rhythm_maker is None:
            rhythm_maker = rhythmmakertools.NoteRhythmMaker(
                beam_specifier=rhythmmakertools.BeamSpecifier(
                    beam_each_division=False,
                    beam_divisions_together=False,
                    ),
                tie_specifier=rhythmmakertools.TieSpecifier(
                    tie_across_divisions=True,
                    ),
                )
        else:
            rhythm_maker = music_specifier.rhythm_maker
        return rhythm_maker

    @staticmethod
    def group_timespans(timespans):
        def grouper(timespan):
            music_specifier = None
            if isinstance(timespan, consort.PerformedTimespan):
                music_specifier = timespan.music_specifier
                if music_specifier is None:
                    music_specifier = consort.MusicSpecifier()
            return music_specifier
        import consort
        for music_specifier, grouped_timespans in itertools.groupby(
            timespans, grouper):
            grouped_timespans = timespantools.TimespanInventory(
                grouped_timespans)
            yield music_specifier, grouped_timespans

    @staticmethod
    def make_simple_music(rhythm_maker, durations, seed):
        music = rhythm_maker(durations, seeds=seed)
        for i, x in enumerate(music):
            if len(x) == 1 and isinstance(x[0], scoretools.Tuplet):
                music[i] = x[0]
            else:
                music[i] = scoretools.Container(x)
        music = scoretools.Container(music)
        for x in music[:]:
            if isinstance(x, scoretools.Tuplet) and x.multiplier == 1:
                mutate(x).swap(scoretools.Container())
        return music

    @staticmethod
    def division_is_silent(division):
        rest_prototype = (
            scoretools.Rest,
            scoretools.MultimeasureRest,
            )
        leaves = division.select_leaves()
        return all(isinstance(leaf, rest_prototype) for leaf in leaves)

    @staticmethod
    def populate_timespan(timespan, seed=None):
        populated_timespans = timespantools.TimespanInventory()
        rhythm_maker = TimeManager.get_rhythm_maker(timespan.music_specifier)
        durations = timespan.divisions[:]
        music = TimeManager.make_simple_music(
            rhythm_maker,
            durations,
            seed,
            )
        assert inspect_(music).get_duration() == timespan.duration
        groups = []
        group = []
        for division in music:
            if TimeManager.division_is_silent(division):
                if group:
                    groups.append(group)
                    group = []
            else:
                group.append(division)
        if group:
            groups.append(group)
        for group in reversed(groups):
            start_offset = inspect_(group[0]).get_timespan().start_offset
            stop_offset = inspect_(group[-1]).get_timespan().stop_offset
            start_offset += timespan.start_offset
            stop_offset += timespan.stop_offset
            container = scoretools.Container(group)
            beam = spannertools.GeneralizedBeam(
                durations=[division._get_duration() for division in music],
                include_long_duration_notes=True,
                include_long_duration_rests=False,
                isolated_nib_direction=None,
                use_stemlets=False,
                )
            attach(beam, container)
            populated_timespan = new(
                timespan,
                divisions=None,
                music=container,
                start_offset=start_offset,
                stop_offset=stop_offset,
                )
            populated_timespans.append(populated_timespan)
        return populated_timespans

    @staticmethod
    def populate_timespans(
        demultiplexed_timespans,
        score_template,
        ):
        counter = collections.Counter()
        voice_names = demultiplexed_timespans.keys()
        voice_names = TimeManager.sort_voice_names(
            score_template=score_template,
            voice_names=voice_names,
            )
        for voice_name in voice_names:
            populated_timespans = timespantools.TimespanInventory()
            for timespan in demultiplexed_timespans[voice_name]:
                music_specifier = timespan.music_specifier
                if music_specifier not in counter:
                    if music_specifier is None:
                        seed = 0
                    else:
                        seed = music_specifier.seed or 0
                    counter[music_specifier] = seed
                result = TimeManager.populate_timespan(timespan, seed=seed)
                populated_timespans.extend(result)
            demultiplexed_timespans[voice_name] = populated_timespans

    @staticmethod
    def populate_multiplexed_timespans(
        dependent=False,
        score=None,
        score_template=None,
        settings=None,
        target_duration=None,
        timespan_inventory=None,
        ):
        target_timespan = timespantools.Timespan(0, target_duration)
        if timespan_inventory is None:
            timespan_inventory = timespantools.TimespanInventory()
        independent_settings = [setting for setting in settings
            if not setting.timespan_maker.is_dependent
            ]
        dependent_settings = [setting for setting in settings
            if setting.timespan_maker.is_dependent
            ]
        if dependent:
            settings = dependent_settings
            start_index = len(independent_settings)
        else:
            settings = independent_settings
            start_index = 0
        for layer, music_setting in enumerate(settings, start_index):
            music_setting(
                layer=layer,
                score=score,
{'fraction_non_alphanumeric': 0.05651261529451954, 'fraction_numerical': 0.001756868869259675, 'mean_word_length': 2.1726273416937607, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3599324', 'n_tokens_mistral': 5734, 'n_tokens_neox': 5465, 'n_words': 994}