from flowz.util import merge_keyed_channels

chan_div_2 = IterChannel(KeyedArtifact(i, i) for i in range(1, 13) if i % 2 == 0)
chan_div_3 = IterChannel(KeyedArtifact(i, i*10) for i in range(1, 13) if i % 3 == 0)
merged = merge_keyed_channels(chan_div_2, chan_div_3)
print_chans(merged)

"""
Explanation: Possible/Extant/All pattern
Another form of joining made possible by the util module is very powerful.  Here is an example reusing the chan_div_2 and chan_div_3 from the previous chapter:
End of explanation
"""


def expensive_deriver(num):
    # 10 minutes pass...
    return num * 100

chan = IterChannel(KeyedArtifact(i, DerivedArtifact(expensive_deriver, i)) for i in range(10))
print_chans(chan.tee())

"""
Explanation: What did that do?  If you think carefully about the values produced by the two channels, you will deduce that for each key, it selects the rightmost value with that key.  In other words, the values from both channels make it into the final channel, but if they both provide values for a particular key, the rightmost channel wins.  (Note, in this case, that 6 and 12 are not in the list.)
Why is that helpful?  That pattern works very well for creating the Possible/Extant/All pattern that is one main raisons d' etre for the flowz framework in the first place.  Here's how it works...
Suppose you have an expensive derivation function.  For giggles and grins, let's say it takes 10 minutes to compute.
End of explanation
"""


possible = chan

"""
Explanation: That would have taken 100 minutes.  If your process died 85 minutes in, you would be bummed to have to do it all again.  So, you would like to have a way to write out the results and, in the case of a crash, pick up where you left off.  That's where the pattern comes in.
The channel just defined represents all of the possible values, so let's call it that.
End of explanation
"""


storage = {num: expensive_deriver(num) for num in range(8)}
print(storage)

"""
Explanation: Suppose that we have already written out our data to durable storage.  We can represent that with an array of previously written values (as though we had run for 85 minutes):
End of explanation
"""


class ExampleExtantArtifact(ExtantArtifact):
    def __init__(self, num):
        super(ExampleExtantArtifact, self).__init__(self.get_me, name='ExampleExtantArtifact')
        self.num = num

    @gen.coroutine
    def get_me(self):
        raise gen.Return(storage[self.num])

"""
Explanation: Now, we need an ExtantArtifact that gets this data out of storage:
End of explanation
"""


keys = sorted(storage.keys())
print(keys)

extant = IterChannel(KeyedArtifact(i, ExampleExtantArtifact(i)) for i in sorted(storage.keys()))
print_chans(extant.tee())

"""
Explanation: Most durable storage mechanisms allow you to determine the keys of your stored items in sorted order, so lets do that and create an extant channel with that order:
End of explanation
"""


all_ = merge_keyed_channels(possible.tee(), extant.tee())
print_chans(all_.tee())

"""
Explanation: Great!  Now we can combine these two channels into our all channel, preferring the items in the extant channel:
End of explanation
"""


config_logging('INFO')
possible = IterChannel(KeyedArtifact(i, DerivedArtifact(expensive_deriver, i, name='expensive')) for i in range(10))
extant = IterChannel(KeyedArtifact(i, ExampleExtantArtifact(i)) for i in keys)
all_ = merge_keyed_channels(possible, extant)
print_chans(all_.tee())
config_logging('WARN')

"""
Explanation: OK.  Something happened there, but it's not clear exactly what, since it looks like the unadulterated output of the possible channel.  Let's turn logging back on, buid everything again (since all the artifacts have already been derived once, teeing won't be illustrative) and see what happens:
End of explanation
"""


config_logging('INFO')
possible = IterChannel(KeyedArtifact(i, DerivedArtifact(expensive_deriver, i, name='expensive')) for i in range(10))
extant = IterChannel(KeyedArtifact(i, ExampleExtantArtifact(i)) for i in sorted(storage.keys()))
all_ = merge_keyed_channels(possible, extant)
print_chans(all_.tee(), mode='ensure', func=lambda a: a)
config_logging('WARN')

"""
Explanation: Boom!  Notice how the expensive_deriver() calls ("DerivedArtifact&lt;expensive&gt; running deriver.") are only called twice at the end.  Our code did not have to consciously figure out how much had already been done and carefully make sure that we only call the deriver for the remaining ones.  The lazy evaluation did it all.
There is yet one more performance improvement to make here, though.  If we have already written out 8 of our expensively derived data sets, not only do we no longer need to derive and write them out, but we don't even need to read them in!  flowz and the ExtantArtifact class allows to optimize things by ensuring each of the items in the channel, rather than getting them.
End of explanation
"""


# A function to write the data, to be passed to a transform() call
def data_writing_transform(key, value):
    storage[key] = value
    return value

# recreate the storage and turn on logging
storage = {num: expensive_deriver(num) for num in range(8)}
config_logging('INFO')

# Run as though we failed after 85 minutes and are picking up again
possible = IterChannel(KeyedArtifact(i, DerivedArtifact(expensive_deriver, i, name='expensive')).transform(data_writing_transform, i) for i in range(10))
extant = IterChannel(KeyedArtifact(i, ExampleExtantArtifact(i)) for i in sorted(storage.keys()))
all_ = merge_keyed_channels(possible, extant)
print_chans(all_.tee(), mode='ensure', func=lambda a: a)

"""
Explanation: ensure on an ExtantArtifact is essentially a no-op just returning True, but it calls the get method on a DerivedArtifact.  So we have done the minimal amount needed to get up to date:
1. A fast operation (getting the keys) to figure out what has already been written
2. The expensive operations for the items remaining to be written
All that remains now is that we haven't written this new data.  Let's try that now.
End of explanation
"""


print(storage)

"""
Explanation: That's odd.  It shows four deviver calls.  Notice that only two of them, however, have "&lt;expensive&gt;" in the log.  It turns out the transform() uses a DerivedArtifact under the covers, too.
End of explanation
"""


possible = IterChannel(KeyedArtifact(i, DerivedArtifact(expensive_deriver, i, name='expensive')).transform(data_writing_transform, i) for i in range(10))
extant = IterChannel(KeyedArtifact(i, ExampleExtantArtifact(i)) for i in sorted(storage.keys()))
all = merge_keyed_channels(possible, extant)
print_chans(all.tee(), mode='ensure', func=lambda a: a)

"""
Explanation: Yes!  Our storage has been updated.  Now, if we run yet again, nothing should be done.
End of explanation
"""


# recreate the storage to not mess up other parts of the notebook when run out of order, and turn off logging
storage = {num: expensive_deriver(num) for num in range(8)}
config_logging('WARN')

"""
Explanation: QED
End of explanation
"""


