after increasing the max_magnitude, the obs is identified.
max_magnitude = 5
max_magnitude = 6
.fits data:
datafits.zip (4.0 MB)
here i’m comparing max_magnitude from 5 to 15. between 6 to 15 with success (green) calibration, the image look identic, glitch only on magnitude 5 where no success (red) calibration
clear sky - 2026-07-14 (time in UTC)
my ip cam: TAPO C325WB (original firmware, default 20fps)
this my stvid configuration:
configuration.ini.share2satno.txt (4.3 KB)
[Observer]
cospar = 9999 # COSPAR number of observer's site
name = BALI # Name of observer
latitude = -8.736 # Latitude of location (decimal degrees)
longitude = 115.195 # Longitude of location (decimal degrees)
height = 10 # Height of location (meters)
[Setup]
camera_type = CV2 # CV2 or ASI or PI
tracking_mount = False
alt_sunset = -6.0 # Solar altitude at sunset
alt_sunrise = -6.0 # Solar altitude at sunrise
observations_path = /home/linux/obs
[Credentials]
st-username = x # Space-track.org username
st-password = x # Space-track.org password
[ASI]
device_id = 0 # Device ID
nx = 968 # Camera horizontal pixel count
ny = 608 # Camera vertical pixel count
nframes = 100 # Number of frames for each image
gain = 300 # Camera gain
maxgain = 300 # Maximum gain (for autogain; starts at gain)
autogain = True # Use autogain
exposure = 100000 # Exposure time in microseconds
brightness = 10 # Target brightness of pixels
bandwidth = 90 # Bandwidth usage (percent)
high_speed = 0 # High speed mode (0 for False, 1 for True)
hardware_bin = 0 # Hardware binning (0 for False, 1 for True)
bin = 1 # ASI software binning factor (1: 1x2, 2: 2x2 etc)
software_bin = 2 # STVID software binning factor (1: 1x1, 2: 2x2 etc)
sdk = /path/to/libASICamera2.so # path to the SDK library
[PI]
device_id = 0 # Device ID
nx = 800 # Camera horizontal pixel count
ny = 600 # Camera vertical pixel count
nframes = 100 # Number of frames for each image
framerate = 5 # Take 5 frames per second
exposure = 200000 # Exposure time in us
awb_gain_red = 2 # Gain for red.
awb_gain_blue = 2.3 # Gain for blue
analog_gain = 16 # Analog gain
digital_gain = 64 # Digital gain
# my ip cam: tapo c325wb
[CV2]
device_id = 0 # Device ID
device_string = rtsp://x:x@x.x.x.x:554/stream1
# stream1 resolution 2560 x 1440. after software_bin = 2, this become half = 1280 x 720
# then crop 1280 - 480 = 800 and 720 - 120 = 600
software_crop_width = 480
software_crop_height = 120
#result after cropped
nx = 800
ny = 600
nframes = 400
software_bin = 2 # STVID software binning factor (1: 1x1, 2: 2x2 etc) (bikin file output jadi lebih kecil)
# C0 blue
# C1 orange
# C2 green
# C3 red
# C4 violet
# C5 brown
# C6 pink
# C7 gray
# C8 yellow
# C9 blue light
# C10 blue
# C11 orange, repeat ...
[Elements]
tlepath = /home/linux/tle
tlefile4 = satno.tle
name4 = Satnogs
abbrev4 = satnogs
color4 = C6
tlefile5 = starlink.tle
name5 = Starlink
abbrev5 = starlink
color5 = C4
tlefile1 = catalog.tle # TLE File
name1 = Catalog # Catalog name
abbrev1 = catalog # Catalog abbreviation
color1 = C0 # Color code
tlefile2 = classfd.tle
name2 = Classified
abbrev2 = classfd
color2 = C1
tlefile3 = inttles.tle
name3 = Integrated
abbrev3 = inttles
color3 = C2
[DiagnosticPlot]
colormap = gray_r
[LineDetection]
min_sigma = 3.2 # Minimum track selection significance, kalau banyak noise dinaikkan angkanya. default 5
min_track_points = 25 # 10 Minimum number of points making up a track
min_track_width = 10 # 7 Minimum track selection width (pixels)
color = C3 # Detected track color code
cpu_count = 2 # CPU threads to use
[Identification]
max_off_track_offset_deg = 0.1 # Maximum off-track offset [angle; deg]
max_along_track_offset_s = 2.0 # Maximum off-track offset [time; sec]
max_direction_difference_deg = 2.0 # Maximum difference between direction of motion [angle; deg]
max_velocity_difference_percent = 5.0 # Maximum velocity difference [percent]
[Astrometry]
min_stars = 25 #ok : 20 Minimum number of stars to attempt plate solve
max_magnitude = 13 # between 12 - 13 still there bending. 13-15 same result. maybe because the star detected is same
solve-field_args = -O -l 10 -z 2 -t 1 --crpix-center # Solve-field arguments
#[Shutter]
#pin = 19 # GPIO pin controlling shutter
#[Aimpoint] # Stop/start when aimpoint is in/out of shadow
#az_deg = 180 # Aimpoint azimuth (deg)
#alt_deg = 60 # Aimpoint altitude (deg)
#height_km = 2000 # Aimpoint orbital height (km)
.fits example from obs and tle used:
tle.zip (2.3 MB)
2026-07-14T20-55-59.725.fits.zip (3.5 MB)
i modified original @cgbsat acquire.py, so now support cropping. get here:
acquire-modif-crop.py.txt (29.5 KB)
#!/usr/bin/env python3
import sys
import os
import numpy as np
import cv2
import time
import ctypes
import multiprocessing
from astropy.coordinates import EarthLocation
from astropy.time import Time
from astropy.io import fits
import astropy.units as u
from stvid.utils import observe_logic
import logging
import configparser
import argparse
def setup_logging(path):
logFormatter = logging.Formatter(
"%(asctime)s [%(processName)-12.12s] [%(levelname)-5.5s] %(message)s"
)
logger = logging.getLogger()
logger.handlers.clear()
logger.setLevel(logging.DEBUG)
fileHandler = logging.FileHandler(os.path.join(path, "acquire.log"))
fileHandler.setFormatter(logFormatter)
logger.addHandler(fileHandler)
consoleHandler = logging.StreamHandler(sys.stdout)
consoleHandler.setFormatter(logFormatter)
logger.addHandler(consoleHandler)
return logger
# Capture images from pi
def capture_pi(image_queue, z1base, t1base, z2base, t2base, nx, ny, nz, tend, device_id, live, conf_file):
global logger
logger = setup_logging(os.getcwd())
cfg = configparser.ConfigParser(inline_comment_prefixes=("#", ";"))
cfg.read(conf_file)
from picamerax.array import PiRGBArray
from picamerax import PiCamera
z1 = np.ctypeslib.as_array(z1base.get_obj()).reshape(ny, nx, nz)
t1 = np.ctypeslib.as_array(t1base.get_obj())
z2 = np.ctypeslib.as_array(z2base.get_obj()).reshape(ny, nx, nz)
t2 = np.ctypeslib.as_array(t2base.get_obj())
# Intialization
first = True
slow_CPU = False
# Initialize cv2 device
camera = PiCamera(sensor_mode=2)
camera.resolution = (nx, ny)
# Turn off any thing automatic.
camera.exposure_mode = "off"
camera.awb_mode = "off"
# ISO needs to be 0 otherwise analog and digital gain won't work.
camera.iso = 0
# set the camea settings
camera.framerate = cfg.getfloat(camera_type, "framerate")
camera.awb_gains = (cfg.getfloat(camera_type, "awb_gain_red"), cfg.getfloat(camera_type, "awb_gain_blue"))
camera.analog_gain = cfg.getfloat(camera_type, "analog_gain")
camera.digital_gain = cfg.getfloat(camera_type, "digital_gain")
camera.shutter_speed = cfg.getint(camera_type, "exposure")
rawCapture = PiRGBArray(camera, size=(nx, ny))
# allow the camera to warmup
time.sleep(0.1)
try:
# Loop until reaching end time
while float(time.time()) < tend:
# Wait for available capture buffer to become available
if (image_queue.qsize() > 1):
logger.warning("Acquiring data faster than your CPU can process")
slow_CPU = True
while (image_queue.qsize() > 1):
time.sleep(0.1)
if slow_CPU:
lost_video = time.time() - t
logger.info("Waited %.3fs for available capture buffer" % lost_video)
slow_CPU = False
# Get frames
i = 0
for frameA in camera.capture_continuous(rawCapture, format="bgr", use_video_port=True):
# Store start time
t0 = float(time.time())
# grab the raw NumPy array representing the image, then initialize the timestamp
frame = frameA.array
# Compute mid time
t = (float(time.time()) + t0) / 2
# Skip lost frames
if frame is not None:
# Convert image to grayscale
z = np.asarray(cv2.cvtColor(
frame, cv2.COLOR_BGR2GRAY)).astype(np.uint8)
# optionally rotate the frame by 2 * 90 degrees.
# z = np.rot90(z, 2)
# Display Frame
if live is True:
cv2.imshow("Capture", z)
cv2.waitKey(1)
# Store results
if first:
z1[:, :, i] = z
t1[i] = t
else:
z2[:, :, i] = z
t2[i] = t
# clear the stream in preparation for the next frame
rawCapture.truncate(0)
# count up to nz frames, then break out of the for loop.
i += 1
if i >= nz:
break
if first:
buf = 1
else:
buf = 2
image_queue.put(buf)
logger.debug("Captured buffer %d" % buf)
# Swap flag
first = not first
reason = "Session complete"
except KeyboardInterrupt:
print()
reason = "Keyboard interrupt"
except ValueError as e:
logger.error("%s" % e)
reason = "Wrong image dimensions? Fix nx, ny in config."
finally:
# End capture
logger.info("Capture: %s - Exiting" % reason)
camera.close()
# Capture images from cv2
def capture_cv2(image_queue, z1base, t1base, z2base, t2base, nx, ny, nz, tend, device_id, live, conf_file):
global logger
logger = setup_logging(os.getcwd())
cfg = configparser.ConfigParser(inline_comment_prefixes=("#", ";"))
cfg.read(conf_file)
z1 = np.ctypeslib.as_array(z1base.get_obj()).reshape(ny, nx, nz)
t1 = np.ctypeslib.as_array(t1base.get_obj())
z2 = np.ctypeslib.as_array(z2base.get_obj()).reshape(ny, nx, nz)
t2 = np.ctypeslib.as_array(t2base.get_obj())
# Intialization
camera_type = "CV2"
first = True
slow_CPU = False
# Initialize cv2 device
if cfg.has_option(camera_type, "device_string"):
device = cv2.VideoCapture(cfg.get(camera_type, "device_string"))
else:
device = cv2.VideoCapture(device_id)
# Test for software binning
try:
software_bin = cfg.getint(camera_type, "software_bin")
except configparser.Error:
software_bin = 1
# Test for software cropping
try:
software_crop_width = cfg.getint(camera_type, "software_crop_width")
except configparser.Error:
software_crop_width = 0
# Test for software cropping
try:
software_crop_height = cfg.getint(camera_type, "software_crop_height")
except configparser.Error:
software_crop_height = 0
# Set properties
device.set(3, nx * software_bin)
device.set(4, ny * software_bin)
try:
# Loop until reaching end time
while float(time.time()) < tend:
# Wait for available capture buffer to become available
if (image_queue.qsize() > 1):
logger.warning("Acquiring data faster than your CPU can process")
slow_CPU = True
while (image_queue.qsize() > 1):
time.sleep(0.1)
if slow_CPU:
lost_video = time.time() - t
logger.info("Waited %.3fs for available capture buffer" % lost_video)
slow_CPU = False
# Get frames
for i in range(nz):
# Store start time
t0 = float(time.time())
# Get frame
res, frame = device.read()
# Compute mid time
t = (float(time.time()) + t0) / 2
# Skip lost frames
if res is True:
# Convert image to grayscale
z = np.asarray(cv2.cvtColor(
frame, cv2.COLOR_BGR2GRAY)).astype(np.uint8)
# Apply software binning
if software_bin > 1:
my, mx = z.shape
z = cv2.resize(z, (mx // software_bin, my // software_bin))
# Apply software cropping
if software_crop_width > 1:
my, mx = z.shape
# Crop n pixels
crop_w = mx - software_crop_width
crop_h = my - software_crop_height
y_start = (my - crop_h) // 2
x_start = (mx - crop_w) // 2
y_end = y_start + crop_h
x_end = x_start + crop_w
z = z[y_start:y_end, x_start:x_end]
# Display Frame
if live is True:
cv2.imshow("Capture", z)
cv2.waitKey(1)
# Store results
if first:
z1[:, :, i] = z
t1[i] = t
else:
z2[:, :, i] = z
t2[i] = t
if first:
buf = 1
else:
buf = 2
image_queue.put(buf)
logger.debug("Captured z%d" % buf)
# Swap flag
first = not first
reason = "Session complete"
except KeyboardInterrupt:
print()
reason = "Keyboard interrupt"
except ValueError as e:
logger.error("%s" % e)
reason = "Wrong image dimensions? Fix nx, ny in config."
finally:
# End capture
logger.info("Capture: %s - Exiting" % reason)
device.release()
# Capture images
def capture_asi(image_queue, z1base, t1base, z2base, t2base, nx, ny, nz, tend, device_id, live, conf_file):
global logger
logger = setup_logging(os.getcwd())
cfg = configparser.ConfigParser(inline_comment_prefixes=("#", ";"))
cfg.read(conf_file)
import zwoasi as asi
z1 = np.ctypeslib.as_array(z1base.get_obj()).reshape(ny, nx, nz)
t1 = np.ctypeslib.as_array(t1base.get_obj())
z2 = np.ctypeslib.as_array(z2base.get_obj()).reshape(ny, nx, nz)
t2 = np.ctypeslib.as_array(t2base.get_obj())
first = True # Array flag
slow_CPU = False # Performance issue flag
camera_type = "ASI"
gain = cfg.getint(camera_type, "gain")
maxgain = cfg.getint(camera_type, "maxgain")
autogain = cfg.getboolean(camera_type, "autogain")
exposure = cfg.getint(camera_type, "exposure")
binning = cfg.getint(camera_type, "bin")
brightness = cfg.getint(camera_type, "brightness")
bandwidth = cfg.getint(camera_type, "bandwidth")
high_speed = cfg.getint(camera_type, "high_speed")
hardware_bin = cfg.getint(camera_type, "hardware_bin")
sdk = cfg.get(camera_type, "sdk")
try:
software_bin = cfg.getint(camera_type, "software_bin")
except configparser.Error:
software_bin = 0
# Initialize device
asi.init(sdk)
num_cameras = asi.get_num_cameras()
if num_cameras == 0:
logger.error("No ZWOASI cameras found")
raise ValueError
sys.exit()
cameras_found = asi.list_cameras() # Models names of the connected cameras
if num_cameras == 1:
device_id = 0
logger.info("Found one camera: %s" % cameras_found[0])
else:
logger.info("Found %d ZWOASI cameras" % num_cameras)
for n in range(num_cameras):
logger.info(" %d: %s" % (n, cameras_found[n]))
logger.info("Using #%d: %s" % (device_id, cameras_found[device_id]))
camera = asi.Camera(device_id)
camera_info = camera.get_camera_property()
logger.debug("ASI Camera info:")
for (key, value) in camera_info.items():
logger.debug(" %s : %s" % (key,value))
camera.set_control_value(asi.ASI_BANDWIDTHOVERLOAD, bandwidth)
camera.disable_dark_subtract()
camera.set_control_value(asi.ASI_GAIN, gain, auto=autogain)
camera.set_control_value(asi.ASI_EXPOSURE, exposure, auto=False)
camera.set_control_value(asi.ASI_AUTO_MAX_GAIN, maxgain)
camera.set_control_value(asi.ASI_AUTO_MAX_BRIGHTNESS, 20)
camera.set_control_value(asi.ASI_WB_B, 99)
camera.set_control_value(asi.ASI_WB_R, 75)
camera.set_control_value(asi.ASI_GAMMA, 50)
camera.set_control_value(asi.ASI_BRIGHTNESS, brightness)
camera.set_control_value(asi.ASI_FLIP, 0)
try:
camera.set_control_value(asi.ASI_HIGH_SPEED_MODE, high_speed)
except:
pass
try:
camera.set_control_value(asi.ASI_HARDWARE_BIN, hardware_bin)
except:
pass
camera.set_roi(bins=binning)
camera.start_video_capture()
camera.set_image_type(asi.ASI_IMG_RAW8)
try:
# Fix autogain
if autogain:
while True:
# Get frame
z = camera.capture_video_frame()
# Break on no change in gain
settings = camera.get_control_values()
if gain == settings["Gain"]:
break
gain = settings["Gain"]
camera.set_control_value(asi.ASI_GAIN, gain, auto=autogain)
# Loop until reaching end time
while float(time.time()) < tend:
# Wait for available capture buffer to become available
if (image_queue.qsize() > 1):
logger.warning("Acquiring data faster than your CPU can process")
slow_CPU = True
while (image_queue.qsize() > 1):
time.sleep(0.1)
if slow_CPU:
lost_video = time.time() - t
logger.info("Waited %.3fs for available capture buffer" % lost_video)
slow_CPU = False
# Get settings
try:
settings = camera.get_control_values()
gain = settings["Gain"]
temp = settings["Temperature"] / 10
except:
gain, temp = 0, 0
logger.info("Capturing frame with gain %d, temperature %.1f" % (gain, temp))
# Set gain
if autogain:
camera.set_control_value(asi.ASI_GAIN, gain, auto=autogain)
# Get frames
for i in range(nz):
# Store start time
t0 = float(time.time())
# Get frame
z = camera.capture_video_frame()
# Apply software binning
if software_bin > 1:
my, mx = z.shape
z = cv2.resize(z, (mx // software_bin, my // software_bin))
# Compute mid time
t = (float(time.time()) + t0) / 2
# Display Frame
if live is True:
cv2.imshow("Capture", z)
cv2.waitKey(1)
# Store results
if first:
z1[:, :, i] = z
t1[i] = t
else:
z2[:, :, i] = z
t2[i] = t
if first:
buf = 1
else:
buf = 2
image_queue.put(buf)
logger.debug("Captured buffer %d (%dx%dx%d)" % (buf, nx, ny, nz))
# Swap flag
first = not first
reason = "Session complete"
except KeyboardInterrupt:
print()
reason = "Keyboard interrupt"
except ValueError as e:
logger.error("%s" % e)
reason = "Wrong image dimensions? Fix nx, ny in config."
except MemoryError as e:
logger.error("Capture: Memory error %s" % e)
finally:
# End capture
logger.info("Capture: %s - Exiting" % reason)
camera.stop_video_capture()
camera.close()
def compress(image_queue, z1base, t1base, z2base, t2base, nx, ny, nz, tend, path, device_id, conf_file):
""" compress: Aggregate nframes of observations into a single FITS file, with statistics.
ImageHDU[0]: mean pixel value nframes (zmax)
ImageHDU[1]: standard deviation of nframes (zstd)
ImageHDU[2]: maximum pixel value of nframes (zmax)
ImageHDU[3]: maximum pixel value frame number (znum)
Also updates a [observations_path]/control/state.txt for interfacing with satttools/runsched and sattools/slewto
"""
global logger
logger = setup_logging(os.getcwd())
cfg = configparser.ConfigParser(inline_comment_prefixes=("#", ";"))
cfg.read(conf_file)
z1 = np.ctypeslib.as_array(z1base.get_obj()).reshape(ny, nx, nz)
t1 = np.ctypeslib.as_array(t1base.get_obj())
z2 = np.ctypeslib.as_array(z2base.get_obj()).reshape(ny, nx, nz)
t2 = np.ctypeslib.as_array(t2base.get_obj())
# Force a restart
controlpath = os.path.join(path, "control")
if not os.path.exists(controlpath):
try:
os.makedirs(controlpath)
except PermissionError:
logger.error("Can not create control path directory: %s" % controlpath)
raise
if not os.path.exists(os.path.join(controlpath, "position.txt")):
with open(os.path.join(controlpath, "position.txt"), "w") as fp:
fp.write("\n")
with open(os.path.join(controlpath, "state.txt"), "w") as fp:
fp.write("restart\n")
try:
# Start processing
while True:
# Check mount state
restart = False
with open(os.path.join(controlpath, "state.txt"), "r") as fp:
line = fp.readline().rstrip()
if line == "restart":
restart = True
# Restart
if restart:
# Log state
with open(os.path.join(controlpath, "state.txt"), "w") as fp:
fp.write("observing\n")
# Get obsid
trestart = time.gmtime()
obsid = "%s_%d/%s" % (time.strftime("%Y%m%d", trestart), device_id, time.strftime("%H%M%S", trestart))
filepath = os.path.join(path, obsid)
logger.info("Storing files in %s" % filepath)
# Create output directory
if not os.path.exists(filepath):
try:
os.makedirs(filepath)
except PermissionError:
logger.error("Can not create output directory: %s" % filepath)
raise
# Get mount position
with open(os.path.join(controlpath, "position.txt"), "r") as fp:
line = fp.readline()
with open(os.path.join(filepath, "position.txt"), "w") as fp:
fp.write(line)
# Wait for completed capture buffer to become available
while image_queue.empty():
time.sleep(0.1)
# Get next buffer # from the work queue
try:
proc_buffer = image_queue.get(timeout=60)
except:
logger.debug("Queue timed out")
break
logger.debug("Processing buffer %d" % proc_buffer)
# Log start time
tstart = time.time()
# Process first buffer
if proc_buffer == 1:
t = t1
z = z1
elif proc_buffer == 2:
t = t2
z = z2
# Format time
nfd = "%s.%03d" % (time.strftime("%Y-%m-%dT%T",
time.gmtime(t[0])), int((t[0] - np.floor(t[0])) * 1000))
t0 = Time(nfd, format="isot")
dt = t - t[0]
# Cast to 32 bit float
z = z.astype("float32")
# Compute statistics
zmax = np.max(z, axis=2)
znum = np.argmax(z, axis=2)
zs1 = np.sum(z, axis=2) - zmax
zs2 = np.sum(z * z, axis=2) - zmax * zmax
zavg = zs1 / float(nz - 1)
zstd = np.sqrt((zs2 - zs1 * zavg) / float(nz - 2))
# Convert to float and flip
zmax = np.flipud(zmax.astype("float32"))
znum = np.flipud(znum.astype("float32"))
zavg = np.flipud(zavg.astype("float32"))
zstd = np.flipud(zstd.astype("float32"))
# Generate fits
ftemp = "%s.temp" % nfd.replace(":", "-")
fname = "%s.fits" % nfd.replace(":", "-")
# Format header
hdr = fits.Header()
hdr["DATE-OBS"] = "%s" % nfd
hdr["MJD-OBS"] = t0.mjd
hdr["EXPTIME"] = dt[-1] - dt[0]
hdr["NFRAMES"] = nz
hdr["CRPIX1"] = float(nx) / 2
hdr["CRPIX2"] = float(ny) / 2
hdr["CRVAL1"] = 0.0
hdr["CRVAL2"] = 0.0
hdr["CD1_1"] = 1 / 3600
hdr["CD1_2"] = 0.0
hdr["CD2_1"] = 0.0
hdr["CD2_2"] = 1 / 3600
hdr["CTYPE1"] = "RA---TAN"
hdr["CTYPE2"] = "DEC--TAN"
hdr["CUNIT1"] = "deg"
hdr["CUNIT2"] = "deg"
hdr["CRRES1"] = 0.0
hdr["CRRES2"] = 0.0
hdr["EQUINOX"] = 2000.0
hdr["RADECSYS"] = "ICRS"
hdr["COSPAR"] = cfg.getint("Observer", "cospar")
hdr["OBSERVER"] = cfg.get("Observer", "name")
hdr["SITELONG"] = cfg.getfloat("Observer", "longitude")
hdr["SITELAT"] = cfg.getfloat("Observer", "latitude")
hdr["ELEVATIO"] = cfg.getfloat("Observer", "height")
if cfg.getboolean("Setup", "tracking_mount"):
hdr["TRACKED"] = 1
else:
hdr["TRACKED"] = 0
for i in range(nz):
hdr["DT%04d" % i] = dt[i]
for i in range(10):
hdr["DUMY%03d" % i] = 0.0
# Write fits file
hdu = fits.PrimaryHDU(data=np.array([zavg, zstd, zmax, znum]),
header=hdr)
hdu.writeto(os.path.join(filepath, ftemp))
os.rename(os.path.join(filepath, ftemp), os.path.join(filepath, fname))
logger.info("Compressed %s in %.2f sec" % (fname, time.time() - tstart))
# Exit on end of capture
if t[-1] > tend:
break
logger.debug("Processed buffer %d" % proc_buffer)
except KeyboardInterrupt:
pass
except MemoryError as e:
logger.error("Compress: Memory error %s" % e)
finally:
# Exiting
logger.info("Exiting compress")
# Main function
if __name__ == '__main__':
multiprocessing.set_start_method("spawn", force=True)
# Read commandline options
conf_parser = argparse.ArgumentParser(description="Capture and compress" +
" live video frames.")
conf_parser.add_argument("-c", "--conf_file",
help="Specify configuration file(s). If no file" +
" is specified 'configuration.ini' is used.",
action="append",
nargs="?",
metavar="FILE")
conf_parser.add_argument("-t", "--test",
nargs="?",
action="store",
default=False,
help="Testing mode - Start capturing immediately for (optional) seconds",
metavar="s")
conf_parser.add_argument("-l", "--live", action="store_true",
help="Display live image while capturing")
args = conf_parser.parse_args()
# Process commandline options and parse configuration
cfg = configparser.ConfigParser(inline_comment_prefixes=("#", ";"))
conf_file = args.conf_file if args.conf_file else "configuration.ini"
result = cfg.read(conf_file)
if not result:
print("Could not read config file: %s\nExiting..." % conf_file)
sys.exit()
# Setup logging
logFormatter = logging.Formatter("%(asctime)s [%(threadName)-12.12s] " +
"[%(levelname)-5.5s] %(message)s")
logger = logging.getLogger()
# Generate directory
path = os.path.abspath(cfg.get("Setup", "observations_path"))
if not os.path.exists(path):
try:
os.makedirs(path)
except PermissionError:
logger.error("Can not create observations_path: %s" % path)
sys.exit()
fileHandler = logging.FileHandler(os.path.join(path, "acquire.log"))
fileHandler.setFormatter(logFormatter)
logger.addHandler(fileHandler)
consoleHandler = logging.StreamHandler(sys.stdout)
consoleHandler.setFormatter(logFormatter)
logger.addHandler(consoleHandler)
logger.setLevel(logging.DEBUG)
logger.info("Using config: %s" % conf_file)
# Testing mode
if args.test is None:
test_duration = 31
testing = True
elif args.test is not False:
test_duration = int(args.test)
testing = True
else:
testing = False
logger.info("Test mode: %s" % testing)
if (testing):
logger.info("Test duration: %ds" % test_duration)
# Live mode
live = True if args.live else False
logger.info("Live mode: %s" % live)
# Get camera type
camera_type = cfg.get("Setup", "camera_type")
# Get device id
device_id = cfg.getint(camera_type, "device_id")
# Current time
tnow = Time.now()
# Set location
loc = EarthLocation(lat=cfg.getfloat("Observer", "latitude") * u.deg,
lon=cfg.getfloat("Observer", "longitude") * u.deg,
height=cfg.getfloat("Observer", "height") * u.m)
if not testing:
# Reference altitudes
refalt_set = cfg.getfloat("Setup", "alt_sunset") * u.deg
refalt_rise = cfg.getfloat("Setup", "alt_sunrise") * u.deg
# Aimpoint configuration
if cfg.has_section("Aimpoint"):
aimpoint_az = cfg.getfloat("Aimpoint", "az_deg") * u.deg
aimpoint_alt = cfg.getfloat("Aimpoint", "alt_deg") * u.deg
aimpoint_height = cfg.getfloat("Aimpoint", "height_km") * u.km
else:
aimpoint_az, aimpoint_alt, aimpoint_height = None, None, None
# Get logic
action, wait_time, tend, state = observe_logic(tnow, loc, refalt_set, refalt_rise,
aimpoint_az, aimpoint_alt, aimpoint_height)
# Wait for observation start
logger.info(state)
if action == "wait":
logger.info(f"Waiting for {wait_time:.0f} seconds.")
try:
time.sleep(wait_time)
except KeyboardInterrupt:
sys.exit()
else:
tend = tnow + test_duration * u.s
# Read shutter config
if cfg.has_section("Shutter"):
from stvid.shutter import Shutter
shutter = Shutter(cfg.getint("Shutter", "pin"))
else:
shutter = None
logger.info("Starting data acquisition")
logger.info("Acquisition will end after "+tend.isot)
# Get settings
nx = cfg.getint(camera_type, "nx")
ny = cfg.getint(camera_type, "ny")
nz = cfg.getint(camera_type, "nframes")
# Initialize arrays
z1base = multiprocessing.Array(ctypes.c_uint8, nx * ny * nz)
t1base = multiprocessing.Array(ctypes.c_double, nz)
z2base = multiprocessing.Array(ctypes.c_uint8, nx * ny * nz)
t2base = multiprocessing.Array(ctypes.c_double, nz)
image_queue = multiprocessing.Queue()
# Set processes
pcompress = multiprocessing.Process(target=compress,
name="compress",
args=(image_queue,
z1base, t1base, z2base, t2base,
nx, ny, nz, tend.unix,
path, device_id, conf_file))
if camera_type == "PI":
pcapture = multiprocessing.Process(target=capture_pi,
name="capture_pi",
args=(image_queue,
z1base, t1base, z2base, t2base,
nx, ny, nz, tend.unix,
device_id, live, conf_file))
elif camera_type == "CV2":
pcapture = multiprocessing.Process(target=capture_cv2,
name="capture_cv2",
args=(image_queue,
z1base, t1base, z2base, t2base,
nx, ny, nz, tend.unix,
device_id, live, conf_file))
elif camera_type == "ASI":
pcapture = multiprocessing.Process(target=capture_asi,
name="capture_asi",
args=(image_queue,
z1base, t1base, z2base, t2base,
nx, ny, nz, tend.unix,
device_id, live, conf_file))
try:
# Open shutter
if shutter:
shutter.open()
# Start
pcapture.start()
pcompress.start()
# End
try:
pcapture.join()
pcompress.join()
except (KeyboardInterrupt, ValueError):
time.sleep(0.1) # Allow a little time for a graceful exit
except MemoryError as e:
logger.error("Memory error %s" % e)
finally:
pcapture.terminate()
pcompress.terminate()
# Release device
if live is True:
cv2.destroyAllWindows()
finally:
if shutter:
shutter.close()
73!
Btw do you think a 3B would suffice for this task? I’m really interested to revive this project after failed to install/setup the software a while ago. The raspberry 4B that meant for this is now runs the allsky camera.
Camera that can be used either Raspberry HQ camera or ZWO 178MM (which is more preferable ofc).
maybe it can run, but not sure if it can handle the time delayed when process the n frames. if it can handle, it should work. but on my raspberrypi 4 with 2 gb ram i should reduce max the nframe and size of pixel. ideally bigger ram, better
Right lets see, or maybe its better to run on Linux PC?
Found a cheap f 1.4 lens on the marketplace and I think it should be perfect for this.
of course. pc is fastest and the best.. more core and more ram, to run processing.py
No, the 3B would not suffice. I am running the Pi5 with (only) 2GB and the Pi HQ camera, I get the memory up to 2 percent free, and processor(s) over 110 percent while running process.py
Added to my topic about the Pi5 setup
thank you ben! for everyone should read @pe2bz thread first!! bookmark recommended!
Roger, I’ll try to run it on Linux PC instead with the camera sticking outside to the South.
I think that’s a better idea, the Linux PC, pointing to the south however also results in a lot of sunshine / heat during daytime !
tips for targeting a specific satelit for optical obs:
first, you should know azimuth and elevation of your current camera setup.
the azimuth and elevation of your current cam, can you find when you open skymap with random sucess iod file (.dat) file from last observation, for example:
skymap -d 2026-07-15T12-36-20.282_66919_catalog.dat -c catalog.tle -i 66919
look at below image, you will find A for azimuth, and E for elevation
if you got the A and E, then
run this command:
skymap -c satno.tle -i 40931 -A 300 -E 50
satno.tle = tle catalog file
40931 = norad id targeted sat
300 = azimuth of your cam
50 = elevation of your cam
then use +/- for zoom
, and . for back/forward time
when the sat trail pass the square, its good. if the satelit not pass inside square, try change the azimuth or altitude. if you feel good with the azimuth and altitude, then set your camera position with this new setting.
make sure the trail color is yellow, that mean the sat exposed by sun light.
Thanks for this instructions Bali!
I tried to give this a go on a detected satellite from last night.
skymap -d 2026-07-15T20-57-06.713_60210_catalog.dat -c ~/software/tle/catalog.tle -i 60210
60210 24 125H 9926 G 20260715205707859 17 25 1325245+812984 37 S
2026-07-15T20:57:07 60210 4.9 0.7 d 427.97 km 111.67 deg
which shows
so, my Azimuth is 347.7 and my elevation is 56.9 , would you know if that’s the elevation in center of the image ?
This image shows the yellow trail, I zoomed in a bit.
After that I ran
skymap -c ~/software/tle/catalog.tle -i 60210 -A 348 -E 56.9
showing no trace / yellow line. Any idea why not ? EDit:
I think I understand, the last plot shows the current time so you can browse for upcoming FOV passes.
Thanks,
Ben
look at the time. default without set the time, the skymap will show current time. you can set time with option -t
t date/time (yyyy-mm-ddThh:mm:ss.sss) [default: now]
or just use button , and . for back forward time
note: you can see the color of sky is different. darker mean night, lighter mean day
i compare the .png file output from processing.py. and skymap and i found it maybe different around 5 degree on azimuth on my wide view ip cam
also something the camera position in reality maybe there is a little tilt that make little different angle if see the result of png or skymap that flat
before:
skymap -d 2026-07-15T21-17-39.492_01314_catalog.dat -c ~/tle/catalog.tle -i 01314 -t 2026-07-15T21:17:39.492
after azimuth adjustment:
skymap -d 2026-07-15T21-17-39.492_01314_catalog.dat -c ~/tle/catalog.tle -i 01314 -t 2026-07-15T21:17:39.492 -A 353.6 -E 50.3
and here i adjust with gimp, little tilt to right
i notice there is file cameras.txt on sattools/data folder. edit this and put your cam fov at first line
# List of available cameras field of view
# Camera mnemonic (Camera/Optics), field width (deg), field height (deg)
TAPO 65.19 48.82
1600_85 11.83 8.67
1600_85 8.67 11.83
174_55 11.57 7.27
1600_55 18.3 13.9
W50H 7.21 5.31
D85V 9.97 14.96
FULL 360 240
5x5 5.4 5.4