OpenGalactic logo
HomeSource CodeContribute

Tutorial 3: Scepter-Javelin comm­unication

Now that you have been introduced to Scepter and Javelin, and have become more familiar with what they are and how they work, we can now build the workbench.
This tutorial will focus on preparing the database with all the relevant information regarding the satellite and ground station couple for our workbench. We will then go through the process of selecting a contact instruction and converting the database information into a command ready to send in the next tutorial.

Installation and environment setup

For this tutorial, we will use the OSSO Tutorials environment. To configure the environment, run the following commands from the base directory.
1
python -m venv venv
2
source venv/bin/activate
3
pip install --upgrade pip
4
pip install -r requirements.txt

Workbench section 1: Scepter setup

Tutorial 1 showed us how to communicate with Scepter and gave a brief explanation on the formats and inputs necessary. This tutorial will take that one step further and take users through the creation of both a ground station and satellite in the Scepter database, including the commands which will be sent. In the OSSO workbench directory, this software is given in the form of packages named new_satellite and new_groundstation, the third package given in the directory (scepter_other) provides other scepter endpoints not under the satellite or ground station sections but useful for this tutorial and beyond. The following process is done using Python function examples. Equivalent curl commands have been given, however, we do recommend that for applying multiple commands at once, you use the Python scripts.

Creating the ground station

The first thing we can do is create a ground station. For this, we will use the ground station located at Sydney University, however, you may choose any ground station you like and simply update the example input files to suit. Let's create a new ground station!
First, we need to create the initial overview. This will essentially serve as our frame which we can add the remaining components and information to. For our example, we chose the USYD ground station, however, you may use any ground station you like. To do so, simply update the JSON files in the examples directory or change the file paths in the Python functions. The following function and JSON were used for this step.
1
def new_groundstation():
2
    url = base_url + '/groundstation'
3
    filename = 'new_groundstation/examples/new_groundstation_overveiw.json'
4
    with open(filename, 'r') as f:
5
        input_json = json.load(f)
6
    response = requests.post(url=url, headers=headers, json=input_json)
7
    return response.json()
1
{
2
  "gs_name": "USYD Groundstation",
3
  "gs_hostname": "localhost",
4
  "gs_port": 5000,
5
  "commission_date": "2022-07-12T00:10:01.439Z",
6
  "latitude": -33.88969741038586,
7
  "longitude":  151.19403072520822,
8
  "altitude": 41.2,
9
  "address": "Engineering Walk, Darlington NSW 2008",
10
  "config": "BSPK",
11
  "publicly_visible": false
12
}
Following this, we can add the telecommunications information. Although this won't be used for this workbench, it is a good thing to help grasp the concepts around satellite operations. To do this, we used the following function and input.
1
def new_groundstation_telecom(gs_id):
2
    url = base_url + '/groundstation/' + gs_id + '/telecom'
3
    # filename = input('Enter input filename :')
4
    filename = 'new_groundstation/examples/new_groundstation_telecom.json'
5
    with open(filename, 'r') as f:
6
        input_json = json.load(f)
7
    response = requests.post(url=url, headers=headers, json=input_json)
8
    return response.json()
1
{
2
  "frequency_centre": 145e6,
3
  "frequency_lower": 144e6,
4
  "frequency_upper": 146e6,
5
  "telecom_name": "ISS Telecom",
6
  "telecom_type": "Antenna",
7
  "antenna_gain": 0,
8
  "modulations": {},
9
  "rotatable": true,
10
  "beamwidth": 10,
11
  "noise_factor": 1,
12
  "transmits": true,
13
  "receives": true,
14
  "address": "Engineering Walk, Darlington NSW 2008",
15
  "port": 5000
16
}
Finally, we want to update our ground station's status to active. To do this, the following function and JSON was used. Notice how the JSON requires a satellite ID, for ours we placed one of our existing satellites, which was then updated later on. For yours, we recommend that you create a satellite overview (steps shown below) to get the satellite ID needed for this step.
1
def new_groundstation_status(gs_id):
2
    url = base_url + '/groundstation/' + gs_id + '/status'
3
    # filename = input('Enter input filename :')
4
    filename = 'new_groundstation/examples/new_groundstation_status.json'
5
    with open(filename, 'r') as f:
6
        input_json = json.load(f)
7
    response = requests.post(url=url, headers=headers, json=input_json)
8
    return response.json()
1
{
2
  "update_time": "2022-07-13T00:42:27.223Z",
3
  "device": "antenna",
4
  "properties": {},
5
  "status": "OK",
6
  "errors": "minimal",
7
  "sat_tracking": "s012bf7"
8
}
Congratulations, you have now made a new ground station in Scepter.

Creating a satellite

Similar to the ground station, the first step in creating a satellite in Scepter is to create the overview. This will generate and return a satellite ID which should be saved as it will be used extensively later on. To complete this step, the following function and JSON was used.
1
def new_sat_overview():
2
    url = base_url + '/satellite'
3
    # input_filename = input('Enter filename/path with new satellite schema: ')
4
    input_filename = "new_satellite/examples/new_satellite_overview.json"
5
    with open(input_filename, 'r') as json_input:
6
        json_input = json.load(json_input)
7
        response = requests.post(url=url, headers=headers, json=json_input)
8
    return response.json()
1
{
2
  "commission_date": "2000-11-02T00:00:00",
3
  "decoder": "SSTV",
4
  "description": "The international space station",
5
  "encoder": "SSTV",
6
  "launch_date": "1998-11-20T00:00:00",
7
  "norad_id": "25544",
8
  "nssdc_id": "1998-067A",
9
  "publicly_visible": true,
10
  "sat_name": "ISS (ZARYA)",
11
  "tle": {
12
    "norad_cat_id": "25544",
13
    "tle_line_0": "ISS(ZARYA)",
14
    "tle_line_1": "1 25544U 98067A   08264.51782528 -.00002182  00000-0 -11606-4 0  2927",
15
    "tle_line_2": "2 25544  51.6416 247.4627 0006703 130.5360 325.0288 15.72125391563537",
16
    "last_updated": "22-06-21T17:52:30",
17
    "object_type": "Space Station"
18
  },
19
  "config": {}
20
}
As mentioned, this will return a JSON containing a satellite ID. To make matters easier later on, it is recommended that you save this in a variable via the following lines.
1
new_sat = new_sat_overview()
2
sat_id = new_sat['sat_id']
Now we can start adding in extras to our satellite. Our first step will be to add a new subsystem. This will hold the information about our satellite's On Board Computer (OBC) which we will look to ping later. To add a subsystem, the following Python function and JSON input is used.
1
def add_subsystem(sat_id):
2
    url = base_url + '/satellite' + '/' + sat_id + '/systems'
3
    # input_filename = input('Enter filename for input: ')
4
    input_filename = "new_satellite/examples/new_satellite_subsystem.json"
5
    with open(input_filename, 'r') as input_json:
6
        input_json = json.load(input_json)
7
    response = requests.post(url=url, headers=headers, json=input_json)
8
    return response.json()
1
{
2
  "name": "On Board Computer",
3
  "address": "localhost:8000"
4
}
Following on from this, we can also add our telecommunications information.
1
def new_telecom(sat_id):
2
    url = base_url + '/satellite/' + sat_id + '/telecom'
3
    # filename = input('Enter input filename: ')
4
    filename = 'new_satellite/examples/new_satellite_telecom.json'
5
    with open(filename, 'r') as f:
6
        input_json = json.load(f)
7
    response = requests.post(url=url, headers=headers, json=input_json)
8
    return response.json()
1
{
2
  "frequency": 145e6,
3
  "telecom_name": "ISS Telecom",
4
  "telecom_type": "SSTV",
5
  "modulations": {},
6
  "transmits": true,
7
  "receives": true,
8
  "encrypted": true
9
  }
The next additions will be important to the workbench; these are the command, script, and contact additions. The format of some of these are not set in stone, however, the functions and process designed for this workbench are based on these formats, so it is recommended that for an initial use, the following command formats are maintained. To add a command to Scepter, use the following function and input:
1
def update_command(command_id, sat_id):
2
    url = base_url + '/satellite/' + sat_id + '/commands/' + command_id
3
    # filename = input('Enter input filename: ')
4
    filename = "new_satellite/examples/new_satellite_command.json"
5
    with open(filename, 'r') as f:
6
        input_json = json.load(f)
7
    response = requests.put(url=url, headers=headers, json=input_json)
8
    return response.json()
1
{
2
  "name": "OBC Health Check",
3
  "args": {"services": ["app", "monitor", "scheduler", "telem"],
4
  "service_ports":  [8000, 8030, 8060, 8006],
5
  "command_type": "query{ping}"},
6
  "entrypoint": "localhost",
7
  "requires_approval": false,
8
  "encoder_data": {},
9
  "code": "graphQL"
10
}
This command outlines the services and GraphQL command that we want to send to the OBC. For all of them, we simply want to ping the service to check that each is functioning. We put in other little bits of information such as the entry point and the port numbers for records. When adding a new command, you can set the command ID as seen above. It is worth keeping this as a variable for use later on.
In a similar way, a script can also be added to Scepter through the following.
1
def update_script(sat_id, script_id):
2
    url = base_url + '/satellite/' + sat_id + '/scripts/' + script_id
3
    # filename = input('Enter input filename: ')
4
    filename = 'new_satellite/examples/update_satellite_script.json'
5
    with open(filename, 'r') as f:
6
        input_json = json.load(f)
7
    response = requests.put(url=url, headers=headers, json=input_json)
8
    return response.json()
1
{
2
  "name": "OBC1 Script",
3
  "args": {},
4
  "entrypoint": "localhost"
5
}
Since we are not going to be using the script anywhere further on, we decided to leave the args section blank, however you may add any JSON dictionary you like.
Finally, we can create a contact for this satellite. This can be done through the following.
1
def new_contact(sat_id):
2
    url = base_url + '/satellite/' + sat_id + '/contacts'
3
    filename = "new_satellite/examples/new_satellite_contact.json"
4
    with open(filename, 'r') as f:
5
        input_json = json.load(f)
6
    response = requests.post(url=url, headers=headers, json=input_json)
7
    return response.json()
1
{
2
  "instructions": [
3
    {
4
      "script_id": "OBC1",
5
      "args": {},
6
      "sequence_number": 0
7
    },
8
    {
9
      "command_id": "OBC1",
10
      "args": {},
11
      "sequence_number": 0
12
    }
13
  ],
14
  "gs_id": "g06cc6",
15
  "issued_at": "2022-07-11T05:58:40"
16
}
Once this has run successfully, you will need to extract and save the contact ID from the output. This can be done in a similar was as the satellite and ground station IDs.
Finally, just like our ground station, we can update the status of our satellite to show it is active.
1
def new_sat_status(sat_id):
2
    url = base_url + '/satellite/' + sat_id + '/status'
3
    # filename = input('Enter input filename: ')
4
    filename = "new_satellite/examples/new_satellite_status.json"
5
    with open(filename, 'r') as f:
6
        input_json = json.load(f)
7
    response = requests.post(url=url, headers=headers, json=input_json)
8
    return response.json()
1
{
2
  "update_time": "2022-07-11T05:00:00",
3
  "status": "active"
4
}
Well done! You have now completed the first section of the workbench.

Workbench section 2: preparing for Javelin

To prepare the commands for Javelin, the following steps are undertaken:
  1. Grab the contact information from Scepter.
  2. Using this, grab the relevant information regarding the commands.
  3. Convert this information into the required format, ready to be used by Javelin.
Once we have set up our satellite and ground station, we can begin sending commands to the satellite. Our first step is to select a contact for us to use in this operation. This can be done through the following function.
1
def get_single_contact(sat_id, contact_id):
2
    url = base_url + '/satellite/' + sat_id + '/contacts/' + contact_id
3
    response = requests.get(url=url, headers=headers)
4
    return response.json()
5


6
sat_id = "s012bf7"
7
contact_id = "1f1ca968-dc49-47cb-91ef-e47581a21a3b"
8
pprint.pprint(get_single_contact(sat_id, contact_id))
If users wish to see a complete list of contacts, the following two functions will provide the relevant information.
1
def list_of_contacts(sat_id):
2
    url = base_url + '/satellite/' + sat_id + '/contacts'
3
    filename = 'new_satellite/examples/list_satellite_contacts.json'
4
    with open(filename, 'r') as f:
5
        input_json = json.load(f)
6
    response = requests.get(url=url, headers=headers, params=input_json)
7
    return response.json()
8


9


10
def list_contact_ids(contact_list):
11
    print('\nContact IDs: Creation Datetime: Dispatched')
12
    print('----------------------------------------------')
13
    for n in range(len(contact_list['contacts'])):
14
        string = contact_list['contacts'][n]['id'] + ': ' + contact_list['contacts'][n]['created_at'] + ': ' + \
15
                 str(contact_list['contacts'][n]['dispatched_at'])
16
        print(string)
17
    print('------------------------------------------------\n')
18


19


20
# To use these functions
21
if __name__ == '__main__':
22
  sat_id = "s012bf7"
23
  list_contact_ids(list_of_contacts(sat_id))
24


25
"""
26
The input format for the .JSON file
27
{
28
  "upcoming": true,
29
  "include_cancelled": true,
30
  "cursor": "string",
31
  "page_size": 100
32
}
33
"""
A contact contains all the information we need to communicate with our satellite. It states which overpass and ground station we need to use, when this contact was issued and dispatched, who issued the contact, and the instructions to be completed.
For this example, we will be completing the command instruction for the following contact.
1
{
2
  "cancelled_at": None,
3
  "created_at": "2022-07-15T13:49:39.670866",
4
  "dispatched_at": "2022-07-15T04:11:41.687566",
5
  "id": "1f1ca968-dc49-47cb-91ef-e47581a21a3b",
6
  "instructions": [
7
    {
8
      "args": {},
9
      "ended_at": None,
10
      "entrypoint": "localhost",
11
      "id": 0,
12
      "log": None,
13
      "script_id": "OBC1",
14
      "sequence_number": 0,
15
      "started_at": None,
16
      "type": "script"
17
    },
18
    {
19
      "args": {},
20
      "command_id": "OBC1",
21
      "ended_at": "2022-07-11T06:27:32",
22
      "entrypoint": "localhost",
23
      "id": 1,
24
      "log": "string",
25
      "sequence_number": 0,
26
      "started_at": "2022-07-11T06:22:43",
27
      "type": "command"
28
    }
29
  ],
30
  "issued_at": "2022-07-11T05:58:40",
31
  "issued_by": "user1",
32
  "overpass": {
33
    "AOS": "2022-07-15T03:50:09.680280",
34
    "LOS": "2022-07-15T03:51:09.680280",
35
    "TCA": "2022-07-15T03:50:39.680280",
36
    "gs_id": "g06cc6",
37
    "id": "3f6fa738-4483-4f1e-898f-be664d66c057",
38
    "maximum_el": 73.0,
39
    "sat_id": "s23d752",
40
    "tle_updated_at": "2022-07-15T03:50:09.680280"
41
  }
42
}
For tutorials 3 and 4, the overpass information will be neglected as we will be sending our commands immediately. However, Tutorial 5 will upgrade this process to account for the overpass requirements of this contact.
As stated, we will need the command instructions from this contact for our example. This was done using the following function.
1
def instructions_from_contact(sat_id, contact_id):
2
    contact = get_single_contact(sat_id, contact_id)
3


4
    if contact['dispatched_at'] == 'None':
5
        contact_dispatch = dispatch_contact(sat_id, contact_id)
6
        pprint.pprint(contact_dispatch)
7


8
    instructions = contact['instructions']
9
    pprint.pprint(instructions)
10
    return instructions
11


12
# To use this function
13
sat_id = "s012bf7"
14
contact_id = "1f1ca968-dc49-47cb-91ef-e47581a21a3b"
15
instructions = instructions_from_contact(sat_id, contact_id)
Once we have our list of instructions, we can sort them into command instructions and script instructions. For the example above, we will have one of each. We did this by using the following loop.
1
cmd_instruction_ids = []
2
    script_instruction_ids = []
3
    for n in range(len(instructions)):
4
        if instructions[n]['type'] == 'command':
5
            cmd_instruction_ids.append(int(instructions[n]['id']))
6
        elif instructions[n]['type'] == 'script':
7
            script_instruction_ids.append(int(instructions[n]['id']))
To grab the finer details of the command, we will use the following function to get the specific command ID we want to execute. This function will also notify Scepter that we have started to execute this command and to dispatch the instruction if it has not already been done.
1
def start_cmd_instruction(instruction_id: int, sat_id, contact_id):
2
    contact = get_single_contact(sat_id, contact_id)
3
    instructions = contact['instructions']
4
    cmd_id = None
5
    if instructions[instruction_id]['type'] != 'command':
6
        print('invalid Instruction Type')
7
        return
8
    else:
9
        cmd_id = instructions[instruction_id]['command_id']
10


11
    pprint.pprint(start_contact_instructions(sat_id, contact_id, str(instruction_id)))
12


13
    return cmd_id
14


15
# To use this
16
cmd_id = start_cmd_instruction(cmd_instruction_ids[0], "s012bf7", "1f1ca968-dc49-47cb-91ef-e47581a21a3b")
Once Scepter has been notified, we can now grab our command information and convert it into a form that Javelin will understand. This is done through the following function.
1
def scepter_cmd_to_input(cmd_id, sat_id):
2
    # Get the cmd output
3
    cmd = get_command(cmd_id, sat_id)
4
    cmd_args = cmd['args']
5
    cmd_id = cmd['id']
6
    cmd_name = cmd['name']
7
    cmd_code = cmd['code']
8
    print('\nConverting Commands to Javelin Input...\n')
9
    time.sleep(5)
10
    print('Command ID: %s' % cmd_id)
11
    print('Command Name: %s' % cmd_name)
12
    if cmd['requires_approval']:
13
        print('This command has not been approved...\n')
14
        check = input('Would you like to continue with the command send? (Y/N):')
15
        if check == 'Y' or check == 'y':
16
            pass
17
        else:
18
            print('...Workbench Exited')
19
            return None
20
        print('Preparing Commands to be Sent:')
21
    time.sleep(1)
22
    print('Commands to be sent:')
23
    time.sleep(1)
24
    cmd_lines = []
25
    for n in range(len(cmd_args['services'])):
26
        line = convert_cmd_args(service=cmd_args['services'][n], cmd_type=cmd_args['command_type'])
27
        print(line)
28
        cmd_lines.append(line)
29
        time.sleep(1)
30
    return cmd_lines
31


32
# The python script to use this function
33
sat_id = "s012bf7"
34
cmd_id = start_cmd_instruction(cmd_instruction_ids[0], "s012bf7", "1f1ca968-dc49-47cb-91ef-e47581a21a3b")
35
cmd_lines = scepter_cmd_to_input(cmd_id, sat_id)
The command lines are what the send_cmd_script.py file used in the previous tutorial use to communicate to the satellite.
Congratulations, you have successfully used a contact in Scepter to prepare satellite commands that are ready to be sent. This tutorial will now follow on to Tutorial 4 where we will go through the setup of a KubOS OBC (or placeholder) and send the commands.
Table of contents