standalone-lights & logging & more #35
1
.gitignore
vendored
1
.gitignore
vendored
@ -5,3 +5,4 @@ credentials/
|
||||
.idea/
|
||||
stream_sources/chyron.txt
|
||||
stream_sources/brb.txt
|
||||
*.log
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
FROM python:3.10.0a7-alpine3.13
|
||||
FROM python:3.7.10-alpine3.12
|
||||
|
||||
WORKDIR /Praxis
|
||||
|
||||
|
||||
11
Dockerfile_standalone_lights
Normal file
11
Dockerfile_standalone_lights
Normal file
@ -0,0 +1,11 @@
|
||||
FROM python:3.7.10-alpine3.12
|
||||
|
||||
WORKDIR /Praxis
|
||||
|
||||
COPY requirements_sa_command.txt requirements_sa_command.txt
|
||||
RUN apk add --update gcc libc-dev linux-headers && rm -rf /var/cache/apk/*
|
||||
RUN pip3 install -r requirements_sa_command.txt
|
||||
|
||||
COPY . .
|
||||
|
||||
CMD [ "python3", "standalone_lights.py"]
|
||||
@ -2,6 +2,10 @@ from abc import ABCMeta
|
||||
|
||||
from channel_rewards.channelRewards_base import AbstractChannelRewards
|
||||
|
||||
from json import loads
|
||||
from urllib.parse import urlencode
|
||||
import requests
|
||||
|
||||
class ChannelReward_Hydration_v2(AbstractChannelRewards, metaclass=ABCMeta):
|
||||
"""
|
||||
this is the hydration reward.
|
||||
@ -15,7 +19,26 @@ class ChannelReward_Hydration_v2(AbstractChannelRewards, metaclass=ABCMeta):
|
||||
|
||||
def do_ChannelReward(self, source = AbstractChannelRewards.ChannelRewardsSource.default, user = "User", rewardName = "", rewardPrompt = "", userInput = "", bonusData = None):
|
||||
|
||||
#print("sending:",user, 16, "!lights hydration")
|
||||
self.dothething(user, 16, "!lights hydration", "")
|
||||
return None
|
||||
|
||||
def dothething(self, username, light_group, command, rest):
|
||||
# todo need to url-escape command and rest
|
||||
params = urlencode({'user_name': username, 'light_group': light_group, 'command': command, 'rest':rest})
|
||||
#standalone_lights
|
||||
url = "http://standalone_lights:42069/api/v1/exec_lights?%s" % params
|
||||
resp = requests.get(url)
|
||||
if resp.status_code == 200:
|
||||
print("Got the following message: %s" % resp.text)
|
||||
data = loads(resp.text)
|
||||
msg = data['message']
|
||||
if msg is not None:
|
||||
return msg
|
||||
# todo send to logger and other relevent services
|
||||
else:
|
||||
# todo handle failed requests
|
||||
pass
|
||||
|
||||
def get_help(self):
|
||||
return self.help
|
||||
@ -1,9 +1,11 @@
|
||||
from abc import ABCMeta
|
||||
|
||||
import lights_module
|
||||
from commands.command_base import AbstractCommand
|
||||
|
||||
import utilities_script as utility
|
||||
from json import loads
|
||||
from urllib.parse import urlencode
|
||||
import requests
|
||||
|
||||
|
||||
class Command_lights_v2(AbstractCommand, metaclass=ABCMeta):
|
||||
"""
|
||||
@ -19,57 +21,28 @@ class Command_lights_v2(AbstractCommand, metaclass=ABCMeta):
|
||||
|
||||
def do_command(self, source = AbstractCommand.CommandSource.default, user = "User", command = "", rest = "", bonusData = None):
|
||||
returnString = ""
|
||||
print("\n Command>: " + command + rest)
|
||||
|
||||
tempBool = True
|
||||
if tempBool == True:
|
||||
LightModule = lights_module.Lights_Module()
|
||||
LightModule.main()
|
||||
#bot.return_message("\nRGB Command Detected!")
|
||||
tempFix = command + " " + rest
|
||||
|
||||
tempParsedMessage = tempFix.split(" ")
|
||||
sceneCommand = False
|
||||
if (len(tempParsedMessage)) > 2:
|
||||
#bot.return_message("RGB Command!")
|
||||
rgb_r = float(tempParsedMessage[1])
|
||||
rgb_g = float(tempParsedMessage[2])
|
||||
rgb_b = float(tempParsedMessage[3])
|
||||
xy_result = LightModule.rgb_to_xy(rgb_r, rgb_g, rgb_b)
|
||||
#bot.return_message("got XY")
|
||||
LightModule.bridge_.set_group(16, "xy", xy_result)
|
||||
#bot.return_message("sent color to [Lights_Module]")
|
||||
else:
|
||||
if "stream" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
LightModule.bridge_.run_scene("Downstairs", "Stream")
|
||||
elif "normal" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
LightModule.bridge_.run_scene("Downstairs", "Bright")
|
||||
elif "haxor" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
LightModule.bridge_.run_scene("Downstairs", "hacker vibes")
|
||||
elif "off" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
LightModule.bridge_.set_group("Downstairs", "on", False)
|
||||
elif "on" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
LightModule.bridge_.set_group("Downstairs", "on", True)
|
||||
elif "ravemode" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
LightModule.raveMode()
|
||||
else:
|
||||
#bot.return_message("Color Command!")
|
||||
xy_result = LightModule.color_string_parser(tempParsedMessage)
|
||||
#bot.return_message("got XY")
|
||||
LightModule.bridge_.set_group(16, "xy", xy_result)
|
||||
#bot.return_message("sent color to [Lights_Module]")
|
||||
|
||||
#if sceneCommand == True:
|
||||
#bot.return_message("Scene Command!")
|
||||
|
||||
returnString = user + " changed the light's color!"
|
||||
returnString = self.dothething(user, 16, command, rest)
|
||||
|
||||
return returnString
|
||||
|
||||
def dothething(self, username, light_group, command, rest):
|
||||
# todo need to url-escape command and rest
|
||||
params = urlencode({'user_name': username, 'light_group': light_group, 'command': command, 'rest':rest})
|
||||
#standalone_lights
|
||||
url = "http://standalone_lights:42069/api/v1/exec_lights?%s" % params
|
||||
resp = requests.get(url)
|
||||
if resp.status_code == 200:
|
||||
print("Got the following message: %s" % resp.text)
|
||||
data = loads(resp.text)
|
||||
msg = data['message']
|
||||
if msg is not None:
|
||||
return msg
|
||||
# todo send to logger and other relevent services
|
||||
else:
|
||||
# todo handle failed requests
|
||||
return None
|
||||
|
||||
def get_help(self):
|
||||
return self.help
|
||||
@ -10,7 +10,7 @@ user_module: bool = True
|
||||
|
||||
autoJoin_TwitchChannel = "thecuriousnerd"
|
||||
autoJoin_TwitchChannels = ["thecuriousnerd"]
|
||||
whitelisted_TwitchPowerUsers = ["thecuriousnerd"]
|
||||
allowedCommandsList_TwitchPowerUsers = ["thecuriousnerd"]
|
||||
|
||||
#Twitch Module Configs
|
||||
block_TwitchChannelsMessaging = [""] # Blocks the ability to send messages to twitch channels
|
||||
|
||||
@ -2,25 +2,43 @@ version: '3.7'
|
||||
services:
|
||||
standalone_command:
|
||||
image: standalone_command
|
||||
volumes:
|
||||
- "./:/Praxis/"
|
||||
ports:
|
||||
- 6009:6009
|
||||
environment:
|
||||
- ISDOCKER=cat
|
||||
standalone_channelrewards:
|
||||
image: standalone_channelrewards
|
||||
volumes:
|
||||
- "./:/Praxis/"
|
||||
ports:
|
||||
- 6969:6969
|
||||
environment:
|
||||
- ISDOCKER=cat
|
||||
standalone_lights:
|
||||
image: standalone_lights
|
||||
volumes:
|
||||
- "./:/Praxis/"
|
||||
ports:
|
||||
- 42069:42069
|
||||
environment:
|
||||
- ISDOCKER=cat
|
||||
standalone_twitchscript:
|
||||
image: standalone_twitchscript
|
||||
volumes:
|
||||
- "./:/Praxis/"
|
||||
environment:
|
||||
- ISDOCKER=cat
|
||||
standalone_twitch_pubsub:
|
||||
image: standalone_twitch_pubsub
|
||||
volumes:
|
||||
- "./:/Praxis/"
|
||||
environment:
|
||||
- ISDOCKER=cat
|
||||
standalone_discordscript:
|
||||
image: standalone_discordscript
|
||||
volumes:
|
||||
- "./:/Praxis/"
|
||||
environment:
|
||||
- ISDOCKER=cat
|
||||
@ -1,5 +1,6 @@
|
||||
docker build --file Dockerfile_standalone_command --tag standalone_command .
|
||||
docker build --file Dockerfile_standalone_channelRewards --tag standalone_channelrewards .
|
||||
docker build --file Dockerfile_standalone_lights --tag standalone_lights .
|
||||
docker build --file Dockerfile_standalone_DiscordScript --tag standalone_discordscript .
|
||||
docker build --file Dockerfile_standalone_TwitchScript --tag standalone_twitchscript .
|
||||
docker build --file Dockerfile_standalone_Twitch_Pubsub --tag standalone_twitch_pubsub .
|
||||
14
praxis_logging.py
Normal file
14
praxis_logging.py
Normal file
@ -0,0 +1,14 @@
|
||||
import logging
|
||||
import utilities_script
|
||||
|
||||
class praxis_logger():
|
||||
def init(self, name):
|
||||
super().__init__()
|
||||
self.logName = "logs/" + name + ".log"
|
||||
utilities_script.get_dir("logs")
|
||||
logging.basicConfig(filename=self.logName, level=logging.DEBUG)
|
||||
logging.info('Application running!')
|
||||
|
||||
def log(self, msg):
|
||||
print(self.logName, msg)
|
||||
logging.info(msg)
|
||||
@ -4,6 +4,12 @@ from flask import request
|
||||
import channel_rewards.channelRewards_loader as rewards_loader
|
||||
from channel_rewards.channelRewards_base import AbstractChannelRewards
|
||||
|
||||
import os
|
||||
import praxis_logging
|
||||
praxis_logger_obj = praxis_logging.praxis_logger()
|
||||
praxis_logger_obj.init(os.path.basename(__file__))
|
||||
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
|
||||
|
||||
api = flask.Flask(__name__)
|
||||
# enable/disable this to get web pages of crashes returned
|
||||
api.config["DEBUG"] = True
|
||||
@ -12,7 +18,7 @@ loadedRewards = {}
|
||||
|
||||
def init():
|
||||
# todo load entire reward library and cache it here
|
||||
print("init stuff")
|
||||
praxis_logger_obj.log("init stuff")
|
||||
loadedRewards[AbstractChannelRewards.ChannelRewardsType.channelPoints] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.channelPoints)
|
||||
loadedRewards[AbstractChannelRewards.ChannelRewardsType.twitch_bits] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.twitch_bits)
|
||||
loadedRewards[AbstractChannelRewards.ChannelRewardsType.twitch_subs] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.twitch_subs)
|
||||
@ -22,11 +28,12 @@ def is_reward(reward_name, reward_type) -> bool:
|
||||
#global loadedRewards
|
||||
tempType = reward_type.replace('ChannelRewardsType.', '')
|
||||
realTempType = AbstractChannelRewards.ChannelRewardsType.__dict__[tempType]
|
||||
#praxis_logger_obj.log(loadedRewards[realTempType])
|
||||
|
||||
for reward in loadedRewards[realTempType]:
|
||||
print("found: ",reward,"type: ",type(reward))
|
||||
#praxis_logger_obj.log("found: ", reward, "type: ", type(reward))
|
||||
if reward_name == reward:
|
||||
print("Equal")
|
||||
praxis_logger_obj.log("Equal")
|
||||
return True
|
||||
|
||||
|
||||
@ -38,22 +45,25 @@ def is_reward(reward_name, reward_type) -> bool:
|
||||
|
||||
def handle_reward(source, username, reward_name, reward_type, rewardPrompt, userInput, bonusData):
|
||||
#reward:AbstractChannelRewards = loadedRewards[reward_name]
|
||||
tempType = reward_type.replace('ChannelRewardsType.', '')
|
||||
realTempType = AbstractChannelRewards.ChannelRewardsType.__dict__[tempType]
|
||||
reward:AbstractChannelRewards = loadedRewards[realTempType][reward_name]
|
||||
if reward is not None:
|
||||
reward_response = reward.do_ChannelReward(source, username, reward_name, rewardPrompt, userInput, bonusData)
|
||||
return flask.make_response("{\"message\":\"%s\"}" % reward_response, 200, {"Content-Type": "application/json"})
|
||||
|
||||
#print("Doing a reward")
|
||||
praxis_logger_obj.log("\n trying to handle reward: " + reward_name)
|
||||
try:
|
||||
tempType = reward_type.replace('ChannelRewardsType.', '')
|
||||
realTempType = AbstractChannelRewards.ChannelRewardsType.__dict__[tempType]
|
||||
reward:AbstractChannelRewards = loadedRewards[realTempType][reward_name]
|
||||
if reward is not None:
|
||||
reward_response = reward.do_ChannelReward(source, username, reward_name, rewardPrompt, userInput, bonusData)
|
||||
return flask.make_response("{\"message\":\"%s\"}" % reward_response, 200, {"Content-Type": "application/json"})
|
||||
except:
|
||||
return flask.make_response("This is a magic test", 500)
|
||||
#praxis_logger_obj.log("Doing a reward")
|
||||
|
||||
|
||||
@api.route('/api/v1/reward', methods=['GET'])
|
||||
def reward_check():
|
||||
if 'reward_name' in request.args and 'reward_type' in request.args:
|
||||
print("reward_name:", request.args['reward_name'],"reward_type:", request.args['reward_type'])
|
||||
#praxis_logger_obj.log("reward_name: "+ request.args['reward_name']+"reward_type: "+ request.args['reward_type'])
|
||||
if is_reward(request.args['reward_name'], request.args['reward_type']):
|
||||
print("about to send")
|
||||
praxis_logger_obj.log("about to send")
|
||||
return flask.make_response('', 200)
|
||||
else:
|
||||
return flask.make_response('', 404)
|
||||
@ -64,7 +74,7 @@ def exec_reward():
|
||||
if 'reward_name' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'reward_name\' not in request"}', 400)
|
||||
if 'reward_type' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'reward_name\' not in request"}', 400)
|
||||
return flask.make_response('{\"text\":"Argument \'reward_type\' not in request"}', 400)
|
||||
if 'reward_prompt' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'reward_prompt\' not in request"}', 400)
|
||||
if 'user_input' not in request.args:
|
||||
@ -78,7 +88,14 @@ def exec_reward():
|
||||
username = "User"
|
||||
else:
|
||||
username = request.args['user_name']
|
||||
praxis_logger_obj.log("\n About to try a reward")
|
||||
|
||||
praxis_logger_obj.log(request.args['reward_source'])
|
||||
praxis_logger_obj.log(request.args['reward_name'])
|
||||
praxis_logger_obj.log(request.args['reward_type'])
|
||||
praxis_logger_obj.log(request.args['reward_prompt'])
|
||||
praxis_logger_obj.log(request.args['user_input'])
|
||||
praxis_logger_obj.log(request.args['bonus_data'])
|
||||
return handle_reward(
|
||||
request.args['reward_source'],
|
||||
username,
|
||||
|
||||
@ -4,6 +4,12 @@ from flask import request
|
||||
import commands.loader as command_loader
|
||||
from commands.command_base import AbstractCommand
|
||||
|
||||
import os
|
||||
import praxis_logging
|
||||
praxis_logger_obj = praxis_logging.praxis_logger()
|
||||
praxis_logger_obj.init(os.path.basename(__file__))
|
||||
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
|
||||
|
||||
api = flask.Flask(__name__)
|
||||
# enable/disable this to get web pages of crashes returned
|
||||
api.config["DEBUG"] = True
|
||||
|
||||
@ -25,6 +25,12 @@ import discord.abc
|
||||
|
||||
from cooldowns import Cooldown_Module
|
||||
|
||||
import os
|
||||
import praxis_logging
|
||||
praxis_logger_obj = praxis_logging.praxis_logger()
|
||||
praxis_logger_obj.init(os.path.basename(__file__))
|
||||
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
|
||||
|
||||
class Discord_Module(discord.Client):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
@ -56,6 +62,9 @@ class Discord_Module(discord.Client):
|
||||
#print(message.author.mention)
|
||||
print(message.content)
|
||||
|
||||
debugLogString= "\n\n{" + message.guild.name + "}[ " + str(message.channel) + " ](" + message.author.display_name +")> " + message.content + "\n"
|
||||
praxis_logger_obj.log(debugLogString)
|
||||
|
||||
if not await self.isSenderBot(message):
|
||||
# This will check for the praxis_bot-tts channel and will TTS stuff from there.
|
||||
#await self.eval_triggeredEvents(message)
|
||||
|
||||
292
standalone_lights.py
Normal file
292
standalone_lights.py
Normal file
@ -0,0 +1,292 @@
|
||||
from time import sleep
|
||||
import phue
|
||||
from phue import Bridge
|
||||
|
||||
import random
|
||||
import utilities_script as utilities
|
||||
|
||||
import credentials
|
||||
import config
|
||||
|
||||
import flask
|
||||
from flask import request
|
||||
|
||||
import os
|
||||
import praxis_logging
|
||||
praxis_logger_obj = praxis_logging.praxis_logger()
|
||||
praxis_logger_obj.init(os.path.basename(__file__))
|
||||
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
|
||||
|
||||
api = flask.Flask(__name__)
|
||||
# enable/disable this to get web pages of crashes returned
|
||||
api.config["DEBUG"] = True
|
||||
|
||||
class Lights_Module():
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.bridge_:Bridge = Bridge('192.168.191.146')
|
||||
|
||||
def main(self):
|
||||
praxis_logger_obj.log("\nStarting up [Lights_Module]...")
|
||||
self.bridge_.connect()
|
||||
|
||||
self.bridge_.get_api()
|
||||
|
||||
light_list = self.bridge_.lights
|
||||
group_list:list = []
|
||||
groups = self.bridge_.get_group()
|
||||
groupCount = 0
|
||||
|
||||
#praxis_logger_obj.log("\n -Listing Lights...")
|
||||
for l in light_list:
|
||||
pass
|
||||
#praxis_logger_obj.log(l.name)
|
||||
#praxis_logger_obj.log("\n -Counting Groups...")
|
||||
for g in groups:
|
||||
#praxis_logger_obj.log(g)
|
||||
groupCount = int(g)
|
||||
|
||||
|
||||
for gc in range(groupCount):
|
||||
try:
|
||||
#praxis_logger_obj.log("group n:" + str(gc))
|
||||
group = self.bridge_.get_group(gc ,'name')
|
||||
#praxis_logger_obj.log(group)
|
||||
group_list.append(group)
|
||||
#praxis_logger_obj.log(" --done adding")
|
||||
except:
|
||||
pass
|
||||
#praxis_logger_obj.log(" --adding failed")
|
||||
|
||||
#self.bridge_.set_group(18, "bri", 254) #This is max Brightness
|
||||
#self.bridge_.set_group(18, "on", True) #This is will turn ON
|
||||
#xy_result = self.rgb_to_xy(0,0,1) #This will take an rgb value and make it xy
|
||||
#self.bridge_.set_group(16, "xy", xy_result) #This will make the lights in the group turn blue
|
||||
|
||||
# The Following will make a rave
|
||||
#for rave in range(10):
|
||||
#rgb_r = random.random()
|
||||
#rgb_g = random.random()
|
||||
#rgb_b = random.random()
|
||||
#xy_result = self.rgb_to_xy(rgb_r, rgb_g, rgb_b) #This will take an rgb value and make it xy
|
||||
#self.bridge_.set_group(16, "xy", xy_result)
|
||||
#sleep(0.1)
|
||||
|
||||
#for stuffz in self.bridge_.scenes:
|
||||
#praxis_logger_obj.log(stuffz)
|
||||
|
||||
# This will set the group Downstairs to the Stream scene
|
||||
#self.bridge_.run_scene("Downstairs", "Stream")
|
||||
|
||||
#self.bridge_.run_scene("Downstairs", "Stream")
|
||||
praxis_logger_obj.log("-[Lights_Module] Setup Complete")
|
||||
|
||||
def setLight():
|
||||
pass
|
||||
|
||||
def setLights():
|
||||
pass
|
||||
|
||||
def setGroup():
|
||||
pass
|
||||
|
||||
def setGroups():
|
||||
pass
|
||||
|
||||
def hydration(self):
|
||||
self.bridge_.run_scene("Downstairs", "hydration")
|
||||
#sleep(20)
|
||||
#self.bridge_.run_scene("Downstairs", "Stream")
|
||||
|
||||
def raveMode(self):
|
||||
for rave in range(30):
|
||||
rgb_r = random.random()
|
||||
rgb_g = random.random()
|
||||
rgb_b = random.random()
|
||||
xy_result = self.rgb_to_xy(rgb_r, rgb_g, rgb_b) #This will take an rgb value and make it xy
|
||||
self.bridge_.set_group(16, "xy", xy_result)
|
||||
sleep(0.3)
|
||||
self.bridge_.run_scene("Downstairs", "Stream")
|
||||
|
||||
def rgb_to_xy(self, red, green, blue):
|
||||
""" conversion of RGB colors to CIE1931 XY colors
|
||||
Formulas implemented from: https://gist.github.com/popcorn245/30afa0f98eea1c2fd34d
|
||||
Args:
|
||||
red (float): a number between 0.0 and 1.0 representing red in the RGB space
|
||||
green (float): a number between 0.0 and 1.0 representing green in the RGB space
|
||||
blue (float): a number between 0.0 and 1.0 representing blue in the RGB space
|
||||
Returns:
|
||||
xy (list): x and y
|
||||
"""
|
||||
# gamma correction
|
||||
red = pow((red + 0.055) / (1.0 + 0.055), 2.4) if red > 0.04045 else (red / 12.92)
|
||||
green = pow((green + 0.055) / (1.0 + 0.055), 2.4) if green > 0.04045 else (green / 12.92)
|
||||
blue = pow((blue + 0.055) / (1.0 + 0.055), 2.4) if blue > 0.04045 else (blue / 12.92)
|
||||
|
||||
# convert rgb to xyz
|
||||
x = red * 0.649926 + green * 0.103455 + blue * 0.197109
|
||||
y = red * 0.234327 + green * 0.743075 + blue * 0.022598
|
||||
z = green * 0.053077 + blue * 1.035763
|
||||
|
||||
# convert xyz to xy
|
||||
x = x / (x + y + z)
|
||||
y = y / (x + y + z)
|
||||
|
||||
# TODO check color gamut if known
|
||||
return [x, y]
|
||||
|
||||
def color_string_parser(self, message):
|
||||
maxDigits = config.colorParse_maxDigits
|
||||
praxis_logger_obj.log("Searching for color...")
|
||||
xy_color = [0, 0]
|
||||
for text in message:
|
||||
#praxis_logger_obj.log("testing word")
|
||||
if "red" in text.lower():
|
||||
xy_color = self.rgb_to_xy(1,0,0)
|
||||
praxis_logger_obj.log("-found: red")
|
||||
if "blue" in text.lower():
|
||||
praxis_logger_obj.log("-found: blue")
|
||||
xy_color = self.rgb_to_xy(0,0,1)
|
||||
if "green" in text.lower():
|
||||
praxis_logger_obj.log("-found: green")
|
||||
xy_color = self.rgb_to_xy(0,1,0)
|
||||
|
||||
if "yellow" in text.lower():
|
||||
praxis_logger_obj.log("-found: yellow")
|
||||
xy_color = self.rgb_to_xy(
|
||||
0.7,
|
||||
0.64,
|
||||
0)
|
||||
|
||||
|
||||
if "cyan" in text.lower():
|
||||
praxis_logger_obj.log("-found: cyan")
|
||||
xy_color = self.rgb_to_xy(0,1,1)
|
||||
if "aquamarine" in text.lower():
|
||||
praxis_logger_obj.log("-found: aquamarine")
|
||||
xy_color = self.rgb_to_xy(
|
||||
round(utilities.rescale_value(111,0,254),maxDigits),
|
||||
round(utilities.rescale_value(218,0,254),maxDigits),
|
||||
round(utilities.rescale_value(146,0,254),maxDigits))
|
||||
if "turquoise" in text.lower():
|
||||
praxis_logger_obj.log("-found: turquoise")
|
||||
xy_color = self.rgb_to_xy(
|
||||
round(utilities.rescale_value(172,0,254),maxDigits),
|
||||
round(utilities.rescale_value(233,0,254),maxDigits),
|
||||
round(utilities.rescale_value(232,0,254),maxDigits))
|
||||
|
||||
if "orange" in text.lower():
|
||||
praxis_logger_obj.log("-found: orange")
|
||||
xy_color = self.rgb_to_xy(
|
||||
1,
|
||||
round(utilities.rescale_value(126,0,254),maxDigits),
|
||||
0)
|
||||
|
||||
|
||||
if "magenta" in text.lower():
|
||||
praxis_logger_obj.log("-found: magenta")
|
||||
xy_color = self.rgb_to_xy(
|
||||
1,
|
||||
0,
|
||||
1)
|
||||
|
||||
if "purple" in text.lower():
|
||||
praxis_logger_obj.log("-found: purple")
|
||||
xy_color = self.rgb_to_xy(
|
||||
round(utilities.rescale_value(159,0,254),maxDigits),
|
||||
round(utilities.rescale_value(32,0,254),maxDigits),
|
||||
round(utilities.rescale_value(239,0,254),maxDigits))
|
||||
|
||||
if "violet" in text.lower():
|
||||
praxis_logger_obj.log("-found: violet")
|
||||
xy_color = self.rgb_to_xy(
|
||||
round(utilities.rescale_value(237,0,254),maxDigits),
|
||||
round(utilities.rescale_value(129,0,254),maxDigits),
|
||||
round(utilities.rescale_value(237,0,254),maxDigits))
|
||||
|
||||
return xy_color
|
||||
|
||||
|
||||
RGB_Lights = Lights_Module()
|
||||
|
||||
def init():
|
||||
RGB_Lights.main()
|
||||
|
||||
def do_lights_command(user="", lightGroup="all", command = "", rest = ""):
|
||||
returnString = "None"
|
||||
praxis_logger_obj.log("about to do something ......")
|
||||
praxis_logger_obj.log("about to do something with: " + command + " " + rest)
|
||||
#bot.return_message("\nRGB Command Detected!")
|
||||
if rest is not "":
|
||||
tempFix = command + " " + rest
|
||||
else:
|
||||
tempFix = command
|
||||
|
||||
tempParsedMessage = tempFix.split(" ")
|
||||
sceneCommand = False
|
||||
if (len(tempParsedMessage)) > 2:
|
||||
praxis_logger_obj.log("RGB Command!")
|
||||
rgb_r = float(tempParsedMessage[1])
|
||||
rgb_g = float(tempParsedMessage[2])
|
||||
rgb_b = float(tempParsedMessage[3])
|
||||
xy_result = RGB_Lights.rgb_to_xy(rgb_r, rgb_g, rgb_b)
|
||||
praxis_logger_obj.log("got XY")
|
||||
RGB_Lights.bridge_.set_group(16, "xy", xy_result)
|
||||
#bot.return_message("sent color to [Lights_Module]")
|
||||
else:
|
||||
if "stream" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
RGB_Lights.bridge_.run_scene("Downstairs", "Stream")
|
||||
elif "normal" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
RGB_Lights.bridge_.run_scene("Downstairs", "Bright")
|
||||
elif "haxor" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
RGB_Lights.bridge_.run_scene("Downstairs", "hacker vibes")
|
||||
elif "off" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
RGB_Lights.bridge_.set_group("Downstairs", "on", False)
|
||||
elif "on" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
RGB_Lights.bridge_.set_group("Downstairs", "on", True)
|
||||
elif "hydration" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
RGB_Lights.hydration()
|
||||
elif "ravemode" in tempParsedMessage:
|
||||
sceneCommand = True
|
||||
RGB_Lights.raveMode()
|
||||
else:
|
||||
#bot.return_message("Color Command!")
|
||||
xy_result = RGB_Lights.color_string_parser(tempParsedMessage)
|
||||
#bot.return_message("got XY")
|
||||
RGB_Lights.bridge_.set_group(16, "xy", xy_result)
|
||||
#bot.return_message("sent color to [Lights_Module]")
|
||||
|
||||
if sceneCommand == True:
|
||||
praxis_logger_obj.log("Scene Command!")
|
||||
|
||||
returnString = user + " changed the light's color!"
|
||||
|
||||
return flask.make_response("{\"message\":\"%s\"}" % returnString, 200, {"Content-Type": "application/json"})
|
||||
|
||||
|
||||
|
||||
@api.route('/api/v1/exec_lights', methods=['GET'])
|
||||
def exec_lights():
|
||||
if 'user_name' not in request.args:
|
||||
user_name="User"
|
||||
else:
|
||||
user_name=request.args['user_name']
|
||||
if 'light_group' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'light_group\' not in request"}', 400)
|
||||
if 'command' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'scene_name\' not in request"}', 400)
|
||||
|
||||
praxis_logger_obj.log("about to do something ......")
|
||||
RGB_Lights.main()
|
||||
return do_lights_command(user_name, request.args['light_group'], request.args['command'], request.args['rest'])
|
||||
|
||||
if __name__ == "__main__":
|
||||
init()
|
||||
api.run(host='0.0.0.0', port=42069)
|
||||
#testModule.raveMode()
|
||||
@ -20,6 +20,12 @@ from uuid import UUID
|
||||
|
||||
from cooldowns import Cooldown_Module
|
||||
|
||||
import os
|
||||
import praxis_logging
|
||||
praxis_logger_obj = praxis_logging.praxis_logger()
|
||||
praxis_logger_obj.init(os.path.basename(__file__))
|
||||
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
|
||||
|
||||
class Twitch_Pubsub():
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
@ -88,7 +94,13 @@ class Twitch_Pubsub():
|
||||
userinput = data['data']['redemption']['user_input']
|
||||
except:
|
||||
userinput = ""
|
||||
#print(userinput)
|
||||
praxis_logger_obj.log("\n\n")
|
||||
praxis_logger_obj.log(data['data']['redemption']['user']['display_name'])
|
||||
praxis_logger_obj.log(data['data']['redemption']['reward']['title'])
|
||||
praxis_logger_obj.log(AbstractChannelRewards.ChannelRewardsType.channelPoints)
|
||||
praxis_logger_obj.log(data['data']['redemption']['reward']['prompt'])
|
||||
praxis_logger_obj.log(userinput)
|
||||
praxis_logger_obj.log(data)
|
||||
self.callback_EXEC(
|
||||
data['data']['redemption']['user']['display_name'],
|
||||
data['data']['redemption']['reward']['title'],
|
||||
@ -111,7 +123,7 @@ class Twitch_Pubsub():
|
||||
try:
|
||||
is_actionable = self.is_reward(rewardName, rewardType)
|
||||
if is_actionable:
|
||||
print("Trying to do the thing")
|
||||
praxis_logger_obj.log("Trying to do the thing")
|
||||
if self.cooldownModule.isCooldownActive("twitchChat") == False:
|
||||
self.exec_reward(sender, rewardName, rewardType, rewardPrompt, userInput, raw_data)
|
||||
except:
|
||||
|
||||
@ -12,6 +12,12 @@ from cooldowns import Cooldown_Module
|
||||
import commands.command_base
|
||||
import utilities_script as utility
|
||||
|
||||
import os
|
||||
import praxis_logging
|
||||
praxis_logger_obj = praxis_logging.praxis_logger()
|
||||
praxis_logger_obj.init(os.path.basename(__file__))
|
||||
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
|
||||
|
||||
class Twitch_Module():
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
@ -96,6 +102,8 @@ class Twitch_Module():
|
||||
print("[#" + message.channel + "](" + message.sender + ")> " + message.text)
|
||||
command, rest = utility.parse_line(message.text)
|
||||
|
||||
praxis_logger_obj.log("\n[#" + message.channel + "](" + message.sender + ")> " + message.text)
|
||||
|
||||
try:
|
||||
is_actionable = self.is_command(command)
|
||||
if is_actionable:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user