Commit 33e95c8d authored by markus's avatar markus

Initial Commit

parents
Pipeline #35 failed with stages
__pycache__/
twitter_config.py
*.swp
[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true
[dev-packages]
[packages]
tweepy = "*"
[requires]
python_version = "3.7"
{
"_meta": {
"hash": {
"sha256": "7a655092435ce768e5f48b1a8c05366412580e30771c6c623bc638cc94748b47"
},
"pipfile-spec": 6,
"requires": {
"python_version": "3.7"
},
"sources": [
{
"name": "pypi",
"url": "https://pypi.org/simple",
"verify_ssl": true
}
]
},
"default": {
"certifi": {
"hashes": [
"sha256:046832c04d4e752f37383b628bc601a7ea7211496b4638f6514d0e5b9acc4939",
"sha256:945e3ba63a0b9f577b1395204e13c3a231f9bc0223888be653286534e5873695"
],
"version": "==2019.6.16"
},
"chardet": {
"hashes": [
"sha256:84ab92ed1c4d4f16916e05906b6b75a6c0fb5db821cc65e70cbd64a3e2a5eaae",
"sha256:fc323ffcaeaed0e0a02bf4d117757b98aed530d9ed4531e3e15460124c106691"
],
"version": "==3.0.4"
},
"idna": {
"hashes": [
"sha256:c357b3f628cf53ae2c4c05627ecc484553142ca23264e593d327bcde5e9c3407",
"sha256:ea8b7f6188e6fa117537c3df7da9fc686d485087abf6ac197f9c46432f7e4a3c"
],
"version": "==2.8"
},
"oauthlib": {
"hashes": [
"sha256:bee41cc35fcca6e988463cacc3bcb8a96224f470ca547e697b604cc697b2f889",
"sha256:df884cd6cbe20e32633f1db1072e9356f53638e4361bef4e8b03c9127c9328ea"
],
"version": "==3.1.0"
},
"pysocks": {
"hashes": [
"sha256:15d38914b60dbcb231d276f64882a20435c049450160e953ca7d313d1405f16f",
"sha256:32238918ac0f19e9fd870a8692ac9bd14f5e8752b3c62624cda5851424642210",
"sha256:d9031ea45fdfacbe59a99273e9f0448ddb33c1580fe3831c1b09557c5718977c"
],
"version": "==1.7.0"
},
"requests": {
"hashes": [
"sha256:11e007a8a2aa0323f5a921e9e6a2d7e4e67d9877e85773fba9ba6419025cbeb4",
"sha256:9cf5292fcd0f598c671cfc1e0d7d1a7f13bb8085e9a590f48c010551dc6c4b31"
],
"version": "==2.22.0"
},
"requests-oauthlib": {
"hashes": [
"sha256:bd6533330e8748e94bf0b214775fed487d309b8b8fe823dc45641ebcd9a32f57",
"sha256:d3ed0c8f2e3bbc6b344fa63d6f933745ab394469da38db16bdddb461c7e25140"
],
"version": "==1.2.0"
},
"six": {
"hashes": [
"sha256:3350809f0555b11f552448330d0b52d5f24c91a322ea4a15ef22629740f3761c",
"sha256:d16a0141ec1a18405cd4ce8b4613101da75da0e9a7aec5bdd4fa804d0e0eba73"
],
"version": "==1.12.0"
},
"tweepy": {
"hashes": [
"sha256:8abd828ba51a85a2b5bb7373715d6d3bb32d18ac624e3a4db02e4ef8ab48316b",
"sha256:ecc7f200c86127903017e48824efd008734814e95f3e8e9b45ce0f4120dd08db"
],
"index": "pypi",
"version": "==3.8.0"
},
"urllib3": {
"hashes": [
"sha256:b246607a25ac80bedac05c6f282e3cdaf3afb65420fd024ac94435cabe6e18d1",
"sha256:dbe59173209418ae49d485b87d1681aefa36252ee85884c31346debd19463232"
],
"version": "==1.25.3"
}
},
"develop": {}
}
# R0tor image collection and processing
Work in progress of retrieving, processing and publishing observation data.
#!/usr/bin/env python3
from os import environ
from sys import exit
from time import sleep
import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
import urllib.request
import satnogs_query
# FIXME decouple logic
import twitter
twitter_client = twitter.TwitterClient()
def process_observation(observation, observation_data):
waterfall_image = download_image(observation_data["waterfall_url"])
data_images = [download_image(x) for x in observation_data["data_image_urls"]]
twitter_client.handle_observation(observation, observation_data, waterfall_image, data_images)
def download_image(url):
logger.info(f"Downloading: {url}")
with urllib.request.urlopen(url) as req:
data = req.read()
logger.info("Download finished")
return data
def loop(satnogs_address, station, satellites, observer):
processed_observations = []
start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
try:
while True:
for satellite in satellites:
logger.info(f"Querying for satellite {satellite}")
for observation in satnogs_query.get_past_station_observations(satnogs_address, station, satellite, observer, start_time = start_time):
if observation in processed_observations:
continue
observation_data = satnogs_query.get_observation_data(satnogs_address, observation)
process_observation(observation, observation_data)
processed_observations.append(observation)
sleep(30)
except KeyboardInterrupt:
pass
if __name__ == "__main__":
station = environ.get("SATNOGS_STATION", None)
satellites_plain = environ.get("SATNOGS_SATELLITES", None)
observer = environ.get("SATNOGS_OBSERVER", None)
satnogs_address = environ.get("SATNOGS_ADDRESS", "https://network.satnogs.org")
if not station or not satellites_plain:
print("Sation and satellites to follow must be given")
exit(1)
satellites = [x.strip() for x in satellites_plain.split(" ") if x.strip()]
loop(satnogs_address, station, satellites, observer)
with import <nixpkgs> {}; {
env = stdenv.mkDerivation {
name = "env";
buildInputs = [
pipenv
];
};
}
#!/usr/bin/env python3
from os import environ
from sys import exit
import logging
logger = logging.getLogger(__name__)
import re
import urllib.parse
import urllib.request
def get_past_station_observations(satnogs_address, station, satellite, observer, start_time=None):
data = {
"future": 0,
"bad": 0,
"unvetted": 0,
"failed": 0,
"norad": satellite,
"observer": observer,
"station": station,
"results": "d1", # only those with data
}
if start_time:
data["start-time"] = start_time
values = urllib.parse.urlencode(data)
url = satnogs_address + "/observations/?" + values
logger.info(f"Querying {url}")
with urllib.request.urlopen(url) as response:
page = str(response.read())
observation_ids = [m for m in re.findall(r"data-href=\"/observations/(\d+)/\"", page)]
logger.info(f"Matched the following observations: {','.join(observation_ids)}")
return observation_ids
def get_observation_data(satnogs_address, observation):
url = f"{satnogs_address}/observations/{observation}/"
with urllib.request.urlopen(url) as response:
page = str(response.read())
waterfall_image_url = re.search(f"/media/data_obs/{observation}/waterfall_{observation}_\d+-\d+-\d+T\d+-\d+-\d+.png", page).group(0)
data_image_urls = re.findall(f"src=\"(/media/data_obs/{observation}/data_{observation}_\d+-\d+-\d+T\d+-\d+-\d+.png)", page)
satellite_id = re.search("<a href=\"#\" data-toggle=\"modal\" data-target=\"#SatelliteModal\" data-id=\"(\d+)\">", page).group(1)
satellite_name = re.search(f"\s+{satellite_id}\s+-\s+(.*?)</a>", page).group(1).strip()[:-2] # remove magic \n
data_timeframe_start = re.search(f"data-timeframe-start=\"(.*?)\"", page).group(1).strip()
data_timeframe_end = re.search(f"data-timeframe-end=\"(.*?)\"", page).group(1).strip()
return {
"waterfall_url": f"{satnogs_address}{str(waterfall_image_url)}",
"data_image_urls": [f"{satnogs_address}{data_image_url}" for data_image_url in data_image_urls],
"satellite_id": satellite_id,
"satellite_name": satellite_name,
"data_timeframe_start": data_timeframe_start,
"data_timeframe_end": data_timeframe_end,
"url": url,
"id": observation,
}
if __name__ == "__main__":
station = environ.get("SATNOGS_STATION", None)
satellites_plain = environ.get("SATNOGS_SATELLITES", None)
observer = environ.get("SATNOGS_OBSERVER", None)
satnogs_address = environ.get("SATNOGS_ADDRESS", "https://network.satnogs.org")
if not station or not satellites_plain:
print("Sation and satellites to follow must be given")
exit(1)
satellites = [x.strip() for x in satellites_plain.split(" ") if x.strip()]
logger.setLevel(20)
for satellite in satellites:
for observation in get_past_station_observations(satnogs_address, station, satellite, observer):
print(get_image_urls_from_observation(satnogs_address, observation))
#!/usr/bin/env python3
import tempfile
import time
import tweepy
import logging
logger = logging.getLogger(__name__)
from twitter_config import *
class TwitterClient:
def __init__(self):
global consumer_key, consumer_secret, access_token, access_token_secret
self.auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
self.auth.set_access_token(access_token, access_token_secret)
self.api = tweepy.API(self.auth)
logger.info(f"Logged in to twitter as '{self.api.me().name}")
def handle_observation(self, observation, observation_data, waterfall_image, data_images, tweet_waterfall = False):
with tempfile.TemporaryFile() as tmp:
tmp.write(data_images[0])
text = f"New satellite image from {observation_data['satellite_name']} at {observation_data['data_timeframe_start']} UTC. {observation_data['url']}"
try:
status = self.api.update_with_media("image.png", text, file=tmp)
except tweepy.error.TweepError as e:
logger.warning(f"Couldn't tweet observation {observation}: {e}")
return
logger.info(f"Tweeted: '{text}'")
if tweet_waterfall:
time.sleep(0.1)
with tempfile.TemporaryFile() as tmp:
tmp.write(waterfall_image)
waterfall_status = self.api.update_with_media("image.png", in_reply_to_status_id=status.id, file=tmp)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment