|
| 1 | +import logging |
| 2 | +import os |
| 3 | +from datetime import datetime, timezone |
| 4 | + |
| 5 | +from dotenv import load_dotenv |
| 6 | +from sift_py.grpc.transport import SiftChannelConfig, use_sift_channel |
| 7 | +from sift_py.ingestion.service import IngestionService |
| 8 | +from simulator import Simulator |
| 9 | +from telemetry_config import get_new_flow_config, nostromos_lv_426 |
| 10 | + |
| 11 | +if __name__ == "__main__": |
| 12 | + """ |
| 13 | + Example of telemetering data for the asset of name 'NostromoLV426' with various channels |
| 14 | + and rules. Also shows how to dynamically add new flows to your configuration at run time. |
| 15 | + The simulator will be sending data for various flows at various frequencies. |
| 16 | + """ |
| 17 | + logging.basicConfig(level=logging.DEBUG) |
| 18 | + logger = logging.getLogger(__name__) |
| 19 | + |
| 20 | + load_dotenv() |
| 21 | + |
| 22 | + apikey = os.getenv("SIFT_API_KEY") |
| 23 | + |
| 24 | + if apikey is None: |
| 25 | + raise Exception("Missing 'SIFT_API_KEY' environment variable.") |
| 26 | + |
| 27 | + base_uri = os.getenv("BASE_URI") |
| 28 | + |
| 29 | + if base_uri is None: |
| 30 | + raise Exception("Missing 'BASE_URI' environment variable.") |
| 31 | + |
| 32 | + # Load your telemetry config |
| 33 | + telemetry_config = nostromos_lv_426() |
| 34 | + |
| 35 | + # Create a gRPC transport channel configured specifically for the Sift API |
| 36 | + sift_channel_config = SiftChannelConfig(uri=base_uri, apikey=apikey) |
| 37 | + |
| 38 | + with use_sift_channel(sift_channel_config) as channel: |
| 39 | + # Create ingestion service using the telemetry config we loaded in |
| 40 | + ingestion_service = IngestionService( |
| 41 | + channel, |
| 42 | + telemetry_config, |
| 43 | + end_stream_on_error=True, # End stream if errors occur API-side. |
| 44 | + ) |
| 45 | + |
| 46 | + # Create an optional run as part of this ingestion |
| 47 | + current_ts = datetime.now(timezone.utc) |
| 48 | + run_name = f"[{telemetry_config.asset_name}].{current_ts.timestamp()}" |
| 49 | + ingestion_service.attach_run(channel, run_name, "Run simulation") |
| 50 | + |
| 51 | + # Dynamically add a new flow. This is done before starting the simulator |
| 52 | + # for this example, but can be done at any time after creating the IngestionService. |
| 53 | + new_flow = get_new_flow_config() |
| 54 | + if new_flow.name in ingestion_service.flow_configs_by_name: |
| 55 | + logger.info("New flow already exists, not adding.") |
| 56 | + else: |
| 57 | + logger.info("New flow does not exists, adding.") |
| 58 | + ingestion_service.try_create_flow(new_flow) |
| 59 | + |
| 60 | + # Create our simulator |
| 61 | + simulator = Simulator(ingestion_service) |
| 62 | + |
| 63 | + # Run it |
| 64 | + simulator.run() |
0 commit comments