Python API - Q and A

[sbeadle]

📖 Home  

📖 Python article index  

How to do things in the Geo SCADA Python API

This page is a list of simple samples. It will expand over time. Stuck? Ask a question on the Geo SCADA forum.

1. How to read the value of Server Status metrics?

A. These don't have a database object. You can use validate_tags to get an array of tag details:

    x = connection.validate_tags(["#ICMPPOL.Main Link SystemD2-0.Name / IP address"])
    print(x[0].initial_value.value)

2. How to run an SQL query?

A. A synchronous query is used to get data immediately:

    q = connection.prepare_query("SELECT TOP(10) FULLNAME,Id,CurrentValueFormatted,CurrentTime FROM CDBPOINT", False)
    result = q.execute_sync()
    if result.status == QueryStatus.Succeeded:
        print( "Sync Records:", len(result.rows))
        for x in result.rows:
            print( f"{x.data[0].value},{x.data[1].value},{x.data[2].value}")
    else:
        print( "Error:", result.status)

The array index [0], [1] ... gets the data column values which have a .value property.

3. How to get and set properties of an object?

A. The methods connection.get_property, .get_properties, .set_property all take an object Id as the first argument, then property name(s). But the actual data returned or sent are not basic Python items. This is because the data types need to be encoded and decoded correctly. They are of type Variant. A variant is a class with a type and the data. Here are some examples:

# Get an object from its full name
G = connection.find_object( fullname)
# Read a property
type_desc = connection.get_property(G.id, "TypeDesc")
print( f"Object name: {fullname}, Id: {G.id}, Type: {type_desc.value}.")
# Read multiple properties, also from an aggregate
result = connection.get_properties(temperature_point.id, ["GISLocationSource.Latitude", "GISLocationSource.Longitude"])
latitude, longitude = result[0].value, result[1].value
print( f"Location: {latitude}, {longitude}")

# Create a variant to set a property - the following sets multiple properties
connection.set_properties(point.id, [
            ("PresetQuality", Variant(VariantType.I4, 192)),
            ("PresetTimestamp", Variant(VariantType.Date, datetime.utcnow())),
            ("CurrentValue", Variant(VariantType.R8, float(new_value)))       ])

# Typical variant types are "Date" for a date and time, "I4" for an integer,
# "Bool" for a boolean, "R8" for a floating point number and "BStr" for a string.

# Some properties or method arguments or results are a variant which is an array
# Example of a variant which is an array of double precision float values:
# Create a variant array
vresult = Variant(CombinedVariantType(VariantType.R8, VariantFlags.Array), value_array)
# List members from a variant array
print([x.value for x in vresult.value]) 

# Executing Methods on objects requires the use of an array of variants
# For example, to read a point's historic values:
arguments = [Variant(VariantType.Date, datetime.utcnow() - timedelta(minutes=1)),
             Variant(VariantType.Date, datetime.utcnow()),
             Variant(VariantType.I4, 0),
             Variant(VariantType.I4, 1000),
             Variant(VariantType.Bool, True),
             Variant(VariantType.BStr, "")]
vresult = connection.invoke_method(point.id, "Historic.RawValues", arguments)

4. How to create a Geo SCADA object

A. The connection.create_object() method takes in the class name of the object, it's parent group ID and the name of the new object. The class name to be used can be found with the database schema browser. You'll need to use the connection.find_object() method to get the ID of the parent group where you want to put the new object.

Example code:

    G = connection.find_object('My parent group')
    if (G == None):
        print('Error')
        quit
    try:
        P = connection.create_object( "CPointAlgManual", G.id, "Temperature")
    except Exception as e:
        print("Can't create point " + str(e) )
        quit

5. How to get the current value, time and quality of a point

A. Call the connection.get_properties() method.

Example code:

    P = connection.find_object('WeatherPoints.Temperature')
    result = connection.get_properties( P.id, ["CurrentValue", "CurrentTime", "CurrentQuality"] )
    print( f"Values: {result[0].value}, {result[1].value}, {result[2].value}")

6. How to read raw historic data values and their times

A. The connection.read_raw_history() method lets you read both together.

Example code:

    P = connection.find_object('WeatherPoints.Temperature')
    tags = [HistoricTag( P.full_name + ".Historic",ExtendedSourceFilter.Raw,0,0) ]
    # Requires this: from datetime import datetime, timedelta
    result = connection.read_raw_history( datetime.now() - timedelta(days=1),
                             datetime.now(),
                             100, True, tags)
    # Print the first four values.
    print( f"Raw History: {result[3][0].samples[0].timestamp} {result[3][0].samples[0].value.value}")
    print( f"Raw History: {result[3][0].samples[1].timestamp} {result[3][0].samples[1].value.value}")
    print( f"Raw History: {result[3][0].samples[2].timestamp} {result[3][0].samples[2].value.value}")
    print( f"Raw History: {result[3][0].samples[3].timestamp} {result[3][0].samples[3].value.value}")

The history is returned in result[3]. See the documentation for further details about the returned structure.

7. How to read raw historic data values and their times using RawValues and RawTimeStamps

A. You need to call separate methods for value and time, then combine the results.

Example code:

    P = connection.find_object('WeatherPoints.Temperature')
    arguments = [ Variant(VariantType.Date, datetime.now() - timedelta(hours=1)),
                  Variant(VariantType.Date, datetime.now() ),
                  Variant(VariantType.I4, 0),
                  Variant(VariantType.I4, 1000),
                  Variant(VariantType.Bool, True),
                  Variant(VariantType.BStr, "") ]
    vresult = connection.invoke_method( P.id, "Historic.RawValues", arguments )
    # Returns a variant which is an array of variant numbers
    print( vresult.value[0] ) # First value
    tresult = connection.invoke_method( P.id, "Historic.RawTimeStamps", arguments )
    # Returns a variant which is an array of datetime
    print( tresult.value[0] ) # First time
    # Combine the values and times using zip
    vtresult = zip([x.value for x in vresult.value], [str(x) for x in tresult.value])
    for v, t in vtresult:
        print( v,t)

8. My program running in the Geo SCADA server needs to run on change and get the input

A. Use the trigger type to confirm the program runs on change, and use the geo_scada_args variable to read the properties of the change information. Click the 'On Object Change' checkbox in the program properties and enter the name of an object which is to cause the program to run. Be careful not to run such programs often, and limit the execution queue size.

Example code:

import geo_scada_types
from geoscada.client.interface_scx import InterfaceScx
from typing import Optional

def execute(program_path: str, program_name: str, trigger_type: geo_scada_types.ProgramTrigger, geo_scada_args: dict, connection: Optional[InterfaceScx], *args, **kwargs):

    print(trigger_type._name_)
    print(f"Object #{geo_scada_args["ObjectId"]} '{geo_scada_args["ObjectName"]}' changed." )
    # Note that if the object is a point, the new value would be passed in as: geo_scada_args["Value"]

9. I need to run a Python program from ViewX and pass in arguments

A. You cannot directly do this from a pick action because the pick action generator does not know how many arguments your program takes, or their type. It is simple to create a small script which presents the arguments.

Example script (VBScript):

sub RunPython()
	groupname = Server.GetOPCValue(".FullName")
	set p = Server.FindObject( groupname + ".Program")
	p.Interface.Run 1, 2, 3
end sub

Example Python program:

import geo_scada_types
from geoscada.client.interface_scx import InterfaceScx
from typing import Optional
def execute(program_path: str, program_name: str, trigger_type: geo_scada_types.ProgramTrigger, geo_scada_args: dict, connection: Optional[InterfaceScx], *args, **kwargs):
    A1 = geo_scada_args["Arg1"]
    A2 = geo_scada_args["Arg2"]
    A3 = geo_scada_args["Arg3"]

Please note that the original release of Geo SCADA 2025 has an error which causes the parameters to be sent in reverse order, so in this example A1 will contain the argument for Arg3. This is corrected in the November 2025 update to Geo SCADA 2025.

10. How to call a Python program from Structured Text Logic

A. Simply declare the Run method with the arguments you need.

Example Logic program:

PROGRAM RunPythonWithArguments
METHOD
	Run AT %M(.Program.Run): DINT, DINT, DINT;
END_METHOD
	Run( 1, 2, 3);
END_PROGRAM

11. Why does my program not work when run as a Geo SCADA object?

A. Have you configured the user? Otherwise the connection will be null:

Here's a typical error for the connection: 'NoneType' object has no attribute ...

Traceback (most recent call last):
  File "C:\Windows\TEMP\{F5661CB1-4AC3-41F7-A42C-43599FCD29CC}\geo_scada_bootstrap.py", line 60, in <module>
    sys.exit(user_program.execute("Python Samples.01 Get Root Object Status", "Program", geo_scada_types.ProgramTrigger.Manual, geo_scada_args, None))
             ~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Windows\TEMP\{F5661CB1-4AC3-41F7-A42C-43599FCD29CC}\Program.py", line 12, in execute
    status = connection.get_object_status(0)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'get_object_status'

Ensure you have a user:

📖 Home  

📖 Python article index