We are looking at using a small PC instead of an RPi to connect the driver’s cameras so I was looking at the Rpi cameraserver code.
on the PC:
python v3.11.0 installed
pip install robotpy[cscore]
python .\multiCameraServer.py frc.json
Using robot simulation on the PC for initial testing, I start our robot sim using VSCode using real-drivers station, the wait for the Robot Simulation panel to come up. Next, I start the cameraServer in client mode with the python command above.
Camera light comes on, and I see the client connection in the Sim panel’s NetworkTable Info:
Here’s the code I’m using. There was no CameraServer.setSize(), so that was commented out.
Thanks for the quick reply.
#!/usr/bin/env python3
# Copyright (c) FIRST and other WPILib contributors.
# Open Source Software; you can modify and/or share it under the terms of
# the WPILib BSD license file in the root directory of this project.
import json
import os
import time
import sys
from cscore import CameraServer, VideoSource, UsbCamera, MjpegServer
from ntcore import NetworkTableInstance, EventFlags
##JSON template removed for brevity
configFile = "frc.json"
class CameraConfig:
pass
team:int = 0
server = False
cameraConfigs = []
switchedCameraConfigs = []
cameras = []
def parseError(str):
"""Report parse error."""
print("config error in '" + configFile + "': " + str, file=sys.stderr)
def readCameraConfig(config):
"""Read single camera configuration."""
cam = CameraConfig()
# name
try:
cam.name = config["name"]
except KeyError:
parseError("could not read camera name")
return False
# path
try:
cam.path = config["path"]
except KeyError:
parseError("camera '{}': could not read path".format(cam.name))
return False
# stream properties
cam.streamConfig = config.get("stream")
cam.config = config
cameraConfigs.append(cam)
return True
def readSwitchedCameraConfig(config):
"""Read single switched camera configuration."""
cam = CameraConfig()
# name
try:
cam.name = config["name"]
except KeyError:
parseError("could not read switched camera name")
return False
# path
try:
cam.key = config["key"]
except KeyError:
parseError("switched camera '{}': could not read key".format(cam.name))
return False
switchedCameraConfigs.append(cam)
return True
def readConfig():
"""Read configuration file."""
global team
global server
#global configFile
# parse file
try:
print(os.getcwd())
dir_path = os.path.dirname(os.path.realpath(__file__))
print(dir_path)
#configFile = os.path.join(dir_path, configFile)
with open(configFile, "rt", encoding="utf-8") as f:
j = json.load(f)
except OSError as err:
print("could not open '{}': {}".format(configFile, err), file=sys.stderr)
return False
# top level must be an object
if not isinstance(j, dict):
parseError("must be JSON object")
return False
# team number
try:
team = j["team"]
except KeyError:
parseError("could not read team number")
return False
# ntmode (optional)
if "ntmode" in j:
str = j["ntmode"]
if str.lower() == "client":
server = False
elif str.lower() == "server":
server = True
else:
parseError("could not understand ntmode value '{}'".format(str))
# cameras
try:
cameras = j["cameras"]
except KeyError:
parseError("could not read cameras")
return False
for camera in cameras:
if not readCameraConfig(camera):
return False
# switched cameras
if "switched cameras" in j:
for camera in j["switched cameras"]:
if not readSwitchedCameraConfig(camera):
return False
return True
def startCamera(config):
"""Start running the camera."""
print("Starting camera '{}' on {}".format(config.name, config.path))
camera = UsbCamera(config.name, config.path)
server = CameraServer.startAutomaticCapture(camera=camera)
camera.setConfigJson(json.dumps(config.config))
camera.setConnectionStrategy(VideoSource.ConnectionStrategy.kConnectionKeepOpen)
if config.streamConfig is not None:
server.setConfigJson(json.dumps(config.streamConfig))
return camera
def startSwitchedCamera(config):
"""Start running the switched camera."""
print("Starting switched camera '{}' on {}".format(config.name, config.key))
server = CameraServer.addSwitchedCamera(config.name)
def listener(event):
data = event.data
if data is not None:
value = data.value.value()
if isinstance(value, int):
if value >= 0 and value < len(cameras):
server.setSource(cameras[value])
elif isinstance(value, float):
i = int(value)
if i >= 0 and i < len(cameras):
server.setSource(cameras[i])
elif isinstance(value, str):
for i in range(len(cameraConfigs)):
if value == cameraConfigs[i].name:
server.setSource(cameras[i])
break
NetworkTableInstance.getDefault().addListener(
NetworkTableInstance.getDefault().getEntry(config.key),
EventFlags.kImmediate | EventFlags.kValueAll,
listener)
return server
if __name__ == "__main__":
print("working directory:" + os.getcwd())
if len(sys.argv) >= 2:
configFile = sys.argv[1]
#List cameras
mycameras = UsbCamera.enumerateUsbCameras()
for c in mycameras:
print("name:", c.name,"path:", c.path)
# read configuration
if not readConfig():
sys.exit(1)
# start NetworkTables
ntinst = NetworkTableInstance.getDefault()
if server:
print("Setting up NetworkTables server")
ntinst.startServer()
else:
print("Setting up NetworkTables client for team {}".format(team))
ntinst.startClient4("wpilibpi")
ntinst.setServerTeam(team)
ntinst.startDSClient()
# start cameras
# work around wpilibsuite/allwpilib#5055
###dpl api change?
###CameraServer.setSize(CameraServer.kSize160x120)
for config in cameraConfigs:
cameras.append(startCamera(config))
# start switched cameras
for config in switchedCameraConfigs:
startSwitchedCamera(config)
# loop forever
while True:
time.sleep(10)
On a hunch, I made a small change to startCamera(config):
#server = CameraServer.startAutomaticCapture(camera=camera)
server = CameraServer.startAutomaticCapture(config.path)
where config.path = 0 for my camera. This seems to start the camera stream and now shows in Shuffleboard & Smart Dashboard.
The commented line does not start the stream.
I get one error on the output of the python script, but it continues to stream:
Setting up NetworkTables client for team 2202
Starting camera 'UVC Camera' on 0
CS: UVC Camera: Attempting to connect to USB camera on \\?\usb#vid_fefe&pid_4321&mi_00#b&1577f843&1&0000#{e5323777-f976-4f5b-9b55-b94699c46e44}\global
CS: USB Camera 0: Attempting to connect to USB camera on \\?\usb#vid_fefe&pid_4321&mi_00#b&1577f843&1&0000#{e5323777-f976-4f5b-9b55-b94699c46e44}\global
CS: UVC Camera: Connected to USB camera on \\?\usb#vid_fefe&pid_4321&mi_00#b&1577f843&1&0000#{e5323777-f976-4f5b-9b55-b94699c46e44}\global
CS: USB Camera 0: Connected to USB camera on \\?\usb#vid_fefe&pid_4321&mi_00#b&1577f843&1&0000#{e5323777-f976-4f5b-9b55-b94699c46e44}\global
CS: UVC Camera: SetConfigJson: setting video mode to pixelFormat 1, width 320, height 240, fps 30
Source Reader error: 0xC00D3EA3
I’m not following your request on “… tried removeCamera()”, which location?
Ok, I added the removeCamera call after the config loop and keeping orignal code. That seems to work.
def startCamera(config):
"""Start running the camera."""
print("Starting camera '{}' on {}".format(config.name, config.path))
camera = UsbCamera(config.name, config.path)
server = CameraServer.startAutomaticCapture(camera=camera)
###server = CameraServer.startAutomaticCapture(config.path)
camera.setConfigJson(json.dumps(config.config))
camera.setConnectionStrategy(VideoSource.ConnectionStrategy.kConnectionKeepOpen)
if config.streamConfig is not None:
server.setConfigJson(json.dumps(config.streamConfig))
return camera
Call to remove at end:
# start cameras
# work around wpilibsuite/allwpilib#5055
###CameraServer.setSize(CameraServer.kSize160x120)
for config in cameraConfigs:
cameras.append(startCamera(config))
CameraServer.removeCamera("unused")
This seems to work and removes the “Source Reader error”.
