def main():
   local = sys.argv[1] == "local"


   #define needed variables
   COMMANDS_TOPIC = "streamsx/iot/device/commands/send" #topic to publish commands to
   EVENTS_TOPIC = "streamsx/iot/device/events" #topic to subscribe to for events
   incoming_schema =  schema.StreamSchema("tuple <rstring typeId, rstring deviceId, rstring eventId,rstring jsonString>")
   cmd_schema = schema.StreamSchema('tuple<rstring typeId, rstring deviceId, rstring cmdId, rstring jsonString>')


   topo = Topology('ReadingsFromIot')

   #Subscribe to  events
   events = topo.subscribe(EVENTS_TOPIC, incoming_schema,"AllEventsAsJSON")
   sensor_events = events.filter(lambda tuple: tuple["eventId"] == "sensors","SensorEventsAsJSON")
   readings = sensor_events.map(get_event_data,"ReadingsStream")
   readings.print()

   #send a command
   cmd_stream = sensor_events.map(get_cmd, "CommandsAsJSON")
   #convert the commands stream to a SPL structured schema
   commands_to_publish = cmd_stream.map(lambda x : (x["typeId"],x["deviceId"],x["cmdId"],x["jsonString"],), schema = cmd_schema, name="CommandsToPublish")

   commands_to_publish.publish(COMMANDS_TOPIC, cmd_schema)

   if local and len(sys.argv) > 2:
      username = sys.argv[2]
      password = sys.argv[3]
      result = submit_to_service(topo, local, username, password)
   else:
   	  result = submit_to_service(topo, local)

   print("Submitted job to the service, job id = " + str(result.job.id))

    def test_fn(self):
        topo = Topology()

        s = fn_ecruos(topo)
        self._csl_stream(s, 'source', 'fn_ecruos')

        s = fn_retlif(s)
        self._csl_stream(s, 'filter', 'fn_retlif')

        s = fn_pam(s)
        self._csl_stream(s, 'map', 'fn_pam')

        s = fn_pam_talf(s)
        self._csl_stream(s, 'flat_map', 'fn_pam_talf')
        
        s = fn_gnirts_sa(s)
        self._csl_stream(s, 'as_string', 'fn_gnirts_sa')

        s = fn_nosj_sa(s)
        self._csl_stream(s, 'as_json', 'fn_nosj_sa')

        st = fn_ebircsbus(topo)
        self._csl_stream(st, 'subscribe', 'fn_ebircsbus')

        e = fn_hcae_rof(s)
        self._csl_sink(e, 'for_each', 'fn_hcae_rof')

        e = fn_hsilbup(s)
        self._csl_sink(e, 'publish', 'fn_hsilbup')

        e = fn_hsilbup(topo.source([]), schema=CommonSchema.Json)
        self._csl_sink(e, 'publish', 'fn_hsilbup')

        e = fn_tnirp(s)
        self._csl_sink(e, 'print', 'fn_tnirp')

def main():
    """
    Sample transform application.  This Python application builds a topology that
    * transforms a stream of string tuples from a source operator to a stream of integer tuples 
    * uses `transform` to perform addition on the integer tuples
    * prints the stream to stdout
    * submits the topology in standalone mode (compiles and executes it as a standalone application)
    
    Example:
        > python3 transform_sample.py
    Output:
        342
        474
        9342
    """
    
    # create the container for the topology that will hold the streams
    topo = Topology("transform_sample")
    
    # declare a source stream (`source`) that contains string tuples
    source = topo.source(transform_sample_functions.int_strings_transform)
    
    # transform the stream of string tuples (`source`) to a stream of integer tuples (`i1`)
    i1 = source.transform(transform_sample_functions.string_to_int)
    
    # adds 17 to each integer tuple 
    i2 = i1.transform(transform_sample_functions.AddNum(17))
    
    # terminate the stream by printing each tuple to stdout
    i2.print()
    
    # execute the application in standalone mode
    streamsx.topology.context.submit("STANDALONE", topo.graph)

def main():
    t = Topology("FFT_Sample")
    readings = t.source(signal_generator.Readings(50)).transform(TumblingWindow(10))
    fftStream = readings.transform(fftpack.fft)
    fftStream.sink(print)

    streamsx.topology.context.submit("STANDALONE", t.graph)

def main():
    """
    Sample filtering echo topology application. This Python application builds a
    simple topology that echos its command line arguments to standard output.

    This demonstrates use of Python functional logic to filter the tuples.
    A user-defined function implements the filtering logic, in this
    case only echo tuples that start with the letter `d`.

    Args:
        a list of values
        
    Example:
        python3 filter_echo.py cat dog mouse door
    Output:
        dog
        door
    """
    
    topo = Topology("filter_echo")
    source = topo.source(filter_echo_functions.SysArgv(sys.argv[1:]))
    
    # Declare a stream that will execute functional logic
    # against tuples on the echo stream.
    # For each tuple that will appear on echo, the below
    # `starts_with_d` method will be called.  If it returns
    # True then the tuple will appear on the filtered
    # stream, otherwise the tuple is discarded.
    filtered = source.filter(filter_echo_functions.starts_with_d)
    
    filtered.print()
    
    streamsx.topology.context.submit("STANDALONE", topo.graph)

def main():
    """
    Sample echo topology application. This Python application builds a
    simple topology that echoes its command line arguments to standard output.

    The application implements the typical pattern
    of code that declares a topology followed by
    submission of the topology to a Streams context.
    
    Args:
        a list of values to print to stdout
        
    Example:
        python3 echo.py hello1 hello2 hello3
    Output:
        hello1
        hello2
        hello3
    """
    
    topo = Topology("echo")
    # The command line arguments (sys.argv) are captured by the SysArgv
    # callable class and will be used at runtime as the contents of the
    # echo stream.
    echo = topo.source(echo_functions.SysArgv(sys.argv[1:]))
    
    # print the echo stream to stdout
    echo.print()
    
    # At this point the topology is declared with a single
    # stream that is printed to stdout
    
    # execute the topology by submitting to a standalone context
    streamsx.topology.context.submit("STANDALONE", topo.graph)

    def test_fetch_logs_on_failure(self):
        topo = Topology("fetch_logs_on_failure")
        s = topo.source(["foo"])

        tester = Tester(topo)
        # Causes test to fail
        tester.contents(s, ["bar"])

        try:
            self.tester = tester
            tester.local_check = self._can_retrieve_logs
            tester.test(self.test_ctxtype, self.test_config)
        except AssertionError:
            # This test is expected to fail, do nothing.
            pass

        # Check if logs were downloaded
        if self.can_retrieve_logs:
            logs = tester.result['application_logs']
            exists = os.path.isfile(logs)
            
            self.assertTrue(exists, "Application logs were not downloaded on test failure")
            
            if exists:
                os.remove(logs)

    def _test_submit_sab(self):
        topo = Topology('SabTest', namespace='mynamespace')
        s = topo.source([1,2])
        es = s.for_each(lambda x : None)
        bb = streamsx.topology.context.submit('BUNDLE', topo, {})
        self.assertIn('bundlePath', bb)
        self.assertIn('jobConfigPath', bb)

        sas = self.sc.get_streaming_analytics()

        sr = sas.submit_job(bundle=bb['bundlePath'])
        job_id = sr.get('id', sr.get('jobId'))
        self.assertIsNotNone(job_id)
        self.assertIn('name', sr)
        self.assertIn('application', sr)
        self.assertEqual('mynamespace::SabTest', sr['application'])
        cr = sas.cancel_job(job_id=job_id)

        jn = 'SABTEST:' + str(time.time())
        jc = streamsx.topology.context.JobConfig(job_name=jn)
        sr = sas.submit_job(bundle=bb['bundlePath'], job_config=jc)
        job_id = sr.get('id', sr.get('jobId'))
        self.assertIsNotNone(job_id)
        self.assertIn('application', sr)
        self.assertEqual('mynamespace::SabTest', sr['application'])
        self.assertIn('name', sr)
        self.assertEqual(jn, sr['name'])
        cr = sas.cancel_job(job_id=job_id)
       
        os.remove(bb['bundlePath'])
        os.remove(bb['jobConfigPath'])

def main():
    """
    Sample Hello World topology application. This Python application builds a
    simple topology that prints Hello World to standard output.

    The application implements the typical pattern
    of code that declares a topology followed by
    submission of the topology to a Streams context.
    
    This demonstrates the mechanics of declaring a topology and executing it.
            
    Example:
        python3 hello_world.py
    Output:
        Hello
        World!
    """
    
    # Create the container for the topology that will hold the streams of tuples.
    topo = Topology("hello_world")
    
    # Declare a source stream (hw) with string tuples containing two tuples,
    # "Hello" and "World!".
    hw = topo.source(hello_world_functions.source_tuples)
    
    # Sink hw by printing each of its tuples to standard output
    hw.print()
    
    # At this point the topology is declared with a single
    # stream that is printed to standard output
    
    # Now execute the topology by submitting to a standalone context.
    streamsx.topology.context.submit("STANDALONE", topo.graph)

def main():
    """
    Finds outliers from a sequence of floats (e.g. simulating a sensor reading).
    Demonstrates function logic that maintains state across tuples.
    
    Example:
        python3 find_outliers.py
    Example Output:
        2.753064082105016
        -2.210758753960355
        1.9847958795117937
        2.661689193901883
        2.468061723082693
        ...
    """
    topo = Topology("find_outliers")
    
    # Produce a stream of random float values with a normal
    # distribution, mean 0.0 and standard deviation 1.
    values = topo.source(find_outliers_functions.readings)
    

    # Filters the values based on calculating the mean and standard
    # deviation from the incoming data. In this case only outliers are
    # present in the output stream outliers. An outlier is defined as 
    # more than (threshold * standard deviation) from the mean.  The
    # threshold in this example is 2.0.
    # This demonstrates a functional logic class that is
    # stateful. The threshold, sum_x, and sum_x_squared maintain 
    # their values across multiple invocations.
    outliers = values.filter(find_outliers_functions.IsOutlier(2.0))
    
    outliers.print()
    
    streamsx.topology.context.submit("STANDALONE", topo.graph)

    def test_get_job(self):
        topo = Topology("job_in_result_test")
        topo.source(["foo"])

        tester = Tester(topo)
        self.tester = tester

        tester.local_check = self._correct_job_ids
        tester.test(self.test_ctxtype, self.test_config)

def main():
    ref_signal = signal.hann(10)

    t = Topology("Convolve_Sample")
    readings = t.source(signal_generator.Readings(100)).transform(TumblingWindow(20))
    convolveStream = readings.transform(signal_functions.Convolve(ref_signal))
    convolveStream.sink(print)

    streamsx.topology.context.submit("STANDALONE", t.graph)

def main():
    filter_order = 4
    cutoffFreq = 100
    sampleRate = 1000

    t = Topology("LowpassFilter_Sample")
    readings = t.source(signal_generator.Readings(50000)).transform(TumblingWindow(2000))
    filterStream = readings.transform(butterworth.Lowpass(filter_order, cutoffFreq, sampleRate))
    filterStream.sink(print)

    streamsx.topology.context.submit("STANDALONE", t.graph)

def main():
    """
    Sample continuous (streaming) grep topology application. This Python application builds a
    simple topology that periodically polls a directory for files, reads each file and
    output lines that contain the search term.
    Thus as each file is added to the directory, the application will read
    it and output matching lines.
    
    Args:
        directory (string): a directory that contains files to process
        search_string (string): a search term
        
    Example:
        * Create a subdirectory "dir"
        * Create file1.txt in subdirectory "dir" with the following contents:
            file1 line1
            file1 line2
            file1 line3
        * Create file2.txt in subdirectory "dir" with the following contents:
            file2 line1
            file2 line2
            file2 line3
        * python3 grep.py dir line2
        
    Output:
        file1 line2
        file2 line2
    """
    
    if len(sys.argv) != 3:
        print("Usage: python3 grep.py <directory> <search_string>")
        return
    directory = sys.argv[1]
    term = sys.argv[2]
    topo = Topology("grep")
    
    # Declare a stream that will contain the contents of the files.
    # For each input file, DirectoryWatcher opens the file and reads its contents 
    # as a text file, producing a tuple for each line of the file. The tuple contains
    # the contents of the line, as a string.
    lines = topo.source(util_functions.DirectoryWatcher(directory))
    
    # Filter out non-matching lines. FilterLine is a callable class 
    # that will be executed for each tuple on lines, that is each line
    # read from a file.  Only lines that contain the string `term` will
    # be included in the output stream.
    matching = lines.filter(grep_functions.FilterLine(term))
    
    # print the matching lines to standard out
    matching.print()
    
    # execute the topology
    streamsx.topology.context.submit("STANDALONE", topo)

    def test_get_job(self):
        topo = Topology("job_in_result_test")
        topo.source(["foo"])

        sc = rest.StreamsConnection(username=self.username, password=self.password)
        sc.session.verify = False
        config = {ConfigParams.STREAMS_CONNECTION : sc}

        tester = Tester(topo)
        self.tester = tester

        tester.local_check = self._correct_job_ids
        tester.test(self.test_ctxtype, config)

    def test_always_fetch_logs(self):
        topo = Topology("always_fetch_logs")
        s = topo.source(["foo"])

        tester = Tester(topo)
        tester.contents(s, ["foo"])

        tester.test(self.test_ctxtype, self.test_config, always_collect_logs=True)

        # Check if logs were downloaded
        logs = tester.result['application_logs']
        exists = os.path.isfile(logs)

        self.assertTrue(exists, "Application logs were not downloaded on test success")

        if exists:
            os.remove(logs)

def main():
    """
    The 'Estimator' model accepts a tuple with these elements: (type, X, y), where:
       'type':  't' (for training), 'd' (for data), '' (empty string, same as 'd')
       'X':     is the data
       'y':     is the actual class of the data (only used to train the model)
    """
    training_size = 100
    num_centers = 2
    num_features = 2

    t = Topology("Estimator_Sample")
    trainingStream = t.source(sklearn_sources.Blobs(iterations=training_size, isTraining=True, centers=num_centers, n_features=num_features))
    dataStream = t.source(sklearn_sources.Blobs(centers=num_centers, n_features=num_features))
    combinedStreams = trainingStream.union({dataStream})
    predictionStream = combinedStreams.transform(Estimator(training_size, KNeighborsClassifier()))
    predictionStream.sink(print)

    streamsx.topology.context.submit("STANDALONE", t.graph)

    def run(self, context="DISTRIBUTED"):
        ## Create topology
        topo = Topology("HealthcareDemo")

        ## Ingest, preprocess and aggregate patient data
        patientData = topo.subscribe("ingest-physionet", schema.CommonSchema.Json) \
                          .map(functions.identity) \
                          .filter(healthcare_functions.PatientFilter(self.patient_id)) \
                          .transform(healthcare_functions.GenTimestamp(self.sample_rate)) \
                          .transform(SlidingWindow(length=self.sample_rate, trigger=self.sample_rate-1)) \
                          .transform(healthcare_functions.aggregate) \

        ## Calculate RPeak and RR delta
        rpeak_data_stream = patientmonitoring_functions.streaming_rpeak(patientData, self.sample_rate, data_label='ECG Lead II')

        ## Create a view of the data
        self.view_data = rpeak_data_stream.view()

        ## Compile Python Streams application and submit job
        streamsx.topology.context.submit(context, topo.graph, username=self.username, password=self.password)

    def test_always_fetch_logs(self):
        topo = Topology("always_fetch_logs")
        s = topo.source(["foo"])

        tester = Tester(topo)
        tester.contents(s, ["foo"])

        self.tester = tester
        tester.local_check = self._can_retrieve_logs
        tester.test(self.test_ctxtype, self.test_config, always_collect_logs=True)

        if self.can_retrieve_logs:
            # streams version is >= 4.2.4. Fetching logs is supported.
            # Check if logs were downloaded
            logs = tester.result['application_logs']
            exists = os.path.isfile(logs)

            self.assertTrue(exists, "Application logs were not downloaded on test success")
            
            if exists:
                os.remove(logs)                            

def main():
    """
    Plays Fizz Buzz (https://en.wikipedia.org/wiki/Fizz_buzz)
    
    Example:
        python3 fizz_buzz.py
    Output:
        1
        2
        Fizz!
        4
        Buzz!
        Fizz!
        7
        8
        Fizz!
        Buzz!
        11
        Fizz!
        13
        14
        FizzBuzz!
        ...

    """
    topo = Topology("fizz_buzz")
    
    # Declare a stream of int values
    counting = topo.source(fizz_buzz_functions.int_tuples)
    
    # Print the tuples to standard output
    play_fizz_buzz(counting).print()
    
    # At this point the streaming topology (streaming) is
    # declared, but no data is flowing. The topology
    # must be submitted to a context to be executed.
    
    # execute the topology by submitting to a standalone context
    streamsx.topology.context.submit("STANDALONE", topo.graph)