### Getting Data Summaries from Streams Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Retrieves statistical summaries (e.g., count, min, max, mean) for data within a time range in an SDS stream. Allows specifying the number of intervals for aggregation. Requires namespace, stream ID, start time, end time, and number of intervals. ```python namespace_id = "your-namespace" first_time = "2017-01-11T22:21:23.430Z" last_time = "2017-01-11T22:28:29.430Z" # Get summary with 1 interval (entire range) summary_results = sds_client.Streams.getSummaries( namespace_id, "Tank1", None, # No specific summary types (get all) first_time, last_time, 1 # Number of intervals ) print(summary_results) # Output includes: Count, Minimum, Maximum, Mean, StandardDeviation, etc. # Get multiple interval summaries summary_results = sds_client.Streams.getSummaries( namespace_id, "Tank1", None, first_time, last_time, 3 # Split into 3 intervals ) print(f"Generated {len(summary_results)} interval summaries") ``` -------------------------------- ### Query Stream Data with Window Values Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Retrieves data from an SDS stream within a specified time range. Requires namespace, stream ID, start time, and end time. Can query simple or complex streams. ```python namespace_id = "your-namespace" # Define time window first_time = "2017-01-11T22:21:23.430Z" last_time = "2017-01-11T22:28:29.430Z" # Query pressure stream results = sds_client.Streams.getWindowValues( namespace_id, "Pressure_Tank1", first_time, last_time ) print(f"Retrieved {len(results)} pressure readings") print(f"First value: {results[0]}") # Output: {'value': 346, 'time': '2017-01-11T22:21:23.430Z'} # Query complex tank stream tank_results = sds_client.Streams.getWindowValues( namespace_id, "Tank1", first_time, last_time ) print(f"Tank reading: {tank_results[1]}") # Output: {'pressure': 0, 'temperature': 0, 'time': '2017-01-11T22:22:23.430Z'} ``` -------------------------------- ### Get Community Member Role and Share Stream with Community Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Retrieves the community member role and shares a stream with the community using JSON Patch. This involves fetching roles, identifying the community member role, constructing a JSON Patch operation to add access control entries, and applying the patch to the stream. ```python # Get community member role roles = sds_client.Roles.getRoles() community_role = None for r in roles: if (r.RoleTypeId == sds_client.Roles.CommunityMemberRoleTypeId and r.CommunityId == community_id): community_role = r break print(f"Community member role ID: {community_role.Id}") # Share stream with community using JSON Patch patch = jsonpatch.JsonPatch([ { 'op': 'add', 'path': '/RoleTrusteeAccessControlEntries/-', 'value': { 'AccessRights': 1, # Read access 'AccessType': 0, 'Trustee': { 'ObjectId': community_role.Id, 'TenantId': None, 'Type': 'Role' } } } ]) sds_client.Streams.patchAccessControl( namespace_id, "Pressure_Tank1", patch ) print("Stream shared with community") ``` -------------------------------- ### Configure SDS Client in Python Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Demonstrates how to configure a Python client for AVEVA Data Hub or Edge Data Store using a configuration file. The code loads configuration settings to establish a connection. ```Python import json from adh_sample_library_preview import EDSClient, ADHClient # Load configuration with open('appsettings.json', 'r') as f: appsettings = json.load(f) # Edge Data Store client (local) if appsettings['TenantId'] == 'default' and appsettings['NamespaceId'] == 'default': sds_client = EDSClient( appsettings['ApiVersion'], appsettings['Resource'] ) else: # AVEVA Data Hub client (cloud) sds_client = ADHClient( appsettings['ApiVersion'], appsettings['TenantId'], appsettings['Resource'], appsettings['ClientId'], appsettings['ClientSecret'], False ) ``` -------------------------------- ### Bulk Stream Operations: Create and Query Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Demonstrates creating multiple SDS streams, inserting data into them, and then performing a bulk query across specified streams within a time window. Requires namespace and a list of stream IDs for bulk operations. ```python import time namespace_id = "your-namespace" # Create multiple tank streams tank_streams = ["Vessel", "Tank1", "Tank2"] tank_type_id = "Pressure_Temp_Time" for stream_id in tank_streams: stream = SdsStream(stream_id, tank_type_id, description=f"Stream for {stream_id}") sds_client.Streams.createOrUpdateStream(namespace_id, stream) # Insert data into each stream tank1_data = [ {"pressure": 346, "temperature": 91, "time": "2017-01-11T22:21:23.430Z"}, {"pressure": 385, "temperature": 92, "time": "2017-01-11T22:25:23.430Z"} ] tank2_data = [ {"pressure": 345, "temperature": 89, "time": "2017-01-11T22:20:23.430Z"}, {"pressure": 374, "temperature": 87, "time": "2017-01-11T22:28:23.430Z"} ] sds_client.Streams.insertValues(namespace_id, "Tank1", json.dumps(tank1_data)) sds_client.Streams.insertValues(namespace_id, "Tank2", json.dumps(tank2_data)) time.sleep(5) # Allow time for indexing # Bulk query multiple streams first_time = "2017-01-11T22:20:23.430Z" last_time = "2017-01-11T22:28:29.430Z" results = sds_client.Streams.getStreamsWindow( namespace_id, ["Vessel", "Tank2"], # Query specific streams None, # No filter first_time, last_time ) print(f"Bulk results: {results}") # Returns dictionary with stream IDs as keys and their data as values ``` -------------------------------- ### Create and Manage SDS Streams in Python Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Shows how to create and manage data streams based on the defined SDS types. It demonstrates the creation of streams for pressure, temperature, and combined data. ```Python from adh_sample_library_preview import SdsStream namespace_id = "your-namespace" # Create simple value stream pressure_stream = SdsStream( "Pressure_Tank1", "Value_Time", description="A stream for pressure data of tank1" ) sds_client.Streams.createOrUpdateStream(namespace_id, pressure_stream) temperature_stream = SdsStream( "Temperature_Tank1", "Value_Time", description="A stream for temperature data of tank1" ) sds_client.Streams.createOrUpdateStream(namespace_id, temperature_stream) # Create complex multi-value stream tank_stream = SdsStream( "Tank1", "Pressure_Temp_Time", description="A stream for data of tank1s" ) sds_client.Streams.createOrUpdateStream(namespace_id, tank_stream) print(f"Created streams: {pressure_stream.Id}, {temperature_stream.Id}, {tank_stream.Id}") ``` -------------------------------- ### Insert Time-Series Data into SDS Streams in Python Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Demonstrates inserting time-series data into the created streams. It provides sample data and shows the insertion process. ```Python import json namespace_id = "your-namespace" # Prepare pressure data pressure_data = [ {"value": 346, "time": "2017-01-11T22:21:23.430Z"}, {"value": 0, "time": "2017-01-11T22:22:23.430Z"}, {"value": 386, "time": "2017-01-11T22:24:23.430Z"}, {"value": 385, "time": "2017-01-11T22:25:23.430Z"}, {"value": 384.2, "time": "2017-01-11T22:26:23.430Z"} ] ``` -------------------------------- ### Create SDS Types in Python Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Illustrates the creation of SDS type definitions using the `adh-sample-library-preview` library. It defines and registers time-series data schemas, including single and multiple value types. ```Python from adh_sample_library_preview import SdsType, SdsTypeCode, SdsTypeProperty # Simple value-time type def get_type_value_time(): double_type = SdsType("doubleType", SdsTypeCode.Double) time_type = SdsType("string", SdsTypeCode.DateTime) value = SdsTypeProperty("value", False, double_type) time_prop = SdsTypeProperty("time", True, time_type) # True = indexed property type_value_time = SdsType( "Value_Time", SdsTypeCode.Object, [value, time_prop], description="A Time-Series indexed type with a value" ) return type_value_time # Complex multi-value type def get_type_press_temp_time(): double_type = SdsType("doubleType", SdsTypeCode.Double) time_type = SdsType("string", SdsTypeCode.DateTime) temperature = SdsTypeProperty("temperature", False, double_type) pressure = SdsTypeProperty("pressure", False, double_type) time_prop = SdsTypeProperty("time", True, time_type) type_press_temp_time = SdsType( "Pressure_Temp_Time", SdsTypeCode.Object, [temperature, pressure, time_prop], description="A Time-Series indexed type with 2 values" ) return type_press_temp_time # Register type with SDS namespace_id = "your-namespace" time_value_type = get_type_value_time() time_value_type = sds_client.Types.getOrCreateType(namespace_id, time_value_type) print(f"Created type: {time_value_type.Id}") ``` -------------------------------- ### Using Verbosity for Null Value Handling in Queries Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Controls how null or default values are included in query results. Setting `sds_client.acceptverbosity` to `True` includes nulls, while `False` omits them. This affects the structure of returned data objects. ```python namespace_id = "your-namespace" first_time = "2017-01-11T22:21:23.430Z" last_time = "2017-01-11T22:28:29.430Z" # Default behavior: null values omitted sds_client.acceptverbosity = False results = sds_client.Streams.getWindowValues( namespace_id, "Pressure_Tank1", first_time, last_time ) print(f"Without verbosity: {results[1]}") # Output: {'time': '2017-01-11T22:22:23.430Z'} (value=0 omitted) # With verbosity: null values included as defaults sds_client.acceptverbosity = True results = sds_client.Streams.getWindowValues( namespace_id, "Pressure_Tank1", first_time, last_time ) print(f"With verbosity: {results[1]}") # Output: {'value': 0.0, 'time': '2017-01-11T22:22:23.430Z'} (value=0.0 included) # Reset to default sds_client.acceptverbosity = False ``` -------------------------------- ### Search Community for Streams and Read Data Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Searches a community for streams and reads the last value from a community stream. This involves retrieving streams associated with a community and then fetching the most recent data point from a specific shared stream. ```python # Search community for streams community_streams = sds_client.Communities.getCommunityStreams( community_id, "Pressure_Tank1" ) print(f"Found {len(community_streams)} matching streams") # Read data from community stream community_stream = community_streams[0] community_data = sds_client.SharedStreams.getLastValue( community_stream.TenantId, community_stream.NamespaceId, community_id, community_stream.Id ) print(f"Retrieved last value: {community_data['value']}") ``` -------------------------------- ### Insert Data into SDS Streams Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Inserts single or complex JSON data into specified SDS streams. Requires namespace, stream ID, and JSON-formatted data. Handles both simple and complex data structures. ```python # Insert into simple stream sds_client.Streams.insertValues( namespace_id, "Pressure_Tank1", json.dumps(pressure_data) ) # Prepare complex tank data tank_data = [ {"pressure": 346, "temperature": 91, "time": "2017-01-11T22:21:23.430Z"}, {"pressure": 0, "temperature": 0, "time": "2017-01-11T22:22:23.430Z"}, {"pressure": 386, "temperature": 93, "time": "2017-01-11T22:24:23.430Z"}, {"pressure": 385, "temperature": 92, "time": "2017-01-11T22:25:23.430Z"}, {"pressure": 384.2, "temperature": 92, "time": "2017-01-11T22:26:23.430Z"} ] # Insert into complex stream sds_client.Streams.insertValues( namespace_id, "Tank1", json.dumps(tank_data) ) print(f"Inserted {len(tank_data)} values") ``` -------------------------------- ### Cleanup: Delete Streams and Types Source: https://context7.com/aveva/sample-adh-time_series-python/llms.txt Performs cleanup by deleting specified streams and types from the SDS. This function ensures that resources are properly released after use, handling potential errors during the deletion process. ```python namespace_id = "your-namespace" def suppress_error(sds_call): """Suppress errors during cleanup""" try: sds_call() except Exception as error: print(f"Cleanup error: {error}") # Delete all streams stream_ids = ["Pressure_Tank1", "Temperature_Tank1", "Vessel", "Tank1", "Tank2"] for stream_id in stream_ids: suppress_error(lambda sid=stream_id: sds_client.Streams.deleteStream( namespace_id, sid )) print(f"Deleted stream: {stream_id}") # Delete types (must delete streams first due to dependencies) type_ids = ["Pressure_Temp_Time", "Value_Time"] for type_id in type_ids: suppress_error(lambda tid=type_id: sds_client.Types.deleteType( namespace_id, tid )) print(f"Deleted type: {type_id}") print("Cleanup complete") ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.