Slicer Failover API

import json
import requests
from api_auth import APICredentials, APIParams

class CreateFailoverGroup:
    def __init__(self):
        self.host = "https://services.uplynk.com"

    def run(self):
        self._create_failover_group()

    def _create_failover_group(self):
        
        failover_group_name = 'MyFailoverGroup' # Replace with the desired failover group name.
            
        url = "{}{}".format(self.host, "/api/v4/failover-groups/")

        payload = {
            'name': failover_group_name
        }

        headers = {'Content-Type': 'application/json'}
  
        response = requests.post(
            url, params=APIParams(APICredentials()).get_params({}), data=json.dumps(payload), headers=headers
        )
  
        print(response.status_code)
  
CreateFailoverGroup().run() 

import json
import requests
from api_auth import APICredentials, APIParams

class DeleteFailoverGroup:
    def __init__(self):
        self.host = "https://services.uplynk.com"

    def run(self):
        self._delete_failover_group()

    def _delete_failover_group(self):
        failover_group_id = 'd22a96e815f241319677659316d3fb0f' # Replace with the desired failover group ID.
        
        url = "{}{}{}".format(self.host, "/api/v4/failover-groups/", failover_group_id)

        headers = {'Content-Type': 'application/json'}
  
        response = requests.delete(
            url, params=APIParams(APICredentials()).get_params({}), headers=headers
        )
  
        print(response.status_code)
  
DeleteFailoverGroup().run()

{ 
  "auto_failback": false, 
  "channels": [ "99994a11ead446e7b4d7a4c91c679999", "88884a11ead446e7b4d7a4c91c678888" ],
  "mode": "prioritized", 
  "name": "My Failover Group", 
  "slicers": { "slicer1": { "force_blacklist": false, "priority": 1 },
               "slicer2": { "force_blacklist": false, "priority": 2 } 
             }, 
  "thresholds": {	
      "failover_audio_loss": { "fault_duration": 90, "recovery_duration": 60, "enabled": false, "severity": 5 }, 
      "failover_blackness": { "low": 1, "fault_duration": 120, "recovery_duration": 60, "enabled": false , "severity": 5 },
      "failover_cc_last_seen": { "fault_duration": 60, "recovery_duration": 90, "enabled": false , "severity": 5 },
      "failover_dropped": { "high": 0, "fault_duration": 5, "recovery_duration": 6, "enabled": false , "severity": 5 },
      "failover_input": { "fault_duration": 0, "recovery_duration": 30, "enabled": false , "severity": 5 },
      "failover_nielsen_last_seen": { "fault_duration": 60, "recovery_duration": 90, "enabled": false , "severity": 5 },
      "failover_proc_q": { "high": 8, "fault_duration": 10, "recovery_duration": 30, "enabled": false , "severity": 5 },
      "failover_queue": { "high": 5, "fault_duration": 5, "recovery_duration": 5, "enabled": false , "severity": 5 },
      "failover_scte_last_seen": { "fault_duration": 60, "recovery_duration": 90, "enabled": false , "severity": 5 },
      "failover_static_audio": { "fault_duration": 90, "recovery_duration": 60, "enabled": false , "severity": 5 },
      "failover_static_video": { "fault_duration": 90, "recovery_duration": 60, "enabled": false , "severity": 5 },
      "failover_tr_101_290_stats_P1_errors": { "high": 1, "fault_duration": 0, "recovery_duration": 30, "enabled": false, "severity": 5 },
      "failover_tr_101_290_stats_P2_errors": { "high": 1, "fault_duration": 0, "recovery_duration": 30, "enabled": false , "severity": 5 },
      "failover_video_loss": { "fault_duration": 90, "recovery_duration": 60, "enabled": false },
      "failover_involuntary_blackout": { "enabled": false, "severity":5 }
                } 
}

import json
import requests
from api_auth import APICredentials, APIParams

class UpdateFailoverGroup:
    def __init__(self):
        self.host = "https://services.uplynk.com"

    def run(self):
        self._update_failover_group()

    def _update_failover_group(self):
        failover_group_id = '99f7b514f79440bb940812f2eb8954d7' # Replace with the desired failover group ID.

        url = "{}{}{}".format(self.host, "/api/v4/failover-groups/", failover_group_id)
        
        payload = {
            # Only include 'name, 'slicers', or 'thresholds' if changing values.
            'name': 'MyFailoverGroup (Test)',       # Change name of failover group.
            'slicers': { 
                            "main_slicer": { "force_blacklist": False , "priority": 2},     # Make priority 2.
                            "backup_slicer": { "force_blacklist": False , "priority": 1}    # Make priority 1.
                       },
            'thresholds': {
                            'failover_static_audio': {
                                                        'enabled': True,       # Enable static audio threshold.
                                                        'severity': 3          # Lower severity from default
                                                     },
                            'failover_tr_101_290_stats_P1_errors': {
                                                                     # Enable P1 error threshold.
                                                                     "high": 3,
                                                                     "fault_duration": 1,
                                                                     "recovery_duration": 60,
                                                                     "enabled": True,
                                                                     "severity": 4
                                                                   }
                          }
        }

        headers = {'Content-Type': 'application/json'}
  
        response = requests.patch(
            url, params=APIParams(APICredentials()).get_params({}), data=json.dumps(payload), headers=headers
        )
        
        data=json.dumps(payload)
        print(data)
  
        print(response.status_code)
  
UpdateFailoverGroup().run()

import base64
import os
import zlib, hmac, hashlib, time, json


class APICredentials:
    """
    Stores credentials required to request our API.
    """
    @property
    def user_id(self):
        """
        Set your user ID to the one defined on the User Settings page.
        """
        return "1234567890abcdefghijklmnopqrstu"

    @property
    def secret(self):
        """
        Set your API key to a value defined on the Integration Keys page. 
        """
        return "1234567890abcdefghijklmnopqrstuvwxyz1234"


class APIParams(object):
    """
    Provides API authentication. Learn more at:
        https://docs.uplynk.com   
    """
    def __init__(self, credentials):
        self.credentials = credentials

    def get_params(self, data):
        """
        Encodes and signs <data> into the expected format and returns it.
        """
        data = self._get_params(**data)
        data.update(data)
        return data

    def _get_msg(self, msg=None):
        """
        Encodes and returns the 'msg' parameter.
        """
        msg = msg if msg else {}

        msg.update({
            '_owner': self.credentials.user_id,
            '_timestamp': int(time.time())
        })

        msg = json.dumps(msg)
        msg_compressed = zlib.compress(msg.encode(), 9)
        return base64.b64encode(msg_compressed).strip()

    def _get_params(self, **msg):
        """
        Returns the message and its signature.
        """
        msg = self._get_msg(msg)

        sig = hmac.new(
            self.credentials.secret.encode(), msg, hashlib.sha256
        ).hexdigest()

        return {
            'msg': msg,
            'sig': sig
        }