2022-02-02 07:01:23 +00:00
#!/usr/bin/python3
2022-02-04 15:34:21 +00:00
# Rewrite
2022-02-02 07:01:23 +00:00
import subprocess
2022-02-04 16:02:27 +00:00
import json
2022-02-02 07:01:23 +00:00
import time
import threading
2022-02-04 15:34:21 +00:00
import random
from pathlib import Path
from shutil import which
APP_UUID = " a6db95c2-27db-4b88-a687-c8107d1bc9d6 "
def Popen ( args : list | str , root = False , shell = False , cwd = Path . cwd ( ) ) :
if root :
if shell :
args = " sudo " + args
else :
args = [ " sudo " ] + args
result = subprocess . Popen ( args , shell = shell , cwd = cwd , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
return result
def run ( args : list | str , root = False , shell = False , cwd = Path . cwd ( ) ) :
if root :
if shell :
args = " sudo " + args
else :
args = [ " sudo " ] + args
result = subprocess . run ( args , shell = shell , cwd = cwd , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
result . check_returncode ( )
return result
def show_toast ( message ) :
run ( [ " termux-toast " , f ' { message } ' ] )
def show_notification ( title , message = None , vibration : list [ int ] | str = None , sound = False , ongoing = True ) :
if not which ( " termux-notification " ) :
raise FileNotFoundError ( " termux-notification not found, please install termux-api package and Termux:API app. " )
args = [ " termux-notification " , " -i " , APP_UUID , " --priority " , " max " , " -t " , f ' { title } ' ]
if message :
args + = [ " -c " , f ' { message } ' ]
if vibration :
args + = [ " --vibrate " , " , " . join ( str ( x ) for x in vibration ) if isinstance ( vibration , list ) else vibration ]
if sound :
args + = [ " --sound " ]
if ongoing :
args + = [ " --ongoing " ]
run ( args )
2022-02-04 16:02:27 +00:00
def get_wifi ( ) :
if not which ( " termux-wifi-connectioninfo " ) :
raise FileNotFoundError ( " termux-wifi-connectioninfo not found, please install termux-api package and Termux:API app. " )
return json . loads ( subprocess . check_output ( [ " termux-wifi-connectioninfo " ] ) . decode ( " utf-8 " ) . strip ( ) )
def get_wifi_name ( ) :
return get_wifi ( ) [ " ssid " ]
2022-02-04 15:34:21 +00:00
def get_device_private_ip ( ) :
2022-02-04 16:02:27 +00:00
return get_wifi ( ) [ " ip " ]
2022-02-04 15:34:21 +00:00
def get_port_from_process_name ( name : str , state : str = " LISTEN " ) :
lsof_output = subprocess . check_output ( [ " sudo " , " lsof " , " -i " , " -P " , " -n " ] ) . decode ( " utf-8 " )
for line in lsof_output . split ( " \n " ) :
if name in line and f " ( { state } ) " in line :
line = ' ' . join ( line . split ( ) )
return int ( line . strip ( ) . split ( " " ) [ 8 ] . split ( " : " ) [ 1 ] )
return None
def get_pid_from_port ( port : int , state : str = " LISTEN " ) :
lsof_output = subprocess . check_output ( [ " sudo " , " lsof " , " -i " , " -P " , " -n " ] ) . decode ( " utf-8 " )
for line in lsof_output . split ( " \n " ) :
if str ( port ) in line and f " ( { state } ) " in line :
line = ' ' . join ( line . split ( ) )
return line . strip ( ) . split ( " " ) [ 1 ]
return None
def start_adbd ( ) :
free_port = random . randint ( 0 , 65535 )
while get_pid_from_port ( free_port ) :
free_port = random . randint ( 0 , 65535 )
print ( " Starting adbd on port " + str ( free_port ) )
show_notification ( " Starting adbd " , " Starting adbd on port " + str ( free_port ) )
try :
run ( [ " setprop " , " service.adb.tcp.port " , str ( free_port ) ] , root = True )
except subprocess . CalledProcessError :
print ( " Failed to set adb port " )
show_notification ( " Failed to set adb port to " + str ( free_port ) , " You ' ll need to start adb manually through Developer Settings " )
return
try :
run ( [ " stop " , " adbd " ] , root = True )
except subprocess . CalledProcessError :
print ( " Failed to stop adbd, maybe adbd was not running? " )
try :
run ( [ " start " , " adbd " ] , root = True )
except subprocess . CalledProcessError :
print ( " Failed to start adbd " )
show_notification ( " Failed to start adbd " , " You ' ll need to start adb manually through Developer Settings " )
return
return free_port
def connect_adb ( ip , port ) :
run ( [ " adb " , " connect " , f " { ip } : { port } " ] )
2022-02-02 07:01:23 +00:00
def main ( ) :
2022-02-04 15:34:21 +00:00
print ( " ws-scrcpy launcher for Termux " )
print ( " Checking for ws-scrcpy... " )
if not Path ( " ./ws-scrcpy " ) . exists ( ) :
print ( " ws-scrcpy not found, please install ws-scrcpy. " )
return
print ( " Checking for current adb port... " )
show_notification ( " Checking for current adb port... " )
2022-02-02 07:01:23 +00:00
adb_port = None
try :
2022-02-04 15:34:21 +00:00
adb_port = get_port_from_process_name ( " adbd " )
2022-02-02 07:01:23 +00:00
except :
pass
2022-02-04 15:34:21 +00:00
if adb_port is None :
print ( " Failed to get adb port, starting new adbd... " )
adb_port = start_adbd ( )
if adb_port is None :
print ( " Failed to start adbd... " )
return
device_ip = get_device_private_ip ( )
print ( " Adb server port: " , adb_port )
print ( " Device IP: " , device_ip )
print ( " Connecting to device through adb... " )
show_notification ( " Connecting to device through adb... " )
try :
connect_adb ( device_ip , adb_port )
except subprocess . CalledProcessError :
print ( " Failed to connect to device through adb " )
show_notification ( " Failed to connect to device through adb " , " You ' ll need to connect manually in Termux. " )
return
print ( " Starting ws-scrcpy server... " )
show_notification ( " Starting ws-scrcpy server... " , " You may need to wait for 5 minutes for ws-scrcpy to start. " )
ws_scrcpy = Popen ( [ " npm " , " start " ] , cwd = " ./ws-scrcpy " )
2022-02-02 07:01:23 +00:00
def print_ws_scrcpy ( ) :
for line in ws_scrcpy . stdout :
2022-02-04 15:34:21 +00:00
if line . decode ( " utf-8 " ) . strip ( ) == None :
2022-02-02 07:01:23 +00:00
continue
if " Listening on: " in line . decode ( " utf-8 " ) . strip ( ) :
2022-02-04 15:34:21 +00:00
device_ip = get_device_private_ip ( )
2022-02-02 07:01:23 +00:00
print ( " ========================================= " )
2022-02-04 15:34:21 +00:00
print ( f " ws-scrcpy started on { device_ip } :8000 " )
2022-02-02 07:01:23 +00:00
print ( " ========================================= " )
2022-02-04 15:34:21 +00:00
show_toast ( f " ws-scrcpy: { device_ip } :8000 " )
2022-02-04 16:02:27 +00:00
show_notification ( f " ws-scrcpy: { device_ip } :8000 " , f " Wifi: { get_wifi_name ( ) } , access { device_ip } :8000 in browser to control the device. " , [ 2000 , 1000 , 500 ] , True )
2022-02-02 07:01:23 +00:00
print ( " [ws-scrcpy]: " , line . decode ( " utf-8 " ) . strip ( ) )
ws_scrcpy_thread = threading . Thread ( target = print_ws_scrcpy )
ws_scrcpy_thread . daemon = True
ws_scrcpy_thread . start ( )
2022-02-04 16:02:27 +00:00
print ( " PLEASE WAIT UNTIL WS-SCRCPY FULLY STARTED (ABOUT 5 MINS), IT TAKES A WHILE TO START THE SERVER. " )
2022-02-02 07:01:23 +00:00
try :
while True :
time . sleep ( 5 )
2022-02-04 15:34:21 +00:00
curr_ip = get_device_private_ip ( )
2022-02-02 07:01:23 +00:00
if curr_ip != device_ip :
2022-02-04 15:34:21 +00:00
print ( " Device IP changed, reconnecting adb... " )
show_notification ( " Reconnecting adb... " , " This may take a while, take a cup of coffee. " )
try :
connect_adb ( curr_ip , adb_port )
except subprocess . CalledProcessError :
print ( " Failed to connect to device through adb " )
show_notification ( " Failed to connect to device through adb " , " ws-scrcpy will be stopped. " )
2022-02-02 07:01:23 +00:00
break
2022-02-04 15:34:21 +00:00
device_ip = curr_ip
print ( " ========================================= " )
print ( f " ws-scrcpy started on { device_ip } :8000 " )
print ( " ========================================= " )
show_toast ( f " ws-scrcpy: { device_ip } :8000 " )
2022-02-04 16:02:27 +00:00
show_notification ( f " ws-scrcpy: { device_ip } :8000 " , f " Wifi: { get_wifi_name ( ) } , access { device_ip } :8000 in browser to control the device. " , [ 2000 , 1000 , 500 ] , True )
2022-02-04 15:34:21 +00:00
except :
pass
if ws_scrcpy . poll ( ) == None :
print ( " Stopping ws-scrcpy server... " )
show_notification ( " Stopping ws-scrcpy server... " )
ws_scrcpy . terminate ( )
try :
# if this returns, the process completed
ws_scrcpy . wait ( timeout = 15 )
except subprocess . TimeoutExpired :
print ( " ws_scrcpy doesn ' t exit after 15 seconds, killing process... " )
ws_scrcpy . kill ( )
2022-02-04 16:02:27 +00:00
2022-02-04 15:34:21 +00:00
print ( " ws-scrcpy has been stopped. " )
show_notification ( " ws-scrcpy has been stopped. " , ongoing = False )
2022-02-02 07:01:23 +00:00
if __name__ == ' __main__ ' :
main ( )