mirror of
https://github.com/jayofelony/pwnagotchi.git
synced 2025-07-01 18:37:27 -04:00
Merge branch 'noai' of https://github.com/V0r-T3x/pwnagotchi into noai
This commit is contained in:
V0r-T3x
@ -9,7 +9,6 @@ _name = None
|
||||
config = None
|
||||
_cpu_stats = {}
|
||||
|
||||
|
||||
def set_name(new_name):
|
||||
if new_name is None:
|
||||
return
|
||||
|
||||
@ -2,6 +2,7 @@ import logging
|
||||
|
||||
import pwnagotchi.plugins as plugins
|
||||
from pwnagotchi.ai.epoch import Epoch
|
||||
import os
|
||||
|
||||
|
||||
# basic mood system
|
||||
@ -136,7 +137,12 @@ class Automata(object):
|
||||
self.set_grateful()
|
||||
|
||||
plugins.on('epoch', self, self._epoch.epoch - 1, self._epoch.data())
|
||||
|
||||
if self._epoch.blind_for % 10 == 2:
|
||||
logging.info("two blind epochs -> restarting wifi.recon...", self._epoch.blind_for)
|
||||
self.run('wifi.recon on')
|
||||
if self._epoch.blind_for and self._epoch.blind_for % 5 == 0:
|
||||
logging.info("%d epochs without visible access points -> restarting bettercap...", self._epoch.blind_for)
|
||||
os.system("systemctl restart bettercap")
|
||||
if self._epoch.blind_for >= self._config['main']['mon_max_blind_epochs']:
|
||||
logging.critical("%d epochs without visible access points -> restarting ...", self._epoch.blind_for)
|
||||
self._restart()
|
||||
|
||||
335
pwnagotchi/cli.py
Normal file
335
pwnagotchi/cli.py
Normal file
@ -0,0 +1,335 @@
|
||||
import logging
|
||||
import argparse
|
||||
import time
|
||||
import signal
|
||||
import sys
|
||||
import toml
|
||||
import requests
|
||||
import os
|
||||
import re
|
||||
|
||||
import pwnagotchi
|
||||
from pwnagotchi import utils
|
||||
from pwnagotchi.google import cmd as google_cmd
|
||||
from pwnagotchi.plugins import cmd as plugins_cmd
|
||||
from pwnagotchi import log
|
||||
from pwnagotchi import fs
|
||||
from pwnagotchi.utils import DottedTomlEncoder, parse_version as version_to_tuple
|
||||
|
||||
|
||||
def pwnagotchi_cli():
|
||||
def do_clear(display):
|
||||
logging.info("clearing the display ...")
|
||||
display.clear()
|
||||
sys.exit(0)
|
||||
|
||||
def do_manual_mode(agent):
|
||||
logging.info("entering manual mode ...")
|
||||
|
||||
agent.mode = 'manual'
|
||||
agent.last_session.parse(agent.view(), args.skip_session)
|
||||
if not args.skip_session:
|
||||
logging.info(
|
||||
"the last session lasted %s (%d completed epochs, trained for %d), average reward:%s (min:%s max:%s)" % (
|
||||
agent.last_session.duration_human,
|
||||
agent.last_session.epochs,
|
||||
agent.last_session.train_epochs,
|
||||
agent.last_session.avg_reward,
|
||||
agent.last_session.min_reward,
|
||||
agent.last_session.max_reward))
|
||||
|
||||
while True:
|
||||
display.on_manual_mode(agent.last_session)
|
||||
time.sleep(5)
|
||||
if grid.is_connected():
|
||||
plugins.on('internet_available', agent)
|
||||
|
||||
def do_auto_mode(agent):
|
||||
logging.info("entering auto mode ...")
|
||||
|
||||
agent.mode = 'auto'
|
||||
agent.start()
|
||||
|
||||
while True:
|
||||
try:
|
||||
# recon on all channels
|
||||
agent.recon()
|
||||
# get nearby access points grouped by channel
|
||||
channels = agent.get_access_points_by_channel()
|
||||
# for each channel
|
||||
for ch, aps in channels:
|
||||
time.sleep(1)
|
||||
agent.set_channel(ch)
|
||||
|
||||
if not agent.is_stale() and agent.any_activity():
|
||||
logging.info("%d access points on channel %d" % (len(aps), ch))
|
||||
|
||||
# for each ap on this channel
|
||||
for ap in aps:
|
||||
# send an association frame in order to get for a PMKID
|
||||
agent.associate(ap)
|
||||
# deauth all client stations in order to get a full handshake
|
||||
for sta in ap['clients']:
|
||||
agent.deauth(ap, sta)
|
||||
time.sleep(1) # delay to not trigger nexmon firmware bugs
|
||||
|
||||
# An interesting effect of this:
|
||||
#
|
||||
# From Pwnagotchi's perspective, the more new access points
|
||||
# and / or client stations nearby, the longer one epoch of
|
||||
# its relative time will take ... basically, in Pwnagotchi's universe,
|
||||
# Wi-Fi electromagnetic fields affect time like gravitational fields
|
||||
# affect ours ... neat ^_^
|
||||
agent.next_epoch()
|
||||
|
||||
if grid.is_connected():
|
||||
plugins.on('internet_available', agent)
|
||||
|
||||
except Exception as e:
|
||||
if str(e).find("wifi.interface not set") > 0:
|
||||
logging.exception("main loop exception due to unavailable wifi device, likely programmatically disabled (%s)", e)
|
||||
logging.info("sleeping 60 seconds then advancing to next epoch to allow for cleanup code to trigger")
|
||||
time.sleep(60)
|
||||
agent.next_epoch()
|
||||
else:
|
||||
logging.exception("main loop exception (%s)", e)
|
||||
|
||||
def add_parsers(parser):
|
||||
"""
|
||||
Adds the plugins and google subcommands
|
||||
"""
|
||||
subparsers = parser.add_subparsers()
|
||||
|
||||
# Add parsers from plugins_cmd
|
||||
plugins_cmd.add_parsers(subparsers)
|
||||
|
||||
# Add parsers from google_cmd
|
||||
google_cmd.add_parsers(subparsers)
|
||||
|
||||
parser = argparse.ArgumentParser(prog="pwnagotchi")
|
||||
# pwnagotchi --help
|
||||
parser.add_argument('-C', '--config', action='store', dest='config', default='/etc/pwnagotchi/default.toml',
|
||||
help='Main configuration file.')
|
||||
parser.add_argument('-U', '--user-config', action='store', dest='user_config', default='/etc/pwnagotchi/config.toml',
|
||||
help='If this file exists, configuration will be merged and this will override default values.')
|
||||
|
||||
parser.add_argument('--manual', dest="do_manual", action="store_true", default=False, help="Manual mode.")
|
||||
parser.add_argument('--skip-session', dest="skip_session", action="store_true", default=False,
|
||||
help="Skip last session parsing in manual mode.")
|
||||
|
||||
parser.add_argument('--clear', dest="do_clear", action="store_true", default=False,
|
||||
help="Clear the ePaper display and exit.")
|
||||
|
||||
parser.add_argument('--debug', dest="debug", action="store_true", default=False,
|
||||
help="Enable debug logs.")
|
||||
|
||||
parser.add_argument('--version', dest="version", action="store_true", default=False,
|
||||
help="Print the version.")
|
||||
|
||||
parser.add_argument('--print-config', dest="print_config", action="store_true", default=False,
|
||||
help="Print the configuration.")
|
||||
|
||||
# Jayofelony added these
|
||||
parser.add_argument('--wizard', dest="wizard", action="store_true", default=False,
|
||||
help="Interactive installation of your personal configuration.")
|
||||
parser.add_argument('--check-update', dest="check_update", action="store_true", default=False,
|
||||
help="Check for updates on Pwnagotchi. And tells current version.")
|
||||
parser.add_argument('--donate', dest="donate", action="store_true", default=False,
|
||||
help="How to donate to this project.")
|
||||
|
||||
# pwnagotchi plugins --help
|
||||
add_parsers(parser)
|
||||
args = parser.parse_args()
|
||||
|
||||
if plugins_cmd.used_plugin_cmd(args):
|
||||
config = utils.load_config(args)
|
||||
log.setup_logging(args, config)
|
||||
rc = plugins_cmd.handle_cmd(args, config)
|
||||
sys.exit(rc)
|
||||
if google_cmd.used_google_cmd(args):
|
||||
config = utils.load_config(args)
|
||||
log.setup_logging(args, config)
|
||||
rc = google_cmd.handle_cmd(args)
|
||||
sys.exit(rc)
|
||||
|
||||
if args.version:
|
||||
print(pwnagotchi.__version__)
|
||||
sys.exit(0)
|
||||
|
||||
if args.wizard:
|
||||
def is_valid_hostname(hostname):
|
||||
if len(hostname) > 255:
|
||||
return False
|
||||
if hostname[-1] == ".":
|
||||
hostname = hostname[:-1] # strip exactly one dot from the right, if present
|
||||
allowed = re.compile("(?!-)[A-Z\d-]{1,63}(?<!-)$", re.IGNORECASE)
|
||||
return all(allowed.match(x) for x in hostname.split("."))
|
||||
|
||||
pwn_restore = input("Do you want to restore the previous configuration?\n\n"
|
||||
"[Y/N]: ")
|
||||
if pwn_restore in ('y', 'yes'):
|
||||
os.system("cp -f /etc/pwnagotchi/config.toml.bak /etc/pwnagotchi/config.toml")
|
||||
print("Your previous configuration is restored, and I will restart in 5 seconds.")
|
||||
time.sleep(5)
|
||||
os.system("service pwnagotchi restart")
|
||||
else:
|
||||
pwn_check = input("This will create a new configuration file and overwrite your current backup, are you sure?\n\n"
|
||||
"[Y/N]: ")
|
||||
if pwn_check.lower() in ('y', 'yes'):
|
||||
os.system("mv -f /etc/pwnagotchi/config.toml /etc/pwnagotchi/config.toml.bak")
|
||||
with open("/etc/pwnagotchi/config.toml", "a+") as f:
|
||||
f.write("# Do not edit this file if you do not know what you are doing!!!\n\n")
|
||||
# Set pwnagotchi name
|
||||
print("Welcome to the interactive installation of your personal Pwnagotchi configuration!\n"
|
||||
"My name is Jayofelony, how may I call you?\n\n")
|
||||
pwn_name = input("Pwnagotchi name (no spaces): ")
|
||||
if pwn_name == "":
|
||||
pwn_name = "Pwnagotchi"
|
||||
print("I shall go by Pwnagotchi from now on!")
|
||||
pwn_name = f"main.name = \"{pwn_name}\"\n"
|
||||
f.write(pwn_name)
|
||||
else:
|
||||
if is_valid_hostname(pwn_name):
|
||||
print(f"I shall go by {pwn_name} from now on!")
|
||||
pwn_name = f"main.name = \"{pwn_name}\"\n"
|
||||
f.write(pwn_name)
|
||||
else:
|
||||
print("You have chosen an invalid name. Please start over.")
|
||||
exit()
|
||||
pwn_whitelist = input("How many networks do you want to whitelist? "
|
||||
"We will also ask a MAC for each network?\n"
|
||||
"Each SSID and BSSID count as 1 network. \n\n"
|
||||
"Be sure to use digits as your answer.\n\n"
|
||||
"Amount of networks: ")
|
||||
if int(pwn_whitelist) > 0:
|
||||
f.write("main.whitelist = [\n")
|
||||
for x in range(int(pwn_whitelist)):
|
||||
ssid = input("SSID (Name): ")
|
||||
bssid = input("BSSID (MAC): ")
|
||||
f.write(f"\t\"{ssid}\",\n")
|
||||
f.write(f"\t\"{bssid}\",\n")
|
||||
f.write("]\n")
|
||||
# set bluetooth tether
|
||||
pwn_bluetooth = input("Do you want to enable BT-Tether?\n\n"
|
||||
"[Y/N] ")
|
||||
if pwn_bluetooth.lower() in ('y', 'yes'):
|
||||
f.write("main.plugins.bt-tether.enabled = true\n\n")
|
||||
pwn_bluetooth_device = input("What device do you use? Android or iOS?\n\n"
|
||||
"Device: ")
|
||||
if pwn_bluetooth_device.lower() == "android":
|
||||
f.write("main.plugins.bt-tether.devices.android-phone.enabled = true\n")
|
||||
pwn_bluetooth_mac = input("What is the bluetooth MAC of your device?\n\n"
|
||||
"MAC: ")
|
||||
if pwn_bluetooth_mac != "":
|
||||
f.write(f"main.plugins.bt-tether.devices.android-phone.mac = \"{pwn_bluetooth_mac}\"\n")
|
||||
elif pwn_bluetooth_device.lower() == "ios":
|
||||
f.write("main.plugins.bt-tether.devices.ios-phone.enabled = true\n")
|
||||
pwn_bluetooth_mac = input("What is the bluetooth MAC of your device?\n\n"
|
||||
"MAC: ")
|
||||
if pwn_bluetooth_mac != "":
|
||||
f.write(f"main.plugins.bt-tether.devices.ios-phone.mac = \"{pwn_bluetooth_mac}\"\n")
|
||||
# set up display settings
|
||||
pwn_display_enabled = input("Do you want to enable a display?\n\n"
|
||||
"[Y/N]: ")
|
||||
if pwn_display_enabled.lower() in ('y', 'yes'):
|
||||
f.write("ui.display.enabled = true\n")
|
||||
pwn_display_type = input("What display do you use?\n\n"
|
||||
"Be sure to check for the correct display type @ \n"
|
||||
"https://github.com/jayofelony/pwnagotchi/blob/master/pwnagotchi/utils.py#L240-L501\n\n"
|
||||
"Display type: ")
|
||||
if pwn_display_type != "":
|
||||
f.write(f"ui.display.type = \"{pwn_display_type}\"\n")
|
||||
pwn_display_invert = input("Do you want to invert the display colors?\n"
|
||||
"N = Black background\n"
|
||||
"Y = White background\n\n"
|
||||
"[Y/N]: ")
|
||||
if pwn_display_invert.lower() in ('y', 'yes'):
|
||||
f.write("ui.invert = true\n")
|
||||
f.close()
|
||||
if pwn_bluetooth.lower() in ('y', 'yes'):
|
||||
if pwn_bluetooth_device.lower == "android":
|
||||
print("To visit the webui when connected with your phone, visit: http://192.168.44.44:8080\n"
|
||||
"Your configuration is done, and I will restart in 5 seconds.")
|
||||
elif pwn_bluetooth_device.lower == "ios":
|
||||
print("To visit the webui when connected with your phone, visit: http://172.20.10.6:8080\n"
|
||||
"Your configuration is done, and I will restart in 5 seconds.")
|
||||
else:
|
||||
print("Your configuration is done, and I will restart in 5 seconds.")
|
||||
time.sleep(5)
|
||||
os.system("service pwnagotchi restart")
|
||||
else:
|
||||
print("Ok, doing nothing.")
|
||||
sys.exit(0)
|
||||
|
||||
if args.donate:
|
||||
print("Donations can be made @ \n "
|
||||
"https://github.com/sponsors/jayofelony \n\n"
|
||||
"But only if you really want to!")
|
||||
sys.exit(0)
|
||||
|
||||
if args.check_update:
|
||||
resp = requests.get("https://api.github.com/repos/jayofelony/pwnagotchi/releases/latest")
|
||||
latest = resp.json()
|
||||
latest_ver = latest['tag_name'].replace('v', '')
|
||||
|
||||
local = version_to_tuple(pwnagotchi.__version__)
|
||||
remote = version_to_tuple(latest_ver)
|
||||
if remote > local:
|
||||
user_input = input("There is a new version available! Update from v%s to v%s?\n[Y/N] " % (pwnagotchi.__version__, latest_ver))
|
||||
# input validation
|
||||
if user_input.lower() in ('y', 'yes'):
|
||||
if os.path.exists('/root/.auto-update'):
|
||||
os.system("rm /root/.auto-update && systemctl restart pwnagotchi")
|
||||
else:
|
||||
logging.error("You should make sure auto-update is enabled!")
|
||||
print("Okay, give me a couple minutes. Just watch pwnlog while you wait.")
|
||||
elif user_input.lower() in ('n', 'no'): # using this elif for readability
|
||||
print("Okay, guess not!")
|
||||
else:
|
||||
print("You are currently on the latest release, v%s." % pwnagotchi.__version__)
|
||||
sys.exit(0)
|
||||
|
||||
config = utils.load_config(args)
|
||||
|
||||
if args.print_config:
|
||||
print(toml.dumps(config, encoder=DottedTomlEncoder()))
|
||||
sys.exit(0)
|
||||
|
||||
from pwnagotchi.identity import KeyPair
|
||||
from pwnagotchi.agent import Agent
|
||||
from pwnagotchi.ui import fonts
|
||||
from pwnagotchi.ui.display import Display
|
||||
from pwnagotchi import grid
|
||||
from pwnagotchi import plugins
|
||||
|
||||
pwnagotchi.config = config
|
||||
fs.setup_mounts(config)
|
||||
log.setup_logging(args, config)
|
||||
fonts.init(config)
|
||||
|
||||
pwnagotchi.set_name(config['main']['name'])
|
||||
|
||||
plugins.load(config)
|
||||
|
||||
display = Display(config=config, state={'name': '%s>' % pwnagotchi.name()})
|
||||
|
||||
if args.do_clear:
|
||||
do_clear(display)
|
||||
sys.exit(0)
|
||||
|
||||
agent = Agent(view=display, config=config, keypair=KeyPair(view=display))
|
||||
|
||||
def usr1_handler(*unused):
|
||||
logging.info('Received USR1 signal. Restart process ...')
|
||||
agent._restart("MANU" if args.do_manual else "AUTO")
|
||||
|
||||
signal.signal(signal.SIGUSR1, usr1_handler)
|
||||
|
||||
if args.do_manual:
|
||||
do_manual_mode(agent)
|
||||
else:
|
||||
do_auto_mode(agent)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
pwnagotchi_cli()
|
||||
Binary file not shown.
@ -174,7 +174,7 @@ class BTNap:
|
||||
"""
|
||||
Wait for device
|
||||
|
||||
returns device if found None if not
|
||||
returns device if found, None if not
|
||||
"""
|
||||
logging.debug("BT-TETHER: Waiting for device")
|
||||
|
||||
|
||||
233
pwnagotchi/plugins/default/gps_listener.py
Normal file
233
pwnagotchi/plugins/default/gps_listener.py
Normal file
@ -0,0 +1,233 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import threading
|
||||
|
||||
import pwnagotchi.plugins as plugins
|
||||
import pwnagotchi.ui.fonts as fonts
|
||||
from pwnagotchi.ui.components import LabeledValue
|
||||
from pwnagotchi.ui.view import BLACK
|
||||
|
||||
"""
|
||||
# Android
|
||||
# Termux:API : https://f-droid.org/en/packages/com.termux.api/
|
||||
# Termux : https://f-droid.org/en/packages/com.termux/
|
||||
pkg install termux-api socat bc
|
||||
|
||||
-----
|
||||
#!/data/data/com.termux/files/usr/bin/bash
|
||||
|
||||
# Server details
|
||||
SERVER_IP="192.168.44.45" # IP of the socat receiver
|
||||
SERVER_PORT="5000" # UDP port to send data to
|
||||
|
||||
# Function to calculate checksum
|
||||
calculate_checksum() {
|
||||
local sentence="$1"
|
||||
local checksum=0
|
||||
# Loop through each character in the sentence
|
||||
for ((i = 0; i < ${#sentence}; i++)); do
|
||||
checksum=$((checksum ^ $(printf '%d' "'${sentence:i:1}")))
|
||||
done
|
||||
# Return checksum in hexadecimal
|
||||
printf "%02X" $checksum
|
||||
}
|
||||
|
||||
# Infinite loop to send GPS data
|
||||
while true; do
|
||||
# Get location data
|
||||
LOCATION=$(termux-location -p gps)
|
||||
|
||||
# Extract latitude, longitude, altitude, speed, and bearing
|
||||
LATITUDE=$(echo "$LOCATION" | jq '.latitude')
|
||||
LONGITUDE=$(echo "$LOCATION" | jq '.longitude')
|
||||
ALTITUDE=$(echo "$LOCATION" | jq '.altitude')
|
||||
SPEED=$(echo "$LOCATION" | jq '.speed') # Speed in meters per second
|
||||
BEARING=$(echo "$LOCATION" | jq '.bearing')
|
||||
|
||||
# Convert speed from meters per second to knots and km/h
|
||||
SPEED_KNOTS=$(echo "$SPEED" | awk '{printf "%.1f", $1 * 1.943844}')
|
||||
SPEED_KMH=$(echo "$SPEED" | awk '{printf "%.1f", $1 * 3.6}')
|
||||
|
||||
# Format latitude and longitude for NMEA
|
||||
LAT_DEGREES=$(printf "%.0f" "${LATITUDE%.*}")
|
||||
LAT_MINUTES=$(echo "(${LATITUDE#${LAT_DEGREES}} * 60)" | bc -l)
|
||||
LAT_DIRECTION=$(if (( $(echo "$LATITUDE >= 0" | bc -l) )); then echo "N"; else echo "S"; fi)
|
||||
LON_DEGREES=$(printf "%.0f" "${LONGITUDE%.*}")
|
||||
LON_MINUTES=$(echo "(${LONGITUDE#${LON_DEGREES}} * 60)" | bc -l)
|
||||
LON_DIRECTION=$(if (( $(echo "$LONGITUDE >= 0" | bc -l) )); then echo "E"; else echo "W"; fi)
|
||||
|
||||
# Format the NMEA GGA sentence
|
||||
RAW_NMEA_GGA="GPGGA,123519,$(printf "%02d%07.4f" ${LAT_DEGREES#-} $LAT_MINUTES),$LAT_DIRECTION,$(printf "%03d%07.4f" ${LON_DEGREES#-} $LON_MINUTES),$LON_DIRECTION,1,08,0.9,$(printf "%.1f" $ALTITUDE),M,46.9,M,,"
|
||||
CHECKSUM=$(calculate_checksum "$RAW_NMEA_GGA")
|
||||
NMEA_GGA="\$${RAW_NMEA_GGA}*${CHECKSUM}"
|
||||
|
||||
# Format the VTG sentence
|
||||
RAW_NMEA_VTG="GPVTG,$(printf "%.1f" $BEARING),T,,M,$(printf "%.1f" $SPEED_KNOTS),N,$(printf "%.1f" $SPEED_KMH),K"
|
||||
CHECKSUM_VTG=$(calculate_checksum "$RAW_NMEA_VTG")
|
||||
NMEA_VTG="\$${RAW_NMEA_VTG}*${CHECKSUM_VTG}"
|
||||
|
||||
# Send data via UDP
|
||||
echo "$NMEA_GGA"
|
||||
echo "$NMEA_GGA" | socat - UDP:$SERVER_IP:$SERVER_PORT
|
||||
#echo "$NMEA_VTG"
|
||||
#echo "$NMEA_VTG" | socat - UDP:$SERVER_IP:$SERVER_PORT
|
||||
|
||||
sleep 1
|
||||
done
|
||||
-----
|
||||
|
||||
# Pwnagotchi
|
||||
main.plugins.gps_listener.enabled = true
|
||||
|
||||
# packages
|
||||
sudo apt-get install socat
|
||||
"""
|
||||
|
||||
class GPS(plugins.Plugin):
|
||||
__author__ = 'https://github.com/krishenriksen'
|
||||
__version__ = "1.0.0"
|
||||
__license__ = "GPL3"
|
||||
__description__ = "Receive GPS coordinates via termux-location and save whenever an handshake is captured."
|
||||
|
||||
def __init__(self):
|
||||
self.listen_ip = self.get_ip_address('bnep0')
|
||||
self.listen_port = "5000"
|
||||
self.write_virtual_serial = "/dev/ttyUSB1"
|
||||
self.read_virtual_serial = "/dev/ttyUSB0"
|
||||
self.baud_rate = "19200"
|
||||
self.socat_process = None
|
||||
self.stop_event = threading.Event()
|
||||
self.status_lock = threading.Lock()
|
||||
self.status = '-'
|
||||
self.socat_thread = threading.Thread(target=self.run_socat)
|
||||
|
||||
def get_ip_address(self, interface):
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["ip", "addr", "show", interface],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True
|
||||
)
|
||||
for line in result.stdout.split('\n'):
|
||||
if 'inet ' in line:
|
||||
ip_address = line.strip().split()[1].split('/')[0]
|
||||
return ip_address
|
||||
except subprocess.CalledProcessError:
|
||||
logging.warning(f"Could not get IP address for interface {interface}")
|
||||
return None
|
||||
|
||||
def set_status(self, status):
|
||||
with self.status_lock:
|
||||
self.status = status
|
||||
|
||||
def get_status(self):
|
||||
with self.status_lock:
|
||||
return self.status
|
||||
|
||||
def on_loaded(self):
|
||||
logging.info("GPS Listener plugin loaded")
|
||||
self.cleanup_virtual_serial_ports()
|
||||
self.create_virtual_serial_ports()
|
||||
self.socat_thread.start()
|
||||
|
||||
def cleanup_virtual_serial_ports(self):
|
||||
if os.path.exists(self.write_virtual_serial):
|
||||
self.log.info(f"Removing old {self.write_virtual_serial}")
|
||||
os.remove(self.write_virtual_serial)
|
||||
|
||||
if os.path.exists(self.read_virtual_serial):
|
||||
self.log.info(f"Removing old {self.read_virtual_serial}")
|
||||
os.remove(self.read_virtual_serial)
|
||||
|
||||
def create_virtual_serial_ports(self):
|
||||
self.socat_process = subprocess.Popen(
|
||||
["socat", "-d", "-d", f"pty,link={self.write_virtual_serial},mode=777",
|
||||
f"pty,link={self.read_virtual_serial},mode=777"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
|
||||
def run_socat(self):
|
||||
while not self.stop_event.is_set():
|
||||
self.socat_process = subprocess.Popen(
|
||||
["socat", f"UDP-RECVFROM:{self.listen_port},reuseaddr,bind={self.listen_ip}", "-"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True
|
||||
)
|
||||
|
||||
self.set_status('C')
|
||||
|
||||
with open(self.write_virtual_serial, 'w') as serial_port:
|
||||
for line in self.socat_process.stdout:
|
||||
if self.stop_event.is_set():
|
||||
break
|
||||
serial_port.write(line)
|
||||
serial_port.flush() # Ensure the data is written immediately
|
||||
self.status = 'C'
|
||||
|
||||
self.socat_process.wait()
|
||||
if self.stop_event.is_set():
|
||||
break
|
||||
|
||||
self.set_status('-')
|
||||
|
||||
def cleanup(self):
|
||||
if self.socat_process:
|
||||
self.socat_process.terminate()
|
||||
self.socat_process.wait() # Ensure the process is reaped
|
||||
self.stop_event.set()
|
||||
self.socat_thread.join()
|
||||
self.cleanup_virtual_serial_ports()
|
||||
|
||||
def on_ready(self, agent):
|
||||
if os.path.exists(self.read_virtual_serial):
|
||||
logging.info(
|
||||
f"enabling bettercap's gps module for {self.read_virtual_serial}"
|
||||
)
|
||||
try:
|
||||
agent.run("gps off")
|
||||
except Exception:
|
||||
logging.info(f"bettercap gps module was already off")
|
||||
pass
|
||||
|
||||
agent.run(f"set gps.device {self.read_virtual_serial}")
|
||||
agent.run(f"set gps.baudrate {self.baud_rate}")
|
||||
agent.run("gps on")
|
||||
|
||||
logging.info(f"bettercap gps module enabled on {self.read_virtual_serial}")
|
||||
else:
|
||||
self.set_status('NF')
|
||||
logging.warning("no GPS detected")
|
||||
|
||||
def on_handshake(self, agent, filename, access_point, client_station):
|
||||
info = agent.session()
|
||||
coordinates = info["gps"]
|
||||
gps_filename = filename.replace(".pcap", ".gps.json")
|
||||
|
||||
if coordinates and all([
|
||||
# avoid 0.000... measurements
|
||||
coordinates["Latitude"], coordinates["Longitude"]
|
||||
]):
|
||||
self.set_status('S')
|
||||
logging.info(f"saving GPS to {gps_filename} ({coordinates})")
|
||||
with open(gps_filename, "w+t") as fp:
|
||||
json.dump(coordinates, fp)
|
||||
else:
|
||||
logging.warning("not saving GPS. Couldn't find location.")
|
||||
|
||||
def on_ui_setup(self, ui):
|
||||
with ui._lock:
|
||||
ui.add_element('gps', LabeledValue(color=BLACK, label='GPS', value='-', position=(ui.width() / 2 - 40, 0), label_font=fonts.Bold, text_font=fonts.Medium))
|
||||
|
||||
def on_unload(self, ui):
|
||||
self.cleanup()
|
||||
|
||||
with ui._lock:
|
||||
ui.remove_element('gps')
|
||||
|
||||
def on_ui_update(self, ui):
|
||||
ui.set('gps', self.get_status())
|
||||
75
pwnagotchi/plugins/default/wittypi.py
Normal file
75
pwnagotchi/plugins/default/wittypi.py
Normal file
@ -0,0 +1,75 @@
|
||||
# Witty Pi 4 L3V7
|
||||
#
|
||||
import logging
|
||||
import pwnagotchi.plugins as plugins
|
||||
import pwnagotchi.ui.fonts as fonts
|
||||
from pwnagotchi.ui.components import LabeledValue
|
||||
from pwnagotchi.ui.view import BLACK
|
||||
|
||||
class UPS:
|
||||
I2C_MC_ADDRESS = 0x08
|
||||
I2C_VOLTAGE_IN_I = 1
|
||||
I2C_VOLTAGE_IN_D = 2
|
||||
I2C_CURRENT_OUT_I = 5
|
||||
I2C_CURRENT_OUT_D = 6
|
||||
I2C_POWER_MODE = 7
|
||||
|
||||
def __init__(self):
|
||||
# only import when the module is loaded and enabled
|
||||
import smbus
|
||||
# 0 = /dev/i2c-0 (port I2C0), 1 = /dev/i2c-1 (port I2C1)
|
||||
self._bus = smbus.SMBus(1)
|
||||
|
||||
def voltage(self):
|
||||
try:
|
||||
i = self._bus.read_byte_data(self.I2C_MC_ADDRESS, self.I2C_VOLTAGE_IN_I)
|
||||
d = self._bus.read_byte_data(self.I2C_MC_ADDRESS, self.I2C_VOLTAGE_IN_D)
|
||||
return (i + d / 100)
|
||||
except Exception as e:
|
||||
logging.info(f"register={i} failed (exception={e})")
|
||||
return 0.0
|
||||
|
||||
def current(self):
|
||||
try:
|
||||
i = self._bus.read_byte_data(self.I2C_MC_ADDRESS, self.I2C_CURRENT_OUT_I)
|
||||
d = self._bus.read_byte_data(self.I2C_MC_ADDRESS, self.I2C_CURRENT_OUT_D)
|
||||
return (i + d / 100)
|
||||
except Exception as e:
|
||||
logging.info(f"register={i} failed (exception={e})")
|
||||
return 0.0
|
||||
|
||||
def capacity(self):
|
||||
voltage = max(3.1, min(self.voltage(), 4.2)) # Clamp voltage
|
||||
return round((voltage - 3.1) / (4.2 - 3.1) * 100)
|
||||
|
||||
def charging(self):
|
||||
try:
|
||||
dc = self._bus.read_byte_data(self.I2C_MC_ADDRESS, self.I2C_POWER_MODE)
|
||||
return '+' if dc == 0 else '-'
|
||||
except:
|
||||
return '-'
|
||||
|
||||
class WittyPi(plugins.Plugin):
|
||||
__author__ = 'https://github.com/krishenriksen'
|
||||
__version__ = '1.0.0'
|
||||
__license__ = 'GPL3'
|
||||
__description__ = 'A plugin that will display battery info from Witty Pi 4 L3V7'
|
||||
|
||||
def __init__(self):
|
||||
self.ups = None
|
||||
|
||||
def on_loaded(self):
|
||||
self.ups = UPS()
|
||||
logging.info("wittypi plugin loaded.")
|
||||
|
||||
def on_ui_setup(self, ui):
|
||||
ui.add_element('ups', LabeledValue(color=BLACK, label='UPS', value='0%', position=(ui.width() / 2 + 15, 0), label_font=fonts.Bold, text_font=fonts.Medium))
|
||||
|
||||
def on_unload(self, ui):
|
||||
with ui._lock:
|
||||
ui.remove_element('ups')
|
||||
|
||||
def on_ui_update(self, ui):
|
||||
capacity = self.ups.capacity()
|
||||
charging = self.ups.charging()
|
||||
ui.set('ups', "%2i%s" % (capacity, charging))
|
||||
@ -1,12 +1,12 @@
|
||||
# board GPIO:
|
||||
# Key1:
|
||||
# Key2:
|
||||
# Key3:
|
||||
# Key4:
|
||||
#
|
||||
# Key1: GPIO 16
|
||||
# Key2: GPIO 20
|
||||
# Key3: GPIO 21
|
||||
# Key4: GPIO 26
|
||||
# IR: GPIO 23
|
||||
# Touch chipset:
|
||||
# HW info: https://argon40.com/products/pod-display-2-8inch
|
||||
# HW datasheet:
|
||||
# HW MANUAL: https://cdn.shopify.com/s/files/1/0556/1660/2177/files/ARGON_POD_MANUAL.pdf?v=1668390711%0A
|
||||
|
||||
import logging
|
||||
|
||||
@ -44,6 +44,9 @@ class ArgonPod(DisplayImpl):
|
||||
|
||||
def initialize(self):
|
||||
logging.info("Initializing Argon Pod display")
|
||||
logging.info("Available pins for GPIO Buttons: 16, 20, 21, 26")
|
||||
logging.info("IR available on GPIO 23")
|
||||
logging.info("Backlight pin available on GPIO 18")
|
||||
from pwnagotchi.ui.hw.libs.argon.argonpod.ILI9341 import ILI9341
|
||||
self._display = ILI9341(0, 0, 22, 18)
|
||||
|
||||
|
||||
@ -1,9 +1,7 @@
|
||||
import logging
|
||||
|
||||
import pwnagotchi.ui.fonts as fonts
|
||||
from pwnagotchi.ui.hw.base import DisplayImpl
|
||||
|
||||
|
||||
class DisplayHatMini(DisplayImpl):
|
||||
def __init__(self, config):
|
||||
super(DisplayHatMini, self).__init__(config, 'displayhatmini')
|
||||
@ -29,11 +27,14 @@ class DisplayHatMini(DisplayImpl):
|
||||
'font': fonts.status_font(fonts.Medium),
|
||||
'max': 20
|
||||
}
|
||||
|
||||
return self._layout
|
||||
|
||||
def initialize(self):
|
||||
logging.info("initializing Display Hat Mini")
|
||||
logging.info("Initializing Display Hat Mini")
|
||||
logging.info("Available pins for GPIO Buttons A/B/X/Y: 5, 6, 16, 24")
|
||||
logging.info("Available pins for RGB Led: 17, 27, 22")
|
||||
logging.info("Backlight pin available on GPIO 13")
|
||||
logging.info("I2C bus available on stemma QT and Breakout Garden headers")
|
||||
from pwnagotchi.ui.hw.libs.pimoroni.displayhatmini.ST7789 import ST7789
|
||||
self._display = ST7789(0,1,9,13)
|
||||
|
||||
|
||||
@ -8,7 +8,7 @@
|
||||
# ui.display.blcolor = "olive"
|
||||
#
|
||||
# Contrast should be between 30-50, default is 40
|
||||
# Backlight are predefined in the epd.py
|
||||
# Backlight are predefined in the lcd.py
|
||||
# Available backlight colors:
|
||||
# white, grey, maroon, red, purple, fuchsia, green,
|
||||
# lime, olive, yellow, navy, blue, teal, aqua
|
||||
@ -48,10 +48,16 @@ class GfxHat(DisplayImpl):
|
||||
def initialize(self):
|
||||
contrast = self._config['contrast'] if 'contrast' in self._config else 40
|
||||
blcolor = self._config['blcolor'] if 'blcolor' in self._config else 'OLIVE'
|
||||
logging.info("initializing Pimoroni GfxHat")
|
||||
logging.info("initializing Pimoroni GfxHat - Contrast: %d Backlight color: %s" % (contrast, blcolor))
|
||||
from pwnagotchi.ui.hw.libs.pimoroni.gfxhat.epd import EPD
|
||||
self._display = EPD(contrast=contrast)
|
||||
logging.info("Initializing Pimoroni GfxHat - Contrast: %d Backlight color: %s" % (contrast, blcolor))
|
||||
logging.info("Available config options: ui.display.contrast and ui.display.color")
|
||||
logging.info("Contrast should be between 30-50, default is 40")
|
||||
logging.info("Backlight are predefined in the lcd.py")
|
||||
logging.info("Available backlight colors:")
|
||||
logging.info("white, grey, maroon, red, purple, fuchsia, green,")
|
||||
logging.info("lime, olive, yellow, navy, blue, teal, aqua")
|
||||
logging.info("Touch control work in progress (6 touch buttons with short and long press and LED feedback)")
|
||||
from pwnagotchi.ui.hw.libs.pimoroni.gfxhat.lcd import LCD
|
||||
self._display = LCD(contrast=contrast)
|
||||
self._display.Init(color_name=blcolor)
|
||||
self._display.Clear()
|
||||
|
||||
|
||||
@ -52,11 +52,11 @@ class I2COled(DisplayImpl):
|
||||
i2caddr = self._config['i2c_addr'] if 'i2c_addr' in self._config else 0x3C
|
||||
width = self._config['width'] if 'width' in self._config else 128
|
||||
height = self._config['height'] if 'height' in self._config else 64
|
||||
logging.info("Initializing SSD1306 based %dx%d I2C Oled Display on address 0x%X" % (width, height, i2caddr))
|
||||
logging.info("Available config options: ui.display.width, ui.display.height and ui.display.i2caddr")
|
||||
|
||||
logging.info("initializing %dx%d I2C Oled Display on address 0x%X" % (width, height, i2caddr))
|
||||
|
||||
from pwnagotchi.ui.hw.libs.i2coled.epd import EPD
|
||||
self._display = EPD(address=i2caddr, width=width, height=height)
|
||||
from pwnagotchi.ui.hw.libs.i2coled.oled import OLED
|
||||
self._display = OLED(address=i2caddr, width=width, height=height)
|
||||
self._display.Init()
|
||||
self._display.Clear()
|
||||
|
||||
|
||||
@ -9,7 +9,7 @@ EPD_HEIGHT = 64
|
||||
# disp = SSD1306.SSD1306_96_16(96, 16, address=0x3C)
|
||||
# If you change for different resolution, you have to modify the layout in pwnagotchi/ui/hw/i2coled.py
|
||||
|
||||
class EPD(object):
|
||||
class OLED(object):
|
||||
|
||||
def __init__(self, address=0x3C, width=EPD_WIDTH, height=EPD_HEIGHT):
|
||||
self.width = width
|
||||
@ -9,7 +9,7 @@ LED_MAP = [2, 1, 0, 5, 4, 3]
|
||||
def setup():
|
||||
"""Set up the backlight on GFX HAT."""
|
||||
global _sn3218
|
||||
import sn3218 as _sn3218
|
||||
from . import sn3218 as _sn3218
|
||||
|
||||
_sn3218.enable()
|
||||
_sn3218.enable_leds(0b111111111111111111)
|
||||
|
||||
@ -1,55 +0,0 @@
|
||||
from . import st7567
|
||||
from . import backlight
|
||||
CONTRAST = 40
|
||||
|
||||
# Define RGB colors
|
||||
WHITE = (255, 255, 255)
|
||||
GREY = (255, 255, 255)
|
||||
MAROON = (128, 0, 0)
|
||||
RED = (255, 0, 0)
|
||||
PURPLE = (128, 0, 128)
|
||||
FUCHSIA = (255, 0, 255)
|
||||
GREEN = (0, 128, 0)
|
||||
LIME = (0, 255, 0)
|
||||
OLIVE = (128, 128, 0)
|
||||
YELLOW = (255, 255, 0)
|
||||
NAVY = (0, 0, 128)
|
||||
BLUE = (0, 0, 255)
|
||||
TEAL = (0, 128, 128)
|
||||
AQUA = (0, 255, 255)
|
||||
|
||||
# Map color names to RGB values
|
||||
color_map = {
|
||||
'WHITE': WHITE,
|
||||
'GREY' : GREY,
|
||||
'MAROON': MAROON,
|
||||
'RED': RED,
|
||||
'PURPLE': PURPLE,
|
||||
'FUCHSIA': FUCHSIA,
|
||||
'GREEN' : GREEN,
|
||||
'LIME' : LIME,
|
||||
'OLIVE' : OLIVE,
|
||||
'YELLOW' : YELLOW,
|
||||
'NAVY' : NAVY,
|
||||
'BLUE' : BLUE,
|
||||
'TEAL' : TEAL,
|
||||
'AQUA' : AQUA
|
||||
}
|
||||
|
||||
class EPD(object):
|
||||
|
||||
def __init__(self, contrast=CONTRAST, blcolor=('OLIVE')):
|
||||
self.disp = st7567.ST7567()
|
||||
self.disp.contrast(contrast)
|
||||
|
||||
def Init(self, color_name):
|
||||
self.disp.setup()
|
||||
blcolor = color_map.get(color_name.upper(), OLIVE) # Default to olive if color not found
|
||||
backlight.set_all(*blcolor)
|
||||
backlight.show()
|
||||
|
||||
def Clear(self):
|
||||
self.disp.clear()
|
||||
|
||||
def Display(self, image):
|
||||
self.disp.show(image)
|
||||
@ -1,57 +1,55 @@
|
||||
"""Library for the GFX HAT ST7567 SPI LCD."""
|
||||
from .st7567 import ST7567
|
||||
from . import st7567
|
||||
from . import backlight
|
||||
CONTRAST = 40
|
||||
|
||||
st7567 = ST7567()
|
||||
# Define RGB colors
|
||||
WHITE = (255, 255, 255)
|
||||
GREY = (255, 255, 255)
|
||||
MAROON = (128, 0, 0)
|
||||
RED = (255, 0, 0)
|
||||
PURPLE = (128, 0, 128)
|
||||
FUCHSIA = (255, 0, 255)
|
||||
GREEN = (0, 128, 0)
|
||||
LIME = (0, 255, 0)
|
||||
OLIVE = (128, 128, 0)
|
||||
YELLOW = (255, 255, 0)
|
||||
NAVY = (0, 0, 128)
|
||||
BLUE = (0, 0, 255)
|
||||
TEAL = (0, 128, 128)
|
||||
AQUA = (0, 255, 255)
|
||||
|
||||
dimensions = st7567.dimensions
|
||||
# Map color names to RGB values
|
||||
color_map = {
|
||||
'WHITE': WHITE,
|
||||
'GREY' : GREY,
|
||||
'MAROON': MAROON,
|
||||
'RED': RED,
|
||||
'PURPLE': PURPLE,
|
||||
'FUCHSIA': FUCHSIA,
|
||||
'GREEN' : GREEN,
|
||||
'LIME' : LIME,
|
||||
'OLIVE' : OLIVE,
|
||||
'YELLOW' : YELLOW,
|
||||
'NAVY' : NAVY,
|
||||
'BLUE' : BLUE,
|
||||
'TEAL' : TEAL,
|
||||
'AQUA' : AQUA
|
||||
}
|
||||
|
||||
class LCD(object):
|
||||
|
||||
def clear():
|
||||
"""Clear GFX HAT's display buffer."""
|
||||
st7567.clear()
|
||||
def __init__(self, contrast=CONTRAST, blcolor=('OLIVE')):
|
||||
self.disp = st7567.ST7567()
|
||||
self.disp.contrast(contrast)
|
||||
|
||||
def Init(self, color_name):
|
||||
self.disp.setup()
|
||||
blcolor = color_map.get(color_name.upper(), OLIVE) # Default to olive if color not found
|
||||
backlight.set_all(*blcolor)
|
||||
backlight.show()
|
||||
|
||||
def set_pixel(x, y, value):
|
||||
"""Set a single pixel in GTX HAT's display buffer.
|
||||
def Clear(self):
|
||||
self.disp.clear()
|
||||
|
||||
:param x: X position (from 0 to 127)
|
||||
:param y: Y position (from 0 to 63)
|
||||
:param value: pixel state 1 = On, 0 = Off
|
||||
|
||||
"""
|
||||
st7567.set_pixel(x, y, value)
|
||||
|
||||
|
||||
def show():
|
||||
"""Update GFX HAT with the current buffer contents."""
|
||||
st7567.show()
|
||||
|
||||
|
||||
def contrast(value):
|
||||
"""Change GFX HAT LCD contrast."""
|
||||
st7567.contrast(value)
|
||||
|
||||
|
||||
def rotation(r=0):
|
||||
"""Set the display rotation.
|
||||
|
||||
:param r: Specify the rotation in degrees: 0, or 180
|
||||
|
||||
"""
|
||||
if r == 0:
|
||||
st7567.rotated = False
|
||||
|
||||
elif r == 180:
|
||||
st7567.rotated = True
|
||||
|
||||
else:
|
||||
raise ValueError('Rotation must be 0 or 180 degrees')
|
||||
|
||||
|
||||
def get_rotation():
|
||||
"""Get the display rotation value.
|
||||
|
||||
Returns an integer, either 0, or 180
|
||||
|
||||
"""
|
||||
return 180 if st7567.rotated else 0
|
||||
def Display(self, image):
|
||||
self.disp.show(image)
|
||||
214
pwnagotchi/ui/hw/libs/pimoroni/gfxhat/sn3218.py
Normal file
214
pwnagotchi/ui/hw/libs/pimoroni/gfxhat/sn3218.py
Normal file
@ -0,0 +1,214 @@
|
||||
import sys
|
||||
import warnings
|
||||
|
||||
__version__ = '1.2.7'
|
||||
|
||||
I2C_ADDRESS = 0x54
|
||||
CMD_ENABLE_OUTPUT = 0x00
|
||||
CMD_SET_PWM_VALUES = 0x01
|
||||
CMD_ENABLE_LEDS = 0x13
|
||||
CMD_UPDATE = 0x16
|
||||
CMD_RESET = 0x17
|
||||
|
||||
if sys.version_info < (3, ):
|
||||
SMBUS = "python-smbus"
|
||||
else:
|
||||
SMBUS = "python3-smbus"
|
||||
|
||||
# Helper function to avoid exception chaining output in python3.
|
||||
# use exec to shield newer syntax from older python
|
||||
if sys.version_info < (3, 3):
|
||||
def _raise_from_none(exc):
|
||||
raise exc
|
||||
else:
|
||||
exec("def _raise_from_none(exc): raise exc from None")
|
||||
|
||||
try:
|
||||
from smbus import SMBus
|
||||
except ImportError:
|
||||
err_string = "This library requires {smbus}\nInstall with: sudo apt install {smbus}".format(smbus=SMBUS)
|
||||
import_error = ImportError(err_string)
|
||||
_raise_from_none(import_error)
|
||||
|
||||
def i2c_bus_id():
|
||||
"""
|
||||
Returns the i2c bus ID.
|
||||
"""
|
||||
with open('/proc/cpuinfo') as cpuinfo:
|
||||
revision = [l[12:-1] for l in cpuinfo if l[:8] == "Revision"][0]
|
||||
# https://www.raspberrypi.org/documentation/hardware/raspberrypi/revision-codes/README.md
|
||||
return 1 if int(revision, 16) >= 4 else 0
|
||||
|
||||
|
||||
def enable():
|
||||
"""
|
||||
Enables output.
|
||||
"""
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_ENABLE_OUTPUT, [0x01])
|
||||
|
||||
|
||||
def disable():
|
||||
"""
|
||||
Disables output.
|
||||
"""
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_ENABLE_OUTPUT, [0x00])
|
||||
|
||||
|
||||
def reset():
|
||||
"""
|
||||
Resets all internal registers.
|
||||
"""
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_RESET, [0xFF])
|
||||
|
||||
|
||||
def enable_leds(enable_mask):
|
||||
"""
|
||||
Enables or disables each LED channel. The first 18 bit values are
|
||||
used to determine the state of each channel (1=on, 0=off) if fewer
|
||||
than 18 bits are provided the remaining channels are turned off.
|
||||
|
||||
Args:
|
||||
enable_mask (int): up to 18 bits of data
|
||||
Raises:
|
||||
TypeError: if enable_mask is not an integer.
|
||||
"""
|
||||
if not isinstance(enable_mask, int):
|
||||
raise TypeError("enable_mask must be an integer")
|
||||
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_ENABLE_LEDS,
|
||||
[enable_mask & 0x3F, (enable_mask >> 6) & 0x3F, (enable_mask >> 12) & 0X3F])
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_UPDATE, [0xFF])
|
||||
|
||||
|
||||
def channel_gamma(channel, gamma_table):
|
||||
"""
|
||||
Overrides the gamma table for a single channel.
|
||||
|
||||
Args:
|
||||
channel (int): channel number
|
||||
gamma_table (list): list of 256 gamma correction values
|
||||
Raises:
|
||||
TypeError: if channel is not an integer.
|
||||
ValueError: if channel is not in the range 0..17.
|
||||
TypeError: if gamma_table is not a list.
|
||||
"""
|
||||
global channel_gamma_table
|
||||
|
||||
if not isinstance(channel, int):
|
||||
raise TypeError("channel must be an integer")
|
||||
|
||||
if channel not in range(18):
|
||||
raise ValueError("channel be an integer in the range 0..17")
|
||||
|
||||
if not isinstance(gamma_table, list) or len(gamma_table) != 256:
|
||||
raise TypeError("gamma_table must be a list of 256 integers")
|
||||
|
||||
channel_gamma_table[channel] = gamma_table
|
||||
|
||||
|
||||
def output(values):
|
||||
"""
|
||||
Outputs a new set of values to the driver.
|
||||
|
||||
Args:
|
||||
values (list): channel number
|
||||
Raises:
|
||||
TypeError: if values is not a list of 18 integers.
|
||||
"""
|
||||
if not isinstance(values, list) or len(values) != 18:
|
||||
raise TypeError("values must be a list of 18 integers")
|
||||
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_SET_PWM_VALUES, [channel_gamma_table[i][values[i]] for i in range(18)])
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_UPDATE, [0xFF])
|
||||
|
||||
|
||||
def output_raw(values):
|
||||
"""
|
||||
Outputs a new set of values to the driver.
|
||||
Similar to output(), but does not use channel_gamma_table.
|
||||
|
||||
Args:
|
||||
values (list): channel number
|
||||
Raises:
|
||||
TypeError: if values is not a list of 18 integers.
|
||||
"""
|
||||
# SMBus.write_i2c_block_data does the type check, so we don't have to
|
||||
if len(values) != 18:
|
||||
raise TypeError("values must be a list of 18 integers")
|
||||
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_SET_PWM_VALUES, values)
|
||||
i2c.write_i2c_block_data(I2C_ADDRESS, CMD_UPDATE, [0xFF])
|
||||
|
||||
|
||||
try:
|
||||
i2c = SMBus(i2c_bus_id())
|
||||
except IOError as e:
|
||||
warntxt="""
|
||||
###### ###### ###### ###### ###### ###### ###### ######
|
||||
i2c initialization failed - is i2c enabled on this system?
|
||||
See https://github.com/pimoroni/sn3218/wiki/missing-i2c
|
||||
###### ###### ###### ###### ###### ###### ###### ######
|
||||
"""
|
||||
warnings.warn(warntxt)
|
||||
raise(e)
|
||||
|
||||
# generate a good default gamma table
|
||||
default_gamma_table = [int(pow(255, float(i - 1) / 255)) for i in range(256)]
|
||||
channel_gamma_table = [default_gamma_table] * 18
|
||||
|
||||
enable_leds(0b111111111111111111)
|
||||
|
||||
def test_cycles():
|
||||
print("sn3218 test cycles")
|
||||
|
||||
import time
|
||||
import math
|
||||
|
||||
# enable output
|
||||
enable()
|
||||
enable_leds(0b111111111111111111)
|
||||
|
||||
print(">> test enable mask (on/off)")
|
||||
enable_mask = 0b000000000000000000
|
||||
output([0x10] * 18)
|
||||
for i in range(10):
|
||||
enable_mask = ~enable_mask
|
||||
enable_leds(enable_mask)
|
||||
time.sleep(0.15)
|
||||
|
||||
print(">> test enable mask (odd/even)")
|
||||
enable_mask = 0b101010101010101010
|
||||
output([0x10] * 18)
|
||||
for i in range(10):
|
||||
enable_mask = ~enable_mask
|
||||
enable_leds(enable_mask)
|
||||
time.sleep(0.15)
|
||||
|
||||
print(">> test enable mask (rotate)")
|
||||
enable_mask = 0b100000100000100000
|
||||
output([0x10] * 18)
|
||||
for i in range(10):
|
||||
enable_mask = ((enable_mask & 0x01) << 18) | enable_mask >> 1
|
||||
enable_leds(enable_mask)
|
||||
time.sleep(0.15)
|
||||
|
||||
print(">> test gamma gradient")
|
||||
enable_mask = 0b111111111111111111
|
||||
enable_leds(enable_mask)
|
||||
for i in range(256):
|
||||
output([((j * (256//18)) + (i * (256//18))) % 256 for j in range(18)])
|
||||
time.sleep(0.01)
|
||||
|
||||
print(">> test gamma fade")
|
||||
enable_mask = 0b111111111111111111
|
||||
enable_leds(enable_mask)
|
||||
for i in range(512):
|
||||
output([int((math.sin(float(i)/64.0) + 1.0) * 128.0)]*18)
|
||||
time.sleep(0.01)
|
||||
|
||||
# turn everything off and disable output
|
||||
output([0 for i in range(18)])
|
||||
disable()
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_cycles()
|
||||
@ -1,6 +1,6 @@
|
||||
# board GPIO:
|
||||
# A: GPIO22
|
||||
# B: GPIO23
|
||||
# A: GPIO23
|
||||
# B: GPIO24
|
||||
#
|
||||
# HW datasheet: https://learn.adafruit.com/adafruit-1-3-color-tft-bonnet-for-raspberry-pi/overview
|
||||
|
||||
@ -38,7 +38,10 @@ class MiniPitft(DisplayImpl):
|
||||
return self._layout
|
||||
|
||||
def initialize(self):
|
||||
logging.info("initializing Adafruit Mini Pi Tft 240x240")
|
||||
logging.info("Initializing Adafruit Mini Pi Tft 240x240")
|
||||
logging.info("Available pins for GPIO Buttons: 23, 24")
|
||||
logging.info("Backlight pin available on GPIO 22")
|
||||
logging.info("I2C bus available on stemma QT header")
|
||||
from pwnagotchi.ui.hw.libs.adafruit.minipitft.ST7789 import ST7789
|
||||
self._display = ST7789(0,0,25,22)
|
||||
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
# board GPIO:
|
||||
# A: GPIO22
|
||||
# B: GPIO23
|
||||
# A: GPIO23
|
||||
# B: GPIO24
|
||||
#
|
||||
# HW datasheet: https://learn.adafruit.com/adafruit-1-3-color-tft-bonnet-for-raspberry-pi/overview
|
||||
|
||||
@ -39,6 +39,9 @@ class MiniPitft2(DisplayImpl):
|
||||
|
||||
def initialize(self):
|
||||
logging.info("initializing Adafruit Mini Pi Tft 135x240")
|
||||
logging.info("Available pins for GPIO Buttons: 23, 24")
|
||||
logging.info("Backlight pin available on GPIO 22")
|
||||
logging.info("I2C bus available on stemma QT header")
|
||||
from pwnagotchi.ui.hw.libs.adafruit.minipitft2.ST7789 import ST7789
|
||||
self._display = ST7789(0,0,25,22)
|
||||
|
||||
|
||||
@ -45,6 +45,10 @@ class PirateAudio(DisplayImpl):
|
||||
|
||||
def initialize(self):
|
||||
logging.info("Initializing PirateAudio - display only")
|
||||
logging.info("Available pins for GPIO Buttons A/B/X/Y: 5, 6, 16, 20 or 24")
|
||||
logging.info("refer to the pimoroni site or pinout.xyz")
|
||||
logging.info("Backlight pin available on GPIO 13")
|
||||
logging.info("I2S for the DAC available on pins: 18, 19 and 21")
|
||||
from pwnagotchi.ui.hw.libs.pimoroni.pirateaudio.ST7789 import ST7789
|
||||
self._display = ST7789(0,1,9,13)
|
||||
|
||||
|
||||
@ -50,6 +50,9 @@ class Pitft(DisplayImpl):
|
||||
|
||||
def initialize(self):
|
||||
logging.info("Initializing adafruit pitft 320x240 screen")
|
||||
logging.info("Available pins for GPIO Buttons on the 3,2inch: 17, 22, 23, 27")
|
||||
logging.info("Available pins for GPIO Buttons on the 2,8inch: 26, 13, 12, 6, 5")
|
||||
logging.info("Backlight pin available on GPIO 18")
|
||||
from pwnagotchi.ui.hw.libs.adafruit.pitft.ILI9341 import ILI9341
|
||||
self._display = ILI9341(0, 0, 25, 18)
|
||||
|
||||
|
||||
@ -44,6 +44,9 @@ class TftBonnet(DisplayImpl):
|
||||
|
||||
def initialize(self):
|
||||
logging.info("initializing Adafruit Tft Bonnet")
|
||||
logging.info("Available pins for GPIO Buttons Up/Down/Left/Right/Center/A/B: 17, 22, 27, 23, 4, 5, 6")
|
||||
logging.info("Backlight pin available on GPIO 26")
|
||||
logging.info("I2C bus available on stemma QT header")
|
||||
from pwnagotchi.ui.hw.libs.adafruit.tftbonnet.ST7789 import ST7789
|
||||
self._display = ST7789(0,0,25,26)
|
||||
|
||||
|
||||
@ -49,6 +49,7 @@ class Waveshareoledlcd(DisplayImpl):
|
||||
|
||||
def initialize(self):
|
||||
logging.info("initializing Waveshare OLED/LCD hat")
|
||||
logging.info("Available pins for GPIO Buttons K1/K2/K3/K4: 4, 17, 23, 24")
|
||||
from pwnagotchi.ui.hw.libs.waveshare.oled.oledlcd.ST7789 import ST7789
|
||||
self._display = ST7789(0,0,22,18)
|
||||
|
||||
|
||||
@ -49,6 +49,7 @@ class Waveshareoledlcdvert(DisplayImpl):
|
||||
|
||||
def initialize(self):
|
||||
logging.info("initializing Waveshare OLED/LCD hat vertical mode")
|
||||
logging.info("Available pins for GPIO Buttons K1/K2/K3/K4: 4, 17, 23, 24")
|
||||
from pwnagotchi.ui.hw.libs.waveshare.oled.oledlcd.ST7789vert import ST7789
|
||||
self._display = ST7789(0,0,22,18)
|
||||
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
#import _thread
|
||||
# import _thread
|
||||
import threading
|
||||
import logging
|
||||
import random
|
||||
@ -6,7 +6,6 @@ import time
|
||||
from threading import Lock
|
||||
|
||||
from PIL import ImageDraw
|
||||
from PIL import ImageColor as colors
|
||||
|
||||
import pwnagotchi
|
||||
import pwnagotchi.plugins as plugins
|
||||
@ -19,105 +18,28 @@ from pwnagotchi.ui.components import *
|
||||
from pwnagotchi.ui.state import State
|
||||
from pwnagotchi.voice import Voice
|
||||
|
||||
WHITE = 0x00 # white is actually black on jays image
|
||||
BLACK = 0xFF # black is actually white on jays image
|
||||
|
||||
BACKGROUND_1 = 0
|
||||
FOREGROUND_1 = 1
|
||||
|
||||
BACKGROUND_L = 0
|
||||
FOREGROUND_L = 255
|
||||
|
||||
BACKGROUND_BGR_16 = (0,0,0)
|
||||
FOREGROUND_BGR_16 = (31,63,31)
|
||||
|
||||
BACKGROUND_RGB = (0,0,0)
|
||||
FOREGROUND_RGB = (255,255,255)
|
||||
|
||||
|
||||
ROOT = None
|
||||
|
||||
|
||||
|
||||
|
||||
#1 (1-bit pixels, black and white, stored with one pixel per byte)
|
||||
|
||||
#L (8-bit pixels, grayscale)
|
||||
|
||||
#P (8-bit pixels, mapped to any other mode using a color palette)
|
||||
|
||||
#BGR;16 (5,6,5 bits, for 65k color)
|
||||
|
||||
#RGB (3x8-bit pixels, true color)
|
||||
|
||||
#RGBA (4x8-bit pixels, true color with transparency mask)
|
||||
|
||||
#CMYK (4x8-bit pixels, color separation)
|
||||
|
||||
#YCbCr (3x8-bit pixels, color video format)
|
||||
|
||||
#self.FOREGROUND is the main color
|
||||
#self.BACKGROUNDGROUND is the 2ndary color, used for background
|
||||
|
||||
WHITE = 0x00 # white is actually black on jays image
|
||||
BLACK = 0xFF # black is actually white on jays image
|
||||
|
||||
|
||||
class View(object):
|
||||
def __init__(self, config, impl, state=None):
|
||||
global ROOT, BLACK, WHITE
|
||||
|
||||
#values/code for display color mode
|
||||
|
||||
self.mode = '1' # 1 = (1-bit pixels, black and white, stored with one pixel per byte)
|
||||
if hasattr(impl, 'mode'):
|
||||
self.mode = impl.mode
|
||||
|
||||
|
||||
|
||||
match self.mode:
|
||||
case '1':
|
||||
self.BACKGROUND = BACKGROUND_1
|
||||
self.FOREGROUND = FOREGROUND_1
|
||||
# do stuff is color mode is 1 when View object is created.
|
||||
case 'L':
|
||||
self.BACKGROUND = BACKGROUND_L # black 0 to 255
|
||||
self.FOREGROUND = FOREGROUND_L
|
||||
# do stuff is color mode is L when View object is created.
|
||||
case 'P':
|
||||
pass
|
||||
# do stuff is color mode is P when View object is created.
|
||||
case 'BGR;16':
|
||||
self.BACKGROUND = BACKGROUND_BGR_16 #black tuple
|
||||
self.FOREGROUND = FOREGROUND_BGR_16 #white tuple
|
||||
case 'RGB':
|
||||
self.BACKGROUND = BACKGROUND_RGB #black tuple
|
||||
self.FOREGROUND = FOREGROUND_RGB #white tuple
|
||||
# do stuff is color mode is RGB when View object is created.
|
||||
case 'RGBA':
|
||||
# do stuff is color mode is RGBA when View object is created.
|
||||
pass
|
||||
case 'CMYK':
|
||||
# do stuff is color mode is CMYK when View object is created.
|
||||
pass
|
||||
case 'YCbCr':
|
||||
# do stuff is color mode is YCbCr when View object is created.
|
||||
pass
|
||||
case _:
|
||||
# do stuff when color mode doesnt exist for display
|
||||
self.BACKGROUND = BACKGROUND_1
|
||||
self.FOREGROUND = FOREGROUND_1
|
||||
|
||||
|
||||
|
||||
self.invert = 0
|
||||
self._black = 0xFF
|
||||
self._white = 0x00
|
||||
if 'invert' in config['ui'] and config['ui']['invert'] == True:
|
||||
logging.debug("INVERT:" + str(config['ui']['invert']))
|
||||
logging.debug("INVERT BLACK/WHITES:" + str(config['ui']['invert']))
|
||||
self.invert = 1
|
||||
tmp = self.FOREGROUND
|
||||
self.FOREGROUND = self.FOREGROUND
|
||||
self.FOREGROUND = tmp
|
||||
BLACK = 0x00
|
||||
WHITE = 0xFF
|
||||
self._black = 0x00
|
||||
self._white = 0xFF
|
||||
|
||||
# setup faces from the configuration in case the user customized them
|
||||
faces.load_from_config(config['ui']['faces'])
|
||||
|
||||
|
||||
self._agent = None
|
||||
self._render_cbs = []
|
||||
self._config = config
|
||||
@ -130,40 +52,42 @@ class View(object):
|
||||
self._width = self._layout['width']
|
||||
self._height = self._layout['height']
|
||||
self._state = State(state={
|
||||
'channel': LabeledValue(color=self.FOREGROUND, label='CH', value='00', position=self._layout['channel'],
|
||||
'channel': LabeledValue(color=BLACK, label='CH', value='00', position=self._layout['channel'],
|
||||
label_font=fonts.Bold,
|
||||
text_font=fonts.Medium),
|
||||
'aps': LabeledValue(color=self.FOREGROUND, label='APS', value='0 (00)', position=self._layout['aps'],
|
||||
'aps': LabeledValue(color=BLACK, label='APS', value='0 (00)', position=self._layout['aps'],
|
||||
label_font=fonts.Bold,
|
||||
text_font=fonts.Medium),
|
||||
|
||||
'uptime': LabeledValue(color=self.FOREGROUND, label='UP', value='00:00:00', position=self._layout['uptime'],
|
||||
'uptime': LabeledValue(color=BLACK, label='UP', value='00:00:00', position=self._layout['uptime'],
|
||||
label_font=fonts.Bold,
|
||||
text_font=fonts.Medium),
|
||||
|
||||
'line1': Line(self._layout['line1'], color=self.FOREGROUND),
|
||||
'line2': Line(self._layout['line2'], color=self.FOREGROUND),
|
||||
'line1': Line(self._layout['line1'], color=BLACK),
|
||||
'line2': Line(self._layout['line2'], color=BLACK),
|
||||
|
||||
'face': Text(value=faces.SLEEP, position=(config['ui']['faces']['position_x'], config['ui']['faces']['position_y']), color=self.FOREGROUND, font=fonts.Huge, png=config['ui']['faces']['png']),
|
||||
'face': Text(value=faces.SLEEP,
|
||||
position=(config['ui']['faces']['position_x'], config['ui']['faces']['position_y']),
|
||||
color=BLACK, font=fonts.Huge, png=config['ui']['faces']['png']),
|
||||
|
||||
# 'friend_face': Text(value=None, position=self._layout['friend_face'], font=fonts.Bold, color=self.FOREGROUND),
|
||||
'friend_name': Text(value=None, position=self._layout['friend_face'], font=fonts.BoldSmall, color=self.FOREGROUND),
|
||||
# 'friend_face': Text(value=None, position=self._layout['friend_face'], font=fonts.Bold, color=BLACK),
|
||||
'friend_name': Text(value=None, position=self._layout['friend_face'], font=fonts.BoldSmall, color=BLACK),
|
||||
|
||||
'name': Text(value='%s>' % 'pwnagotchi', position=self._layout['name'], color=self.FOREGROUND, font=fonts.Bold),
|
||||
'name': Text(value='%s>' % 'pwnagotchi', position=self._layout['name'], color=BLACK, font=fonts.Bold),
|
||||
|
||||
'status': Text(value=self._voice.default(),
|
||||
position=self._layout['status']['pos'],
|
||||
color=self.FOREGROUND,
|
||||
color=BLACK,
|
||||
font=self._layout['status']['font'],
|
||||
wrap=True,
|
||||
# the current maximum number of characters per line, assuming each character is 6 pixels wide
|
||||
max_length=self._layout['status']['max']),
|
||||
|
||||
'shakes': LabeledValue(label='PWND ', value='0 (00)', color=self.FOREGROUND,
|
||||
'shakes': LabeledValue(label='PWND ', value='0 (00)', color=BLACK,
|
||||
position=self._layout['shakes'], label_font=fonts.Bold,
|
||||
text_font=fonts.Medium),
|
||||
'mode': Text(value='AUTO', position=self._layout['mode'],
|
||||
font=fonts.Bold, color=self.FOREGROUND),
|
||||
font=fonts.Bold, color=BLACK),
|
||||
})
|
||||
|
||||
if state:
|
||||
@ -173,8 +97,8 @@ class View(object):
|
||||
plugins.on('ui_setup', self)
|
||||
|
||||
if config['ui']['fps'] > 0.0:
|
||||
threading.Thread(target=self._refresh_handler, args=(), name="UI Handler", daemon = True).start()
|
||||
|
||||
threading.Thread(target=self._refresh_handler, args=(), name="UI Handler", daemon=True).start()
|
||||
|
||||
self._ignore_changes = ()
|
||||
else:
|
||||
logging.warning("ui.fps is 0, the display will only update for major changes")
|
||||
@ -189,7 +113,7 @@ class View(object):
|
||||
self._state.has_element(key)
|
||||
|
||||
def add_element(self, key, elem):
|
||||
if self.invert is 1 and hasattr(elem, 'color'):
|
||||
if self.invert is 1 and elem.color:
|
||||
if elem.color == 0xff:
|
||||
elem.color = 0x00
|
||||
elif elem.color == 0x00:
|
||||
@ -250,7 +174,8 @@ class View(object):
|
||||
self.set('uptime', last_session.duration)
|
||||
self.set('channel', '-')
|
||||
self.set('aps', "%d" % last_session.associated)
|
||||
self.set('shakes', '%d (%s)' % (last_session.handshakes, utils.total_unique_handshakes(self._config['bettercap']['handshakes'])))
|
||||
self.set('shakes', '%d (%s)' % (
|
||||
last_session.handshakes, utils.total_unique_handshakes(self._config['bettercap']['handshakes'])))
|
||||
self.set_closest_peer(last_session.last_peer, last_session.peers)
|
||||
self.update()
|
||||
|
||||
@ -340,7 +265,7 @@ class View(object):
|
||||
|
||||
def wait(self, secs, sleeping=True):
|
||||
was_normal = self.is_normal()
|
||||
part = secs/10.0
|
||||
part = secs / 10.0
|
||||
|
||||
for step in range(0, 10):
|
||||
# if we weren't in a normal state before going
|
||||
@ -468,13 +393,13 @@ class View(object):
|
||||
state = self._state
|
||||
changes = state.changes(ignore=self._ignore_changes)
|
||||
if force or len(changes):
|
||||
self._canvas = Image.new(self.mode, (self._width, self._height), self.BACKGROUND)
|
||||
drawer = ImageDraw.Draw(self._canvas, self.mode)
|
||||
self._canvas = Image.new('1', (self._width, self._height), self._white)
|
||||
drawer = ImageDraw.Draw(self._canvas)
|
||||
|
||||
plugins.on('ui_update', self)
|
||||
|
||||
for key, lv in state.items():
|
||||
#lv is a ui element
|
||||
# lv is a ui element
|
||||
lv.draw(self._canvas, drawer)
|
||||
|
||||
web.update_frame(self._canvas)
|
||||
|
||||
Reference in New Issue
Block a user